package gitosync import ( "context" "errors" "log/slog" "strings" "sync" "github.com/nomadcode/nomadcode-core/internal/gitoevents" "github.com/nomadcode/nomadcode-core/internal/roadmapsync" "github.com/nomadcode/nomadcode-core/internal/scheduler" "github.com/nomadcode/nomadcode-core/internal/workitem" ) // Scanner is the develop-scan seam the bridge drives. It is satisfied by // *BranchRevisionScanner and by test fakes. type Scanner interface { Scan(ctx context.Context, ev gitoevents.BranchUpdatedEvent) (ScanOutput, error) } // Enqueuer is the scheduler seam the bridge hands a built job to. It is // satisfied by *scheduler.Client and by test fakes, so the bridge stays // testable without River or a database. type Enqueuer interface { EnqueueRoadmapCreationSync(ctx context.Context, args scheduler.RoadmapCreationSyncJobArgs) error } // ProcessedRevisionStore is the duplicate guard seam: it reports whether a // (repo, branch, after) revision has already been processed and records it. // MarkProcessed must only be called after the revision's enqueue work // succeeded, so a failed run is retried rather than silently swallowed. type ProcessedRevisionStore interface { Seen(repoID, branch, after string) bool MarkProcessed(repoID, branch, after string) } // BridgeConfig configures a Bridge. type BridgeConfig struct { RepoID string Branch string // TodoStateID is the Plane state id the creation sync moves the work item // to. Required; an empty value disables the bridge at construction. TodoStateID string // ExternalSource/ExternalID tag the preservation comment so a re-run is // recognized as nomadcode-authored. ExternalSource defaults to "nomadcode". ExternalSource string Tenant string Project string } // Bridge turns a verified develop scan into one EnqueueRoadmapCreationSync call // per active Milestone. It validates the event's repo/branch, dedups by // (repo, branch, after) revision, fetches the originating Plane work item for // identity-backed docs, and enqueues identity-less docs as missing-create jobs. type Bridge struct { scanner Scanner reader workitem.Reader enqueuer Enqueuer processed ProcessedRevisionStore cfg BridgeConfig logger *slog.Logger } // NewBridge builds a Bridge. scanner, reader, enqueuer, and a non-empty // TodoStateID are required. func NewBridge(scanner Scanner, reader workitem.Reader, enqueuer Enqueuer, processed ProcessedRevisionStore, cfg BridgeConfig, logger *slog.Logger) (*Bridge, error) { if scanner == nil { return nil, errors.New("gitosync: scanner is nil") } if reader == nil { return nil, errors.New("gitosync: work item reader is nil") } if enqueuer == nil { return nil, errors.New("gitosync: enqueuer is nil") } if strings.TrimSpace(cfg.TodoStateID) == "" { return nil, errors.New("gitosync: todo state id is required") } if processed == nil { processed = NewInMemoryRevisionStore() } if strings.TrimSpace(cfg.ExternalSource) == "" { cfg.ExternalSource = "nomadcode" } return &Bridge{ scanner: scanner, reader: reader, enqueuer: enqueuer, processed: processed, cfg: cfg, logger: logger, }, nil } // Handle processes one branch event: it gates on the target repo/branch and the // duplicate-revision guard, scans develop, and enqueues a creation sync job for // each matched Milestone. ErrNotReady from the scanner is a normal not-yet // outcome (the revision is not recorded as processed) and is returned to the // caller to log. A wrong repo/branch is a silent no-op. func (b *Bridge) Handle(ctx context.Context, ev gitoevents.BranchUpdatedEvent) error { if !ev.IsTarget(b.cfg.RepoID, b.cfg.Branch) { b.logInfo("gito event off-target, ignored", "event_repo_id", ev.RepoID, "event_branch", ev.Branch) return nil } if b.processed.Seen(ev.RepoID, ev.Branch, ev.After) { b.logInfo("gito revision already processed, skipped", "repo_id", ev.RepoID, "branch", ev.Branch, "after", ev.After) return nil } out, err := b.scanner.Scan(ctx, ev) if err != nil { // Not-ready is a normal not-yet case; leave the revision unprocessed so a // later event for the same after can retry once develop catches up. return err } if out.SelfAuthored { // The wakeup is NomadCode's own identity-backfill commit looping back. It // is still scanned and enqueued normally — the ledger, not this hint, // decides whether the redelivery is a complete/no-op — but logging it makes // a self-loop visible to an operator rather than looking like fresh work. b.logInfo("gito self-authored backfill revision observed", "repo_id", ev.RepoID, "branch", ev.Branch, "after", ev.After) } for _, doc := range out.Docs { if err := b.enqueueDoc(ctx, out.Result, doc); err != nil { return err } } // Every matched Milestone was enqueued (or there were none); record the // revision so a duplicate event does not re-enqueue the same jobs. b.processed.MarkProcessed(ev.RepoID, ev.Branch, ev.After) return nil } // enqueueDoc fetches the originating Plane work item, builds the job args, and // enqueues one creation sync job for a single scanned Milestone. Missing work // items can be stale roadmap identities, so they are skipped while other fetch // failures still surface. func (b *Bridge) enqueueDoc(ctx context.Context, scan roadmapsync.ScanResult, doc ScannedMilestoneDoc) error { if !doc.HasIdentity { providerID := workitem.ProviderID("plane") if prov, ok := b.reader.(workitem.Provider); ok { providerID = prov.ProviderID() } args := scheduler.RoadmapCreationSyncJobArgs{ Scan: scan, Expected: roadmapsync.Identity{ Shape: roadmapsync.ShapeMilestone, RoadmapMilestonePath: doc.Path, RoadmapItemID: doc.MilestoneID, Provider: providerID, Tenant: b.cfg.Tenant, Project: b.cfg.Project, }, TodoStateID: b.cfg.TodoStateID, MilestoneMarkdown: doc.Markdown, // RoadmapRevision is intentionally omitted on missing-create: the identity // has no completed projection steps yet, so a later reconcile push (after // the authoring roundtrip adds the Plane identity to the milestone file) // must not conflict on revision. ExternalSource: b.cfg.ExternalSource, ExternalID: "", } if err := b.enqueuer.EnqueueRoadmapCreationSync(ctx, args); err != nil { return err } b.logInfo("gito missing-create sync enqueued", "milestone_path", doc.Path, "revision", scan.Revision.Revision, ) return nil } identity := doc.Identity if identity.Tenant == "" { identity.Tenant = b.cfg.Tenant } if identity.Project == "" { identity.Project = b.cfg.Project } ref := workitem.Ref{ Provider: identity.Provider, Tenant: identity.Tenant, Project: identity.Project, ID: identity.WorkItemID, } item, err := b.reader.FetchWorkItem(ctx, ref) if err != nil { if errors.Is(err, workitem.ErrNotFound) { b.logInfo("gito scanned milestone work item missing, skipped", "work_item_id", identity.WorkItemID, "milestone_path", identity.RoadmapMilestonePath, "revision", scan.Revision.Revision, ) return nil } return err } args := scheduler.RoadmapCreationSyncJobArgs{ Scan: scan, Expected: identity, Ref: ref, OriginalBody: originalBody(item), TodoStateID: b.cfg.TodoStateID, MilestoneMarkdown: doc.Markdown, // ProviderRevision is intentionally left empty. A Gito develop push has no // stable provider revision token; the Plane work item State is mutable and // the sync itself moves it to the Todo state, so feeding State here would // make a re-process of the same develop revision observe a different // provider revision and trip the reconcile's provider-revision conflict // before the completed-step ledger is even consulted. The develop revision // in RoadmapRevision is the stable guard for a different-revision reprocess. RoadmapRevision: scan.Revision.Revision, ExternalSource: b.cfg.ExternalSource, ExternalID: identity.WorkItemID, } if err := b.enqueuer.EnqueueRoadmapCreationSync(ctx, args); err != nil { return err } b.logInfo("gito creation sync enqueued", "work_item_id", identity.WorkItemID, "milestone_path", identity.RoadmapMilestonePath, "revision", scan.Revision.Revision, ) return nil } // originalBody prefers the HTML description and falls back to the text // description so the preservation comment keeps the original Plane body. func originalBody(item workitem.WorkItem) string { if strings.TrimSpace(item.DescriptionHTML) != "" { return item.DescriptionHTML } return item.DescriptionText } func (b *Bridge) logInfo(msg string, args ...any) { if b.logger == nil { return } b.logger.Info(msg, args...) } // InMemoryRevisionStore is the MVP duplicate guard: a process-lifetime set of // processed (repo, branch, after) revisions. It is intentionally not persisted; // the durable idempotency lives in the identity/ledger the sync step uses, and // this only suppresses repeated enqueues within one process run. type InMemoryRevisionStore struct { mu sync.Mutex seen map[string]struct{} } // NewInMemoryRevisionStore builds an empty in-memory revision store. func NewInMemoryRevisionStore() *InMemoryRevisionStore { return &InMemoryRevisionStore{seen: make(map[string]struct{})} } func (s *InMemoryRevisionStore) Seen(repoID, branch, after string) bool { s.mu.Lock() defer s.mu.Unlock() _, ok := s.seen[revisionKey(repoID, branch, after)] return ok } func (s *InMemoryRevisionStore) MarkProcessed(repoID, branch, after string) { s.mu.Lock() defer s.mu.Unlock() s.seen[revisionKey(repoID, branch, after)] = struct{}{} } func revisionKey(repoID, branch, after string) string { return repoID + "\x00" + branch + "\x00" + after }