gito/services/core/internal/controlplane/runtime.go

package controlplane

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"git.toki-labs.com/toki/gito/services/core/internal/core"
	"git.toki-labs.com/toki/gito/services/core/internal/events"
	"git.toki-labs.com/toki/gito/services/core/internal/gitengine"
	"git.toki-labs.com/toki/gito/services/core/internal/protosocket"
	"git.toki-labs.com/toki/gito/services/core/internal/storage"
	"git.toki-labs.com/toki/gito/services/core/internal/worker"
)

type EventBroadcaster interface {
	BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error
}

type Runtime struct {
	mu          sync.Mutex
	watches     map[string]BranchWatch
	records     []EventRecord
	broadcaster EventBroadcaster
	store       storage.Store
}

type BranchWatch struct {
	ID        string    `json:"id"`
	RepoID    string    `json:"repo_id"`
	Branch    string    `json:"branch"`
	Provider  string    `json:"provider"`
	CreatedAt time.Time `json:"created_at"`
}

type EventRecord struct {
	ID         string             `json:"id"`
	Type       string             `json:"type"`
	Provider   string             `json:"provider"`
	DeliveryID string             `json:"delivery_id,omitempty"`
	Duplicate  bool               `json:"duplicate,omitempty"`
	Revision   core.RevisionEvent `json:"revision"`
	CreatedAt  time.Time          `json:"created_at"`
}

func NewRuntime(broadcaster EventBroadcaster) *Runtime {
	return &Runtime{
		watches:     make(map[string]BranchWatch),
		broadcaster: broadcaster,
	}
}

// NewRuntimeWithStore constructs a Runtime backed by a durable store.
// When store is nil the behavior is identical to NewRuntime.
func NewRuntimeWithStore(broadcaster EventBroadcaster, store storage.Store) *Runtime {
	return &Runtime{
		watches:     make(map[string]BranchWatch),
		broadcaster: broadcaster,
		store:       store,
	}
}

