// 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) } // 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) { var existingRow db.RoadmapSyncIdentity var completed map[storage.RoadmapSyncStep]bool var found bool var err error // 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 { // Populate identity details from the DB in.Expected.Provider = workitem.ProviderID(existingRow.Provider) in.Expected.Tenant = existingRow.Tenant in.Expected.Project = existingRow.Project in.Expected.WorkItemID = existingRow.WorkItemID in.Expected.ParentWorkItemID = existingRow.ParentWorkItemID in.Ref = workitem.Ref{ Provider: in.Expected.Provider, Tenant: in.Expected.Tenant, Project: in.Expected.Project, ID: in.Expected.WorkItemID, } } else if in.Expected.WorkItemID == "" && in.Ref.ID == "" { // 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 SyncCreationResult{ Action: SyncActionProjected, Reason: "plane work item created for missing active milestone", NextStep: roadmapsync.StepDevelopMatched, }, nil } 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 } // 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: 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 } // 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.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 }