nomadcode/services/core/internal/gitoevents/client.go

package gitoevents

import (
	"context"
	"errors"
	"log/slog"

	"github.com/nomadcode/nomadcode-core/internal/protosocket"
)

// Transport abstracts the proto-socket connection to Gito so the client can be
// driven by a fake transport in tests. Production transports wrap the existing
// websocket / proto-socket dependencies; no new dependency is introduced here.
type Transport interface {
	// Connect establishes the underlying connection.
	Connect(ctx context.Context) error
	// Write sends an envelope to Gito (e.g. the subscribe request).
	Write(ctx context.Context, env protosocket.Envelope) error
	// Read blocks for the next inbound envelope. It returns an error wrapping
	// io.EOF / context cancellation when the stream ends.
	Read(ctx context.Context) (protosocket.Envelope, error)
	// Close releases the connection.
	Close() error
}

// Handler receives decoded, on-target branch events.
type Handler func(ctx context.Context, event BranchUpdatedEvent)

// Client subscribes to Gito branch events over a Transport and forwards
// matching `branch.updated` events to a Handler. Wrong channel/action, wrong
// repo/branch, and malformed payloads are dropped through an observable log
// path instead of reaching the handler.
type Client struct {
	transport Transport
	repoID    string
	branch    string
	handler   Handler
	logger    *slog.Logger
	idFunc    func() string
}

// Options configures a Client.
type Options struct {
	RepoID  string
	Branch  string
	Handler Handler
	Logger  *slog.Logger
	// IDFunc generates envelope ids for outbound requests. Defaults to a
	// fixed seed if nil; production callers should provide a unique generator.
	IDFunc func() string
}

// NewClient builds a Client. RepoID must be non-empty; an empty branch falls
// back to DefaultBranch.
func NewClient(transport Transport, opts Options) (*Client, error) {
	if transport == nil {
		return nil, errors.New("gitoevents: transport is nil")
	}
	if opts.RepoID == "" {
		return nil, errors.New("gitoevents: repo_id is required")
	}
	branch := opts.Branch
	if branch == "" {
		branch = DefaultBranch
	}
	idFunc := opts.IDFunc
	if idFunc == nil {
		idFunc = func() string { return "gito-subscribe" }
	}
	return &Client{
		transport: transport,
		repoID:    opts.RepoID,
		branch:    branch,
		handler:   opts.Handler,
		logger:    opts.Logger,
		idFunc:    idFunc,
	}, nil
}

// Run connects, sends the subscribe request, then reads events until the
// context is cancelled or the stream ends. It returns the terminating error,
// which is nil on context cancellation.
func (c *Client) Run(ctx context.Context) error {
	if err := c.transport.Connect(ctx); err != nil {
		return err
	}
	defer func() { _ = c.transport.Close() }()

	subscribe := BuildSubscribeEnvelope(c.idFunc(), c.repoID, c.branch)
	if err := c.transport.Write(ctx, subscribe); err != nil {
		return err
	}
	c.logInfo("gito branch event subscription sent", "repo_id", c.repoID, "branch", c.branch)

	for {
		if ctx.Err() != nil {
			return nil
		}
		env, err := c.transport.Read(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return err
		}
		c.handleEnvelope(ctx, env)
	}
}

func (c *Client) handleEnvelope(ctx context.Context, env protosocket.Envelope) {
	// Ignore non-event traffic (e.g. the subscribe response) silently.
	if env.Channel != EventChannel || env.Action != BranchUpdatedAction {
		c.logInfo("gito envelope ignored", "channel", env.Channel, "action", env.Action)
		return
	}

	event, err := DecodeBranchUpdatedEnvelope(env)
	if err != nil {
		c.logError("gito branch event decode failed", "error", err, "channel", env.Channel, "action", env.Action)
		return
	}

	if !event.IsTarget(c.repoID, c.branch) {
		c.logInfo("gito branch event off-target",
			"event_repo_id", event.RepoID,
			"event_branch", event.Branch,
			"want_repo_id", c.repoID,
			"want_branch", c.branch,
		)
		return
	}

	if c.handler != nil {
		c.handler(ctx, event)
	}
}

func (c *Client) logInfo(msg string, args ...any) {
	if c.logger == nil {
		return
	}
	c.logger.Info(msg, args...)
}

func (c *Client) logError(msg string, args ...any) {
	if c.logger == nil {
		return
	}
	c.logger.Error(msg, args...)
}