nomadcode/services/core/internal/roadmapsyncpipeline/service.go

// 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
}