nomadcode/services/core/internal/gitosync/bridge.go

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
	}

	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. A scanned
// identity whose provider/work item id does not resolve to a fetchable work
// item is a hard error so a misconfigured identity is surfaced, not swallowed.
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 {
		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
}