// 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 { GetRoadmapSyncIdentityByWorkItem(ctx context.Context, args db.GetRoadmapSyncIdentityByWorkItemParams) (db.RoadmapSyncIdentity, error) UpsertRoadmapSyncIdentity(ctx context.Context, args db.UpsertRoadmapSyncIdentityByWorkItemParams) (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) } // 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 } // 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} } // 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) { candidate := roadmapsync.MatchDevelopMilestone(roadmapsync.MatchInput{ Scan: in.Scan, Expected: in.Expected, }) if !candidate.Ready { return SyncCreationResult{Action: SyncActionNotReady, Reason: candidate.Reason}, nil } projectionInput := roadmapsync.ProjectPlaneTodoInput{ Candidate: candidate, Ref: in.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 } identityRow, completed, err := s.loadOrUpsertIdentityAndSteps(ctx, in, candidate) if err != nil { return SyncCreationResult{}, err } decision := roadmapsync.ReconcileCreationCycle(roadmapsync.ReconcileInput{ Identity: candidate.Identity, ProviderRevision: in.ProviderRevision, RoadmapRevision: in.RoadmapRevision, Actor: in.Actor, SelfActor: s.selfActor, Existing: existingFromRow(identityRow), CompletedSteps: toReconcileSteps(completed), }) switch decision.Action { case roadmapsync.ActionComplete: 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)) } 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 } // loadOrUpsertIdentityAndSteps finds the identity row by provider work item and // upserts a fresh row when none exists, then reads its completed-step set. The // upsert reuses storage's atomic conflict handling so a concurrent re-process // cannot rebind one identity onto another. func (s *Service) loadOrUpsertIdentityAndSteps(ctx context.Context, in SyncCreationInput, candidate roadmapsync.ProjectionCandidate) (db.RoadmapSyncIdentity, map[storage.RoadmapSyncStep]bool, error) { identity := candidate.Identity 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, err } row, err = s.store.UpsertRoadmapSyncIdentity(ctx, upsertParams(identity, in)) if err != nil { return db.RoadmapSyncIdentity{}, nil, err } } completed, err := s.store.CompletedSteps(ctx, row.ID) if err != nil { return db.RoadmapSyncIdentity{}, nil, err } return row, completed, 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. func (s *Service) runStep(ctx context.Context, step roadmapsync.Step, in roadmapsync.ProjectPlaneTodoInput) error { switch step { case roadmapsync.StepDevelopMatched: 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.UpsertRoadmapSyncIdentityByWorkItemParams { return db.UpsertRoadmapSyncIdentityByWorkItemParams{ 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 }