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 { if ctx.Err() != nil { return 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 { if ctx.Err() != nil { return 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 len(event.MilestoneChangedFiles()) == 0 { c.logInfo("gito branch event ignored: no roadmap milestone changes", "event_repo_id", event.RepoID, "event_branch", event.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...) }