func (r *Runtime) RegisterBranchWatch(repoID, branch, provider string) (BranchWatch, error) {
	repoID = strings.TrimSpace(repoID)
	branch = strings.TrimSpace(branch)
	provider = strings.TrimSpace(provider)
	if repoID == "" || branch == "" {
		return BranchWatch{}, fmt.Errorf("repo_id and branch are required")
	}
	if provider == "" {
		provider = "forgejo"
	}

	if r.store != nil && r.store.BranchWatches() != nil {
		cw := core.BranchWatch{
			ID:        "watch-" + stableWatchKey(repoID, branch, provider),
			Provider:  provider,
			RepoID:    repoID,
			Branch:    branch,
			CreatedAt: time.Now().UTC(),
		}
		stored, err := r.store.BranchWatches().UpsertBranchWatch(context.Background(), cw)
		if err != nil {
			return BranchWatch{}, fmt.Errorf("register branch watch: %w", err)
		}
		return branchWatchFromCore(stored), nil
	}

	watch := BranchWatch{
		ID:        "watch-" + stableWatchKey(repoID, branch, provider),
		RepoID:    repoID,
		Branch:    branch,
		Provider:  provider,
		CreatedAt: time.Now().UTC(),
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if existing, ok := r.watches[watch.ID]; ok {
		return existing, nil
	}
	r.watches[watch.ID] = watch
	return watch, nil
}

func (r *Runtime) ListBranchWatches() []BranchWatch {
	if r.store != nil && r.store.BranchWatches() != nil {
		stored, err := r.store.BranchWatches().ListBranchWatches(context.Background())
		if err != nil {
			// fall through to in-memory on transient error
		} else {
			watches := make([]BranchWatch, 0, len(stored))
			for _, w := range stored {
				watches = append(watches, branchWatchFromCore(w))
			}
			sort.Slice(watches, func(i, j int) bool {
				return watches[i].ID < watches[j].ID
			})
			return watches
		}
	}

	r.mu.Lock()
	defer r.mu.Unlock()
	watches := make([]BranchWatch, 0, len(r.watches))
	for _, watch := range r.watches {
		watches = append(watches, watch)
	}
	sort.Slice(watches, func(i, j int) bool {
		return watches[i].ID < watches[j].ID
	})
	return watches
}

func (r *Runtime) ListEvents() []EventRecord {
	r.mu.Lock()
	defer r.mu.Unlock()
	records := append([]EventRecord(nil), r.records...)
	sort.Slice(records, func(i, j int) bool {
		return records[i].CreatedAt.Before(records[j].CreatedAt)
	})
	return records
}

func (r *Runtime) HandleRevision(ctx context.Context, provider, deliveryID string, revision core.RevisionEvent) (EventRecord, bool, error) {
	provider = strings.TrimSpace(provider)
	if provider == "" {
		provider = "forgejo"
	}
	deliveryID = strings.TrimSpace(deliveryID)

	record := EventRecord{
		ID:         "event-" + newID(),
		Type:       events.BranchUpdated,
		Provider:   provider,
		DeliveryID: deliveryID,
		Revision:   revision,
		CreatedAt:  time.Now().UTC().Truncate(time.Microsecond),
	}

	if r.store != nil && r.store.BranchWatches() != nil {
		return r.handleRevisionWithStore(ctx, provider, deliveryID, revision, record)
	}

	r.mu.Lock()
	matched := r.matchesLocked(provider, revision)
	if matched {
		r.records = append(r.records, record)
	}
	r.mu.Unlock()

	if !matched {
		return record, false, nil
	}
	if r.broadcaster != nil {
		if err := r.broadcaster.BroadcastEnvelope(ctx, branchUpdatedEnvelope(record)); err != nil {
			return record, true, err
		}
	}
	return record, true, nil
}

func (r *Runtime) handleRevisionWithStore(ctx context.Context, provider, deliveryID string, revision core.RevisionEvent, record EventRecord) (EventRecord, bool, error) {
	_, matched, err := r.store.BranchWatches().FindBranchWatch(ctx, provider, revision.RepoID, revision.Branch)
	if err != nil {
		return record, false, fmt.Errorf("find branch watch: %w", err)
	}
	if !matched {
		return record, false, nil
	}

	var dedupeKey string
	if r.store.ProviderDeliveries() != nil {
		dedupeKey = dedupeKeyFor(deliveryID, revision)
		delivery := core.ProviderDelivery{
			ID:         newID(),
			Provider:   provider,
			DeliveryID: deliveryID,
			DedupeKey:  dedupeKey,
			EventID:    record.ID,
			RepoID:     revision.RepoID,
			Branch:     revision.Branch,
			Revision:   revision.After,
			CreatedAt:  record.CreatedAt,
		}
		result, err := r.store.ProviderDeliveries().RecordOnce(ctx, delivery)
		if err != nil {
			return record, true, fmt.Errorf("record delivery: %w", err)
		}
		if !result.First {
			// Return idempotent response using original delivery's event id and timestamp.
			record.ID = result.ExistingEventID
			record.CreatedAt = result.ExistingCreatedAt
			record.Duplicate = true
			return record, true, nil
		}
	}

	r.mu.Lock()
	r.records = append(r.records, record)
	r.mu.Unlock()

	if r.broadcaster != nil {
		if err := r.broadcaster.BroadcastEnvelope(ctx, branchUpdatedEnvelope(record)); err != nil {
			// Rollback delivery record on broadcast failure
			if r.store.ProviderDeliveries() != nil && dedupeKey != "" {
				_ = r.store.ProviderDeliveries().DeleteDelivery(ctx, provider, dedupeKey)
			}
			return record, true, err
		}
	}

	if r.store.RevisionCursors() != nil {
		cursor := core.RevisionCursor{
			RepoID:     revision.RepoID,
			Branch:     revision.Branch,
			Revision:   revision.After,
			ObservedAt: revision.ObservedAt,
		}
		if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
			return record, true, fmt.Errorf("upsert revision cursor: %w", err)
		}
	}

	return record, true, nil
}

