// Package roadmapsyncpipeline owns the runtime side-effect orchestration that // drives a Plane-origin Milestone creation sync from a develop scan result all // the way to the projected Plane Todo. The provider-neutral decisions live in // internal/roadmapsync (match, reconcile, projection formatting) and the // persisted identity/step ledger lives in internal/storage; this package is the // caller that wires them together and records a per-step ledger mark right after // each provider side effect succeeds, so a retry resumes only the remaining // steps. package roadmapsyncpipeline import ( "context" "errors" "github.com/nomadcode/nomadcode-core/internal/db" "github.com/nomadcode/nomadcode-core/internal/roadmapsync" "github.com/nomadcode/nomadcode-core/internal/storage" "github.com/nomadcode/nomadcode-core/internal/workitem" ) // SyncAction is the typed outcome of a single SyncCreation call. It mirrors the // reconcile decision and the develop match gate so a worker can log and branch // without re-deriving the cause. type SyncAction string const ( // SyncActionNotReady means the develop scan did not match the expected work // item identity, so no provider side effect ran. SyncActionNotReady SyncAction = "not_ready" // SyncActionComplete means every creation-cycle step is already recorded; // nothing was projected this call. SyncActionComplete SyncAction = "complete" // SyncActionConflict means the trigger disagreed with the persisted identity // (revision or identity-key mismatch); a human/upstream decision is required // rather than a silent overwrite, so no provider side effect ran. SyncActionConflict SyncAction = "conflict" // SyncActionSkipSelfMutation means the trigger was NomadCode's own mutation, // so it is not treated as a new authoring trigger and no side effect ran. SyncActionSkipSelfMutation SyncAction = "skip_self_mutation" // SyncActionProjected means the remaining steps ran and each was recorded in // the ledger. SyncActionProjected SyncAction = "projected" ) // Store is the persistence seam SyncCreation needs: find or upsert the identity // row, read the completed-step set, and mark a step completed. It is satisfied // by *storage.Store and by test fakes so the orchestrator stays unit-testable // without a database. type Store interface { GetRoadmapSyncIdentityByMilestoneID(ctx context.Context, milestoneID string) (db.RoadmapSyncIdentity, error) GetRoadmapSyncIdentityByWorkItem(ctx context.Context, args db.GetRoadmapSyncIdentityByWorkItemParams) (db.RoadmapSyncIdentity, error) UpsertRoadmapSyncIdentity(ctx context.Context, args db.UpsertRoadmapSyncIdentityByMilestoneIDParams) (db.RoadmapSyncIdentity, error) CompletedSteps(ctx context.Context, identityID int64) (map[storage.RoadmapSyncStep]bool, error) MarkStepCompleted(ctx context.Context, identityID int64, step storage.RoadmapSyncStep) (db.RoadmapSyncStep, error) } // Backfiller writes the provider identity block to the develop checkout, // commits only that block, and pushes it to develop. SyncCreation calls it // right after a Plane work item is created (or on resume after a prior push // failure) and only marks the identity_backfilled ledger step once it // returns nil, so a retry never records the step for a push that did not // actually land. type Backfiller interface { BackfillIdentity(ctx context.Context, in BackfillInput) error } // BackfillInput carries the identity to inject, the develop-relative Milestone // path to inject it into, and the scan-verified develop revision the backfill // commit must be based on. RoadmapRevision is the revision the scanner verified // on origin/develop; the backfiller refuses to commit on top of a revision that // no longer matches it, so a Core-authored backfill never lands on stale state. type BackfillInput struct { Identity roadmapsync.Identity MarkdownPath string RoadmapRevision string } // Service drives the creation sync orchestration against a Store and the Plane // Todo provider facets. type Service struct { store Store provider roadmapsync.PlaneTodoProvider // selfActor identifies NomadCode's own provider account so a mutation it made // is not re-interpreted as a fresh user trigger. selfActor string // backfiller performs the Core-authored identity backfill commit/push for // missing-create identities. Nil means backfill is not configured; a // missing-create cycle then fails explicitly instead of silently // completing with only the DB identity. backfiller Backfiller } // NewService builds the orchestrator. provider is the Plane adapter (or a fake) // that satisfies the comment/body/status facets; selfActor is the provider // account NomadCode mutates as. func NewService(store Store, provider roadmapsync.PlaneTodoProvider, selfActor string) *Service { return &Service{store: store, provider: provider, selfActor: selfActor} } // SetBackfiller configures the identity backfiller. It is a post-construction // setter (like workflow.Service.SetEnqueuer) so callers that do not need // missing-create identity backfill can keep using NewService unchanged. func (s *Service) SetBackfiller(b Backfiller) { s.backfiller = b } // SyncCreationInput carries everything one creation sync needs without reaching // into provider DTOs: the develop scan result and the expected work item // identity (the develop match inputs), the Plane work item ref, the original // Plane body to preserve, the Todo state id to move to, the develop Milestone // markdown to project, the actor that caused the trigger, the provider/roadmap // revisions the trigger observed, and the external source/id tagging the // preservation comment. type SyncCreationInput struct { Scan roadmapsync.ScanResult Expected roadmapsync.Identity Ref workitem.Ref OriginalBody string TodoStateID string MilestoneMarkdown string Actor string ProviderRevision string RoadmapRevision string ExternalSource string ExternalID string } // SyncCreationResult is the typed outcome a worker logs and branches on. Reason // is always populated so a non-projected result is never silent. NextStep is set // only for SyncActionProjected and names the last step the orchestrator // recorded. type SyncCreationResult struct { Action SyncAction Reason string NextStep roadmapsync.Step } // SyncCreation runs one creation sync cycle: gate on the develop match, load or // upsert the identity row, reconcile against the persisted step ledger, and — // only when the reconcile says resume — run the remaining provider steps in // order, recording a ledger mark right after each provider call succeeds. A // not-ready/conflict/self-mutation/complete decision returns without any // provider side effect. func (s *Service) SyncCreation(ctx context.Context, in SyncCreationInput) (SyncCreationResult, error) { var existingRow db.RoadmapSyncIdentity var completed map[storage.RoadmapSyncStep]bool var found bool var err error // missingCreateShaped is computed from the trigger's original identity // fields before the found-identity infill below overwrites them, so a // redelivered missing-create job (no work item id/ref on the trigger) is // still recognized as such even after an earlier cycle already created // the Plane item and populated the persisted row. missingCreateShaped := in.Expected.WorkItemID == "" && in.Ref.ID == "" // Load existing identity row (without mutation) if in.Expected.MilestoneID() != "" { existingRow, completed, found, err = s.loadExistingByID(ctx, in.Expected.MilestoneID()) if err != nil { return SyncCreationResult{}, err } } if found { // Fill in missing details from existingRow, but do not overwrite trigger-provided non-empty values if in.Expected.Provider == "" { in.Expected.Provider = workitem.ProviderID(existingRow.Provider) } if in.Expected.Tenant == "" { in.Expected.Tenant = existingRow.Tenant } if in.Expected.Project == "" { in.Expected.Project = existingRow.Project } if in.Expected.WorkItemID == "" { in.Expected.WorkItemID = existingRow.WorkItemID } if in.Expected.ParentWorkItemID == "" { in.Expected.ParentWorkItemID = existingRow.ParentWorkItemID } if in.Ref.Provider == "" { in.Ref.Provider = in.Expected.Provider } if in.Ref.Tenant == "" { in.Ref.Tenant = in.Expected.Tenant } if in.Ref.Project == "" { in.Ref.Project = in.Expected.Project } if in.Ref.ID == "" { in.Ref.ID = in.Expected.WorkItemID } // A prior cycle already created the Plane item and upserted the // identity (existingRow.WorkItemID set), but the identity backfill // commit/push never completed — resume the backfill directly instead // of falling through to develop-match, which cannot succeed until the // backfill push actually lands the identity block on develop. if missingCreateShaped && existingRow.WorkItemID != "" && !completed[storage.StepIdentityBackfilled] { return s.backfillIdentityAndMark(ctx, existingRow, existingFromRow(existingRow).Identity, in.Scan.Revision.Normalize().Revision) } } else if missingCreateShaped { // Plane missing-create scenario: no existing work item id means the // active milestone has no Plane ticket yet. Expected.Provider is set by // the bridge (e.g. "plane"); fall back to "plane" only when the caller // omits it. if in.Expected.Tenant == "" || in.Expected.Project == "" { return SyncCreationResult{}, errors.New("roadmapsyncpipeline: tenant and project are required for missing-create") } creator, ok := s.provider.(workitem.Creator) if !ok { return SyncCreationResult{}, errors.New("roadmapsyncpipeline: provider does not support creation") } providerID := in.Expected.Provider if string(providerID) == "" { providerID = "plane" } projection := roadmapsync.FormatPlaneProjection(in.Expected.RoadmapMilestonePath, in.MilestoneMarkdown) createRes, err := creator.CreateWorkItem(ctx, workitem.CreateInput{ Provider: providerID, Tenant: in.Expected.Tenant, Project: in.Expected.Project, Title: projection.Title, DescriptionHTML: projection.BodyHTML, }) if err != nil { return SyncCreationResult{}, err } identity := roadmapsync.Identity{ Shape: roadmapsync.ShapeMilestone, RoadmapMilestonePath: in.Expected.RoadmapMilestonePath, RoadmapItemID: in.Expected.RoadmapItemID, Provider: createRes.Ref.Provider, Tenant: createRes.Ref.Tenant, Project: createRes.Ref.Project, WorkItemID: createRes.Ref.ID, } upsertParams := db.UpsertRoadmapSyncIdentityByMilestoneIDParams{ Shape: string(identity.Shape), RoadmapMilestonePath: identity.RoadmapMilestonePath, RoadmapItemID: identity.RoadmapItemID, Provider: string(identity.Provider), Tenant: identity.Tenant, Project: identity.Project, WorkItemID: identity.WorkItemID, ParentWorkItemID: identity.ParentWorkItemID, RoadmapRevision: nullableString(in.RoadmapRevision), } identityRow, err := s.store.UpsertRoadmapSyncIdentity(ctx, upsertParams) if err != nil { if errors.Is(err, storage.ErrRoadmapSyncIdentityConflict) { return SyncCreationResult{ Action: SyncActionConflict, Reason: "roadmap sync identity conflict: work item already bound or filename reuse suspected", }, nil } return SyncCreationResult{}, err } if _, err := s.store.MarkStepCompleted(ctx, identityRow.ID, toStorageStep(roadmapsync.StepDevelopMatched)); err != nil { return SyncCreationResult{}, err } return s.backfillIdentityAndMark(ctx, identityRow, identity, in.Scan.Revision.Normalize().Revision) } candidate := roadmapsync.MatchDevelopMilestone(roadmapsync.MatchInput{ Scan: in.Scan, Expected: in.Expected, }) if !candidate.Ready { return SyncCreationResult{Action: SyncActionNotReady, Reason: candidate.Reason}, nil } ref := in.Ref if candidate.Identity.WorkItemID != "" { ref.ID = candidate.Identity.WorkItemID } if string(candidate.Identity.Provider) != "" { ref.Provider = candidate.Identity.Provider } if candidate.Identity.Tenant != "" { ref.Tenant = candidate.Identity.Tenant } if candidate.Identity.Project != "" { ref.Project = candidate.Identity.Project } if in.Ref.Provider == ref.Provider && in.Ref.Tenant == ref.Tenant && in.Ref.Project == ref.Project && in.Ref.ID == ref.ID { ref = in.Ref } projectionInput := roadmapsync.ProjectPlaneTodoInput{ Candidate: candidate, Ref: ref, OriginalBody: in.OriginalBody, TodoStateID: in.TodoStateID, MilestoneMarkdown: in.MilestoneMarkdown, ExternalSource: in.ExternalSource, ExternalID: in.ExternalID, } // Reject missing projection inputs before any identity write or provider call, // matching ProjectPlaneTodo's guard. if err := roadmapsync.ValidateProjectionInput(s.provider, projectionInput); err != nil { return SyncCreationResult{}, err } // 1. Load existing identity row (without mutation) if !found { existingRow, completed, found, err = s.loadExistingIdentityAndSteps(ctx, candidate) if err != nil { return SyncCreationResult{}, err } } existingView := roadmapsync.ExistingIdentity{Found: found} if found { existingView = existingFromRow(existingRow) } // 2. Reconcile conflict using original existing row decision := roadmapsync.ReconcileCreationCycle(roadmapsync.ReconcileInput{ Identity: candidate.Identity, ProviderRevision: in.ProviderRevision, RoadmapRevision: in.RoadmapRevision, Actor: in.Actor, SelfActor: s.selfActor, Existing: existingView, CompletedSteps: toReconcileSteps(completed), }) switch decision.Action { case roadmapsync.ActionComplete: // If attributes changed, update them in the DB before returning complete if found && (existingRow.RoadmapMilestonePath != candidate.Identity.RoadmapMilestonePath || existingRow.Provider != string(candidate.Identity.Provider) || existingRow.Tenant != candidate.Identity.Tenant || existingRow.Project != candidate.Identity.Project || existingRow.WorkItemID != candidate.Identity.WorkItemID || derefString(existingRow.ProviderRevision) != in.ProviderRevision || derefString(existingRow.RoadmapRevision) != in.RoadmapRevision) { _, err = s.store.UpsertRoadmapSyncIdentity(ctx, upsertParams(candidate.Identity, in)) if err != nil { if errors.Is(err, storage.ErrRoadmapSyncIdentityConflict) { return SyncCreationResult{ Action: SyncActionConflict, Reason: "roadmap sync identity conflict: work item already bound or filename reuse suspected", }, nil } return SyncCreationResult{}, err } } return SyncCreationResult{Action: SyncActionComplete, Reason: decision.Reason}, nil case roadmapsync.ActionConflict: return SyncCreationResult{Action: SyncActionConflict, Reason: decision.Reason}, nil case roadmapsync.ActionSkipSelfMutation: return SyncCreationResult{Action: SyncActionSkipSelfMutation, Reason: decision.Reason}, nil case roadmapsync.ActionResume: // fall through to step execution below. default: return SyncCreationResult{}, errors.New("roadmapsyncpipeline: unexpected reconcile action " + string(decision.Action)) } // 3. Reconcile says Resume: Perform mutation (upsert) to record/update path, work item attrs, revisions. var identityRow db.RoadmapSyncIdentity if !found { identityRow, err = s.store.UpsertRoadmapSyncIdentity(ctx, upsertParams(candidate.Identity, in)) if err != nil { if errors.Is(err, storage.ErrRoadmapSyncIdentityConflict) { return SyncCreationResult{ Action: SyncActionConflict, Reason: "roadmap sync identity conflict: work item already bound or filename reuse suspected", }, nil } return SyncCreationResult{}, err } } else { // Update only if anything changed if existingRow.RoadmapMilestonePath != candidate.Identity.RoadmapMilestonePath || existingRow.Provider != string(candidate.Identity.Provider) || existingRow.Tenant != candidate.Identity.Tenant || existingRow.Project != candidate.Identity.Project || existingRow.WorkItemID != candidate.Identity.WorkItemID || derefString(existingRow.ProviderRevision) != in.ProviderRevision || derefString(existingRow.RoadmapRevision) != in.RoadmapRevision { identityRow, err = s.store.UpsertRoadmapSyncIdentity(ctx, upsertParams(candidate.Identity, in)) if err != nil { if errors.Is(err, storage.ErrRoadmapSyncIdentityConflict) { return SyncCreationResult{ Action: SyncActionConflict, Reason: "roadmap sync identity conflict: work item already bound or filename reuse suspected", }, nil } return SyncCreationResult{}, err } } else { identityRow = existingRow } } lastStep, err := s.runRemainingSteps(ctx, identityRow.ID, decision.NextStep, completed, projectionInput) if err != nil { return SyncCreationResult{}, err } return SyncCreationResult{Action: SyncActionProjected, Reason: decision.Reason, NextStep: lastStep}, nil } // loadExistingIdentityAndSteps finds the existing identity row and its completed steps, // returning found=false if not found on either milestone ID or provider work item lookup. func (s *Service) loadExistingIdentityAndSteps(ctx context.Context, candidate roadmapsync.ProjectionCandidate) (db.RoadmapSyncIdentity, map[storage.RoadmapSyncStep]bool, bool, error) { identity := candidate.Identity // 1. Try canonical ID (MilestoneID) lookup first row, err := s.store.GetRoadmapSyncIdentityByMilestoneID(ctx, identity.MilestoneID()) if err != nil { if !errors.Is(err, storage.ErrRoadmapSyncIdentityNotFound) { return db.RoadmapSyncIdentity{}, nil, false, err } // 2. Fall back to provider work item lookup row, err = s.store.GetRoadmapSyncIdentityByWorkItem(ctx, db.GetRoadmapSyncIdentityByWorkItemParams{ Provider: string(identity.Provider), Tenant: identity.Tenant, Project: identity.Project, WorkItemID: identity.WorkItemID, }) if err != nil { if !errors.Is(err, storage.ErrRoadmapSyncIdentityNotFound) { return db.RoadmapSyncIdentity{}, nil, false, err } return db.RoadmapSyncIdentity{}, nil, false, nil } } completed, err := s.store.CompletedSteps(ctx, row.ID) if err != nil { return db.RoadmapSyncIdentity{}, nil, false, err } return row, completed, true, nil } func (s *Service) loadExistingByID(ctx context.Context, milestoneID string) (db.RoadmapSyncIdentity, map[storage.RoadmapSyncStep]bool, bool, error) { row, err := s.store.GetRoadmapSyncIdentityByMilestoneID(ctx, milestoneID) if err != nil { if !errors.Is(err, storage.ErrRoadmapSyncIdentityNotFound) { return db.RoadmapSyncIdentity{}, nil, false, err } return db.RoadmapSyncIdentity{}, nil, false, nil } completed, err := s.store.CompletedSteps(ctx, row.ID) if err != nil { return db.RoadmapSyncIdentity{}, nil, false, err } return row, completed, true, nil } // backfillIdentityAndMark runs the Core-authored identity backfill commit/push // for identityRow and, only once it succeeds, marks the identity_backfilled // ledger step. A missing backfiller fails explicitly rather than silently // completing with only the DB identity, so a misconfigured deployment cannot // mistake a DB-only write for a durable, provider-verifiable identity. func (s *Service) backfillIdentityAndMark(ctx context.Context, identityRow db.RoadmapSyncIdentity, identity roadmapsync.Identity, roadmapRevision string) (SyncCreationResult, error) { if s.backfiller == nil { return SyncCreationResult{}, errors.New("roadmapsyncpipeline: identity backfiller not configured") } if err := s.backfiller.BackfillIdentity(ctx, BackfillInput{Identity: identity, MarkdownPath: identity.RoadmapMilestonePath, RoadmapRevision: roadmapRevision}); err != nil { return SyncCreationResult{}, err } if _, err := s.store.MarkStepCompleted(ctx, identityRow.ID, toStorageStep(roadmapsync.StepIdentityBackfilled)); err != nil { return SyncCreationResult{}, err } return SyncCreationResult{ Action: SyncActionProjected, Reason: "provider identity backfilled to develop", NextStep: roadmapsync.StepIdentityBackfilled, }, nil } // runRemainingSteps executes the creation-cycle steps from nextStep onward, // skipping any already in the completed set, and records a ledger mark right // after each provider call succeeds. A provider failure stops the sequence so a // later step never runs on an unpreserved/unupdated ticket, and the ledger mark // is never written for a step whose provider call did not succeed. It returns // the last step it recorded. func (s *Service) runRemainingSteps(ctx context.Context, identityID int64, nextStep roadmapsync.Step, completed map[storage.RoadmapSyncStep]bool, projectionInput roadmapsync.ProjectPlaneTodoInput) (roadmapsync.Step, error) { var last roadmapsync.Step resuming := false for _, step := range roadmapsync.OrderedSteps { if step == nextStep { resuming = true } if !resuming { continue } if completed[toStorageStep(step)] { continue } if err := s.runStep(ctx, step, projectionInput); err != nil { return "", err } if _, err := s.store.MarkStepCompleted(ctx, identityID, toStorageStep(step)); err != nil { return "", err } last = step } return last, nil } // runStep performs the provider side effect for a single creation-cycle step. // StepDevelopMatched has no provider side effect — it only records that the // develop match gate passed — so it is a no-op here and the caller still records // its ledger mark. StepIdentityBackfilled is likewise a no-op here: by the time // this general resume loop reaches it, the develop scan already matched the // candidate against a Milestone doc that carries the identity block, so the // content backfill Core would otherwise perform is already satisfied; the // dedicated missing-create/resume path in SyncCreation is the only place that // actually calls the Backfiller. func (s *Service) runStep(ctx context.Context, step roadmapsync.Step, in roadmapsync.ProjectPlaneTodoInput) error { switch step { case roadmapsync.StepDevelopMatched, roadmapsync.StepIdentityBackfilled: return nil case roadmapsync.StepOriginalCommentPreserved: return roadmapsync.PreserveOriginalComment(ctx, s.provider, in) case roadmapsync.StepPlaneBodyUpdated: return roadmapsync.UpdatePlaneBody(ctx, s.provider, in) case roadmapsync.StepPlaneTodoMoved: return roadmapsync.MovePlaneTodo(ctx, s.provider, in) default: return errors.New("roadmapsyncpipeline: unknown creation cycle step " + string(step)) } } // upsertParams maps the matched identity and trigger revisions to the storage // upsert parameters. Empty revisions are stored as NULL so a first cycle that // has not observed a revision is not treated as a conflicting empty string. func upsertParams(identity roadmapsync.Identity, in SyncCreationInput) db.UpsertRoadmapSyncIdentityByMilestoneIDParams { return db.UpsertRoadmapSyncIdentityByMilestoneIDParams{ Shape: string(identity.Shape), RoadmapMilestonePath: identity.RoadmapMilestonePath, RoadmapItemID: identity.RoadmapItemID, Provider: string(identity.Provider), Tenant: identity.Tenant, Project: identity.Project, WorkItemID: identity.WorkItemID, ParentWorkItemID: identity.ParentWorkItemID, ProviderRevision: nullableString(in.ProviderRevision), RoadmapRevision: nullableString(in.RoadmapRevision), } } // existingFromRow maps a persisted identity row to the reconcile's // provider-neutral existing-identity view. func existingFromRow(row db.RoadmapSyncIdentity) roadmapsync.ExistingIdentity { return roadmapsync.ExistingIdentity{ Found: row.ID != 0, Identity: roadmapsync.Identity{ Shape: roadmapsync.Shape(row.Shape), RoadmapMilestonePath: row.RoadmapMilestonePath, RoadmapItemID: row.RoadmapItemID, Provider: workitem.ProviderID(row.Provider), Tenant: row.Tenant, Project: row.Project, WorkItemID: row.WorkItemID, ParentWorkItemID: row.ParentWorkItemID, }, ProviderRevision: derefString(row.ProviderRevision), RoadmapRevision: derefString(row.RoadmapRevision), } } // toReconcileSteps maps the storage step set to the reconcile step set. func toReconcileSteps(completed map[storage.RoadmapSyncStep]bool) map[roadmapsync.Step]bool { out := make(map[roadmapsync.Step]bool, len(completed)) for step, done := range completed { if done { out[roadmapsync.Step(step)] = true } } return out } // toStorageStep maps a reconcile/domain step to the storage step. The two enums // share the same string values; the helper keeps the conversion explicit at the // package boundary. func toStorageStep(step roadmapsync.Step) storage.RoadmapSyncStep { return storage.RoadmapSyncStep(step) } func nullableString(v string) *string { if v == "" { return nil } return &v } func derefString(v *string) string { if v == nil { return "" } return *v }