func (r *Runtime) matchesLocked(provider string, revision core.RevisionEvent) bool {
	for _, watch := range r.watches {
		if watch.Provider == provider &&
			watch.RepoID == revision.RepoID &&
			watch.Branch == revision.Branch {
			return true
		}
	}
	return false
}

func dedupeKeyFor(deliveryID string, revision core.RevisionEvent) string {
	if deliveryID != "" {
		return "delivery:" + deliveryID
	}
	return fmt.Sprintf("revision:%s:%s:%s:%s", revision.RepoID, revision.Branch, revision.Before, revision.After)
}

func branchWatchFromCore(w core.BranchWatch) BranchWatch {
	return BranchWatch{
		ID:        w.ID,
		RepoID:    w.RepoID,
		Branch:    w.Branch,
		Provider:  w.Provider,
		CreatedAt: w.CreatedAt,
	}
}

func branchUpdatedEnvelope(record EventRecord) protosocket.Envelope {
	changedFiles := make([]any, 0, len(record.Revision.ChangedFiles))
	for _, file := range record.Revision.ChangedFiles {
		changedFiles = append(changedFiles, map[string]any{
			"path":        file.Path,
			"change_type": file.ChangeType,
		})
	}
	return protosocket.NewEventEnvelope(events.BranchUpdated, map[string]any{
		"id":            record.ID,
		"type":          record.Type,
		"provider":      record.Provider,
		"delivery_id":   record.DeliveryID,
		"repo_id":       record.Revision.RepoID,
		"branch":        record.Revision.Branch,
		"before":        record.Revision.Before,
		"after":         record.Revision.After,
		"changed_files": changedFiles,
		"observed_at":   record.Revision.ObservedAt.UTC().Format(time.RFC3339Nano),
		"created_at":    record.CreatedAt.UTC().Format(time.RFC3339Nano),
	})
}

func stableWatchKey(repoID, branch, provider string) string {
	replacer := strings.NewReplacer("/", "-", " ", "-", "_", "-", ".", "-")
	value := strings.ToLower(provider + "-" + repoID + "-" + branch)
	value = replacer.Replace(value)
	value = strings.Trim(value, "-")
	if value == "" {
		return newID()
	}
	return value
}

func newID() string {
	b := make([]byte, 8)
	if _, err := rand.Read(b); err != nil {
		return fmt.Sprintf("%d", time.Now().UnixNano())
	}
	return hex.EncodeToString(b)
}

type ScanRevisionOptions struct {
	Provider   string
	RepoID     string
	Branch     string
	Workdir    string
	Runner     gitengine.CommandRunner
	ObservedAt time.Time
}

func (r *Runtime) ScanBranchRevision(ctx context.Context, opts ScanRevisionOptions) (EventRecord, bool, error) {
	if r.store == nil || r.store.RevisionCursors() == nil {
		return EventRecord{}, false, fmt.Errorf("revision cursor store not supported")
	}

	repoID := strings.TrimSpace(opts.RepoID)
	branch := strings.TrimSpace(opts.Branch)
	workdir := strings.TrimSpace(opts.Workdir)
	if repoID == "" || branch == "" || workdir == "" {
		return EventRecord{}, false, fmt.Errorf("repo_id, branch, and workdir are required")
	}

	provider := strings.TrimSpace(opts.Provider)
	if provider == "" {
		provider = "forgejo"
	}

	runner := opts.Runner
	if runner == nil {
		runner = gitengine.CLI{}
	}

	after, err := gitengine.HeadRevision(runner, workdir)
	if err != nil {
		return EventRecord{}, false, fmt.Errorf("head revision: %w", err)
	}

	observedAt := opts.ObservedAt
	if observedAt.IsZero() {
		observedAt = time.Now().UTC()
	}

	cursor, found, err := r.store.RevisionCursors().GetRevisionCursor(ctx, repoID, branch)
	if err != nil {
		return EventRecord{}, false, fmt.Errorf("get revision cursor: %w", err)
	}

	if !found {
		newCursor := core.RevisionCursor{
			RepoID:     repoID,
			Branch:     branch,
			Revision:   after,
			ObservedAt: observedAt,
		}
		if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, newCursor); err != nil {
			return EventRecord{}, false, fmt.Errorf("upsert revision cursor: %w", err)
		}
		return EventRecord{}, false, nil
	}

	if cursor.Revision == after {
		cursor.ObservedAt = observedAt
		if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
			return EventRecord{}, false, fmt.Errorf("upsert revision cursor: %w", err)
		}
		return EventRecord{}, false, nil
	}

	before := cursor.Revision
	files, err := gitengine.ChangedFilesWithStatus(runner, workdir, before, after)
	if err != nil {
		return EventRecord{}, false, fmt.Errorf("changed files with status: %w", err)
	}

	revEvent := revisionEventFromScan(repoID, branch, before, after, files, observedAt)
	record, matched, err := r.HandleRevision(ctx, provider, "", revEvent)
	if err != nil {
		return record, matched, err
	}
	if !matched {
		cursor := core.RevisionCursor{
			RepoID:     repoID,
			Branch:     branch,
			Revision:   after,
			ObservedAt: observedAt,
		}
		if err := r.store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
			return record, false, fmt.Errorf("upsert revision cursor: %w", err)
		}
	}
	return record, matched, nil
}

func revisionEventFromScan(repoID, branch, before, after string, files []gitengine.ChangedFile, observedAt time.Time) core.RevisionEvent {
	changedFiles := make([]core.ChangedFile, 0, len(files))
	for _, f := range files {
		changedFiles = append(changedFiles, core.ChangedFile{
			Path:       f.Path,
			ChangeType: f.ChangeType,
		})
	}
	return core.RevisionEvent{
		RepoID:       repoID,
		Branch:       branch,
		Before:       before,
		After:        after,
		ChangedFiles: changedFiles,
		ObservedAt:   observedAt,
	}
}

func (r *Runtime) ScanAgentRunRevision(ctx context.Context, req worker.RevisionScanRequest) (worker.RevisionScanResult, error) {
	opts := ScanRevisionOptions{
		Provider:   req.Provider,
		RepoID:     req.RepoID,
		Branch:     req.Branch,
		Workdir:    req.WorkDir,
		Runner:     req.Runner,
		ObservedAt: req.ObservedAt,
	}
	record, matched, err := r.ScanBranchRevision(ctx, opts)
	if err != nil {
		return worker.RevisionScanResult{}, err
	}

	if matched && record.ID != "" && r.broadcaster == nil {
		return worker.RevisionScanResult{}, fmt.Errorf("event matched but broadcaster is nil (not published)")
	}

	return worker.RevisionScanResult{
		Attempted: true,
		Matched:   matched,
		EventID:   record.ID,
	}, nil
}

type outboxBroadcaster struct {
	store storage.OperationEventStore
}

func NewOutboxBroadcaster(store storage.OperationEventStore) EventBroadcaster {
	return &outboxBroadcaster{store: store}
}

func (o *outboxBroadcaster) BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error {
	if env.Channel != "event" || env.Action != "branch.updated" {
		return nil
	}

	payloadBytes, err := json.Marshal(env.Payload)
	if err != nil {
		return fmt.Errorf("marshal envelope payload: %w", err)
	}

	var m map[string]any
	if err := json.Unmarshal(payloadBytes, &m); err != nil {
		return fmt.Errorf("unmarshal envelope payload map: %w", err)
	}

	eventID, _ := m["id"].(string)
	if eventID == "" {
		eventID = fmt.Sprintf("branch-updated-outbox:%d", time.Now().UnixNano())
	}

	repoID, _ := m["repo_id"].(string)
	subject := "repo:" + repoID

	opEvent := storage.OperationEvent{
		OperationID: "",
		Event: events.Event{
			ID:        eventID,
			Type:      events.BranchUpdated,
			Subject:   subject,
			Payload:   payloadBytes,
			CreatedAt: time.Now().UTC(),
		},
	}

	if err := o.store.AppendEvent(ctx, opEvent); err != nil {
		return fmt.Errorf("append branch.updated outbox event: %w", err)
	}
	return nil
}