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

package controlplane

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"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/provider"
	"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
}

// WebhookSecretResolver is used to resolve a secret reference to its actual value.
type WebhookSecretResolver interface {
	ResolveWebhookSecret(ctx context.Context, secretRef string) (string, bool, error)
}

type Runtime struct {
	mu             sync.Mutex
	watches        map[string]BranchWatch
	subscriptions  map[string]WebhookSubscription
	records        []EventRecord
	broadcaster    EventBroadcaster
	store          storage.Store
	webhookClient  *http.Client
	secretResolver WebhookSecretResolver
}

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"`
	Webhook    *ProviderWebhookRecord `json:"webhook,omitempty"`
	CreatedAt  time.Time              `json:"created_at"`
}

type ProviderWebhookRecord struct {
	EventType   string
	ExternalID  string
	PayloadSize int
	ReceivedAt  time.Time
}

type ProviderPollRevision struct {
	Provider     string
	RepoID       string
	Branch       string
	Before       string
	After        string
	ChangedFiles []core.ChangedFile
	ObservedAt   time.Time
}

type ReconcileStatus string

const (
	ReconcileStatusNoop      ReconcileStatus = "noop"
	ReconcileStatusConflict  ReconcileStatus = "conflict"
	ReconcileStatusProposal  ReconcileStatus = "proposal"
	ReconcileStatusApplied   ReconcileStatus = "applied"
	ReconcileStatusDuplicate ReconcileStatus = "duplicate"
)

type ReconcileRequest struct {
	Provider         string
	RepoID           string
	Branch           string
	ExpectedRevision string
	CurrentRevision  string
	TargetRevision   string
	ChangedFiles     []core.ChangedFile
	IdempotencyKey   string
	ObservedAt       time.Time
}

type ReconcileProposal struct {
	Before  string             `json:"before"`
	After   string             `json:"after"`
	Changed []core.ChangedFile `json:"changed,omitempty"`
}

type ReconcileResult struct {
	Status         ReconcileStatus    `json:"status"`
	Conflict       bool               `json:"conflict,omitempty"`
	Applied        bool               `json:"applied,omitempty"`
	Duplicate      bool               `json:"duplicate,omitempty"`
	Drift          bool               `json:"drift,omitempty"`
	Reason         string             `json:"reason,omitempty"`
	Proposal       *ReconcileProposal `json:"proposal,omitempty"`
	IdempotencyKey string             `json:"idempotency_key,omitempty"`
}

type WebhookSubscription struct {
	ID        string
	Name      string
	TargetURL string
	Events    []string
	RepoID    string
	Branch    string
	SecretRef string
	CreatedAt time.Time
}

func NewRuntime(broadcaster EventBroadcaster) *Runtime {
	return &Runtime{
		watches:       make(map[string]BranchWatch),
		subscriptions: make(map[string]WebhookSubscription),
		broadcaster:   broadcaster,
		webhookClient: &http.Client{
			Timeout: 10 * time.Second,
		},
	}
}

// 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),
		subscriptions: make(map[string]WebhookSubscription),
		broadcaster:   broadcaster,
		store:         store,
		webhookClient: &http.Client{
			Timeout: 10 * time.Second,
		},
	}
}

// WithWebhookClient sets a custom HTTP client for webhook delivery.
func (r *Runtime) WithWebhookClient(client *http.Client) {
	r.webhookClient = client
}

// WithSecretResolver sets a custom secret resolver for webhook signatures.
func (r *Runtime) WithSecretResolver(resolver WebhookSecretResolver) {
	r.secretResolver = resolver
}

// matchingWebhookSubscriptions returns subscriptions that match the given event record.
func (r *Runtime) matchingWebhookSubscriptions(record EventRecord) []WebhookSubscription {
	r.mu.Lock()
	defer r.mu.Unlock()

	var matched []WebhookSubscription
	for _, sub := range r.subscriptions {
		// Check event type match
		eventMatched := false
		for _, event := range sub.Events {
			if event == record.Type {
				eventMatched = true
				break
			}
		}
		if !eventMatched {
			continue
		}

		// Check repo_id match: empty means match all
		if sub.RepoID != "" && sub.RepoID != record.Revision.RepoID {
			continue
		}

		// Check branch match: empty means match all
		if sub.Branch != "" && sub.Branch != record.Revision.Branch {
			continue
		}

		matched = append(matched, sub)
	}
	return matched
}

// buildWebhookPayload builds the HTTP webhook payload from an EventRecord.
// The payload contains only base fields required by any consumer (including NomadCode).
func (r *Runtime) buildWebhookPayload(record EventRecord) (map[string]any, error) {
	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,
		})
	}

	payload := map[string]any{
		"id":            record.ID,
		"type":          record.Type,
		"provider":      record.Provider,
		"delivery_id":   consumerDeliveryID(record),
		"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),
	}
	return payload, nil
}

// signPayload computes HMAC-SHA256 signature over the payload JSON body.
// It returns the hex-encoded HMAC digest.
func signPayload(secret string, body []byte) string {
	hmacSig := hmac.New(sha256.New, []byte(secret))
	hmacSig.Write(body)
	return hex.EncodeToString(hmacSig.Sum(nil))
}

// deliverWebhooks sends HTTP POST to matching webhook subscriptions.
func (r *Runtime) deliverWebhooks(ctx context.Context, record EventRecord) error {
	subscriptions := r.matchingWebhookSubscriptions(record)
	if len(subscriptions) == 0 {
		return nil
	}

	for _, sub := range subscriptions {
		if r.store != nil && r.store.WebhookDeliveries() != nil {
			delivery := core.WebhookDelivery{
				ID:             newID(),
				EventID:        record.ID,
				SubscriptionID: sub.ID,
				Status:         core.WebhookDeliveryPending,
				NextAttemptAt:  time.Now().UTC(),
			}
			res, err := r.store.WebhookDeliveries().EnqueueDelivery(ctx, delivery)
			if err != nil {
				return fmt.Errorf("enqueue delivery: %w", err)
			}
			if res.Created {
				started, ok, err := r.store.WebhookDeliveries().StartDelivery(ctx, res.Delivery.ID, time.Now().UTC())
				if err != nil {
					return fmt.Errorf("start delivery: %w", err)
				}
				if ok {
					if err := r.dispatchAndRecordOutcome(ctx, started, sub, record); err != nil {
						return err
					}
				}
			}
		} else {
			if err := r.postWebhook(ctx, sub, record); err != nil {
				return fmt.Errorf("webhook delivery to %q: %w", sub.Name, err)
			}
		}
	}
	return nil
}

// postWebhook sends a single HTTP POST to the subscription target URL.
func (r *Runtime) postWebhook(ctx context.Context, sub WebhookSubscription, record EventRecord) error {
	resp, err := r.executeWebhookRequest(ctx, sub, record)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	return nil
}

func (r *Runtime) executeWebhookRequest(ctx context.Context, sub WebhookSubscription, record EventRecord) (*http.Response, error) {
	payload, err := r.buildWebhookPayload(record)
	if err != nil {
		return nil, fmt.Errorf("build payload: %w", err)
	}

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

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, sub.TargetURL, bytes.NewReader(bodyBytes))
	if err != nil {
		return nil, fmt.Errorf("create request: %w", err)
	}

	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-Gito-Event", record.Type)
	req.Header.Set("X-Gito-Delivery", consumerDeliveryID(record))

	if sub.SecretRef != "" {
		if r.secretResolver == nil {
			return nil, fmt.Errorf("secret_ref %q is configured but no secret resolver is available", sub.SecretRef)
		}
		secret, resolved, err := r.secretResolver.ResolveWebhookSecret(ctx, sub.SecretRef)
		if err != nil {
			return nil, fmt.Errorf("resolve secret for %q: %w", sub.SecretRef, err)
		}
		if !resolved {
			return nil, fmt.Errorf("secret_ref %q is unresolved", sub.SecretRef)
		}
		signature := signPayload(secret, bodyBytes)
		req.Header.Set("X-Gito-Signature", "sha256="+signature)
	}

	return r.webhookClient.Do(req)
}

func (r *Runtime) dispatchAndRecordOutcome(ctx context.Context, d core.WebhookDelivery, sub WebhookSubscription, record EventRecord) error {
	resp, err := r.executeWebhookRequest(ctx, sub, record)
	if resp != nil {
		defer resp.Body.Close()
	}

	outcome := classifyWebhookDelivery(resp, err, d.AttemptCount, time.Now().UTC())

	updated, recErr := r.store.WebhookDeliveries().RecordOutcome(
		ctx,
		d.ID,
		outcome.status,
		outcome.statusCode,
		outcome.lastErr,
		outcome.nextAttemptAt,
		time.Now().UTC(),
	)
	if recErr != nil {
		return fmt.Errorf("record webhook delivery outcome %s: %w", d.ID, recErr)
	}
	_ = updated
	return nil
}

func (r *Runtime) ProcessPendingWebhooks(ctx context.Context) error {
	if r.store == nil || r.store.WebhookDeliveries() == nil {
		return nil
	}

	picked, err := r.store.WebhookDeliveries().PickPendingDeliveries(ctx, 10, time.Now().UTC())
	if err != nil {
		return fmt.Errorf("pick pending: %w", err)
	}

	for _, d := range picked {
		var targetSub WebhookSubscription
		r.mu.Lock()
		sub, ok := r.subscriptions[d.SubscriptionID]
		r.mu.Unlock()
		if !ok {
			_, recErr := r.store.WebhookDeliveries().RecordOutcome(ctx, d.ID, core.WebhookDeliveryFailed, 0, "subscription not found", time.Time{}, time.Now().UTC())
			if recErr != nil {
				return fmt.Errorf("record outcome for missing subscription %s: %w", d.ID, recErr)
			}
			continue
		}
		targetSub = sub

		var targetRecord EventRecord
		found := false
		r.mu.Lock()
		for _, rec := range r.records {
			if rec.ID == d.EventID {
				targetRecord = rec
				found = true
				break
			}
		}
		r.mu.Unlock()

		if !found {
			_, recErr := r.store.WebhookDeliveries().RecordOutcome(ctx, d.ID, core.WebhookDeliveryFailed, 0, "event record not found", time.Time{}, time.Now().UTC())
			if recErr != nil {
				return fmt.Errorf("record outcome for missing event %s: %w", d.ID, recErr)
			}
			continue
		}

		if err := r.dispatchAndRecordOutcome(ctx, d, targetSub, targetRecord); err != nil {
			return err
		}
	}
	return nil
}

type webhookDeliveryOutcome struct {
	status        core.WebhookDeliveryStatus
	lastErr       string
	statusCode    int
	nextAttemptAt time.Time
}

func classifyWebhookDelivery(resp *http.Response, err error, attempt int, now time.Time) webhookDeliveryOutcome {
	if err != nil {
		return retryableOutcome(now, attempt, 0, err.Error())
	}

	sc := resp.StatusCode
	if sc >= 200 && sc < 300 {
		return succeededOutcome(sc)
	}

	if sc == http.StatusTooManyRequests || sc >= 500 {
		return retryableOutcome(now, attempt, sc, fmt.Sprintf("unexpected status code: %d", sc))
	}

	return failedOutcome(sc, fmt.Sprintf("terminal status code: %d", sc))
}

func succeededOutcome(sc int) webhookDeliveryOutcome {
	return webhookDeliveryOutcome{
		status:     core.WebhookDeliverySucceeded,
		statusCode: sc,
	}
}

func failedOutcome(sc int, errStr string) webhookDeliveryOutcome {
	return webhookDeliveryOutcome{
		status:     core.WebhookDeliveryFailed,
		statusCode: sc,
		lastErr:    errStr,
	}
}

func retryableOutcome(now time.Time, attempt int, sc int, errStr string) webhookDeliveryOutcome {
	backoff := time.Duration(1<<attempt) * time.Second
	if backoff > 60*time.Second {
		backoff = 60 * time.Second
	}

	if attempt >= 5 {
		return failedOutcome(sc, fmt.Sprintf("retry budget exceeded: %s", errStr))
	}

	return webhookDeliveryOutcome{
		status:        core.WebhookDeliveryRetryable,
		statusCode:    sc,
		lastErr:       errStr,
		nextAttemptAt: now.Add(backoff),
	}
}

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) RegisterWebhookSubscription(name, targetURL string, events []string, repoID, branch, secretRef string) (WebhookSubscription, error) {
	name = strings.TrimSpace(name)
	targetURL = strings.TrimSpace(targetURL)
	repoID = strings.TrimSpace(repoID)
	branch = strings.TrimSpace(branch)
	secretRef = strings.TrimSpace(secretRef)
	events = normalizeEventNames(events)
	if targetURL == "" {
		return WebhookSubscription{}, fmt.Errorf("target_url is required")
	}
	parsedTarget, err := url.Parse(targetURL)
	if err != nil || parsedTarget.Host == "" || (parsedTarget.Scheme != "http" && parsedTarget.Scheme != "https") {
		return WebhookSubscription{}, fmt.Errorf("target_url must be an http or https URL")
	}
	if len(events) == 0 {
		return WebhookSubscription{}, fmt.Errorf("events are required")
	}
	if name == "" {
		name = "webhook-subscription"
	}

	subscription := WebhookSubscription{
		ID:        "webhook-sub-" + newID(),
		Name:      name,
		TargetURL: targetURL,
		Events:    events,
		RepoID:    repoID,
		Branch:    branch,
		SecretRef: secretRef,
		CreatedAt: time.Now().UTC(),
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.subscriptions[subscription.ID] = subscription
	return subscription, nil
}

func (r *Runtime) ListWebhookSubscriptions() []WebhookSubscription {
	r.mu.Lock()
	defer r.mu.Unlock()
	subscriptions := make([]WebhookSubscription, 0, len(r.subscriptions))
	for _, subscription := range r.subscriptions {
		subscriptions = append(subscriptions, subscription)
	}
	sort.Slice(subscriptions, func(i, j int) bool {
		return subscriptions[i].ID < subscriptions[j].ID
	})
	return subscriptions
}

func normalizeEventNames(events []string) []string {
	seen := make(map[string]struct{}, len(events))
	result := make([]string, 0, len(events))
	for _, event := range events {
		event = strings.TrimSpace(event)
		if event == "" {
			continue
		}
		if _, ok := seen[event]; ok {
			continue
		}
		seen[event] = struct{}{}
		result = append(result, event)
	}
	return result
}

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
	}

	// Deliver webhooks for matched event
	if err := r.deliverWebhooks(ctx, record); err != nil {
		return record, true, err
	}

	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) HandleProviderWebhook(ctx context.Context, input provider.WebhookEvent) (EventRecord, error) {
	providerName := strings.TrimSpace(input.Provider)
	if providerName == "" {
		return EventRecord{}, fmt.Errorf("provider is required")
	}
	eventType := strings.TrimSpace(input.EventType)
	if eventType == "" {
		eventType = "unknown"
	}
	externalID := strings.TrimSpace(input.ExternalID)
	receivedAt := time.Now().UTC().Truncate(time.Microsecond)
	record := EventRecord{
		ID:         "event-" + newID(),
		Type:       events.ProviderWebhook,
		Provider:   providerName,
		DeliveryID: externalID,
		Webhook: &ProviderWebhookRecord{
			EventType:   eventType,
			ExternalID:  externalID,
			PayloadSize: len(input.Payload),
			ReceivedAt:  receivedAt,
		},
		CreatedAt: receivedAt,
	}

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

	if r.broadcaster != nil {
		if err := r.broadcaster.BroadcastEnvelope(ctx, providerWebhookEnvelope(record)); err != nil {
			return record, err
		}
	}
	return record, nil
}

func (r *Runtime) HandleProviderPollRevision(ctx context.Context, poll ProviderPollRevision) (EventRecord, bool, error) {
	provider := strings.TrimSpace(poll.Provider)
	if provider == "" {
		provider = "forgejo"
	}
	repoID := strings.TrimSpace(poll.RepoID)
	branch := strings.TrimSpace(poll.Branch)
	before := strings.TrimSpace(poll.Before)
	after := strings.TrimSpace(poll.After)
	if repoID == "" || branch == "" || before == "" || after == "" {
		return EventRecord{}, false, fmt.Errorf("repo_id, branch, before, and after are required")
	}
	observedAt := poll.ObservedAt
	if observedAt.IsZero() {
		observedAt = time.Now().UTC()
	}
	revision := core.RevisionEvent{
		RepoID:       repoID,
		Branch:       branch,
		Before:       before,
		After:        after,
		ChangedFiles: append([]core.ChangedFile(nil), poll.ChangedFiles...),
		ObservedAt:   observedAt,
	}
	return r.HandleRevision(ctx, provider, "", revision)
}

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()

	// Deliver webhooks for matched event
	if err := r.deliverWebhooks(ctx, record); err != nil {
		// Rollback delivery record on webhook delivery failure so retry is allowed.
		if r.store.ProviderDeliveries() != nil && dedupeKey != "" {
			_ = r.store.ProviderDeliveries().DeleteDelivery(ctx, provider, dedupeKey)
		}
		return record, true, err
	}

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

// consumerDeliveryID returns a stable non-empty delivery ID for the consumer.
// Priority: record.DeliveryID > deterministic revision-based fallback > "unknown".
// When record.DeliveryID (provider delivery id) is empty, the fallback ID is
// derived deterministically from revision identity so that the same retry
// produces the same consumer delivery id.
func consumerDeliveryID(record EventRecord) string {
	if record.DeliveryID != "" {
		return record.DeliveryID
	}
	// Deterministic fallback from revision identity so that
	// the same revision retry always produces the same
	// consumer-facing delivery id.
	if revFallback := consumerDeliveryIDFromRevision(record); revFallback != "" {
		return revFallback
	}
	return "unknown"
}

// consumerDeliveryIDFromRevision derives a stable non-empty delivery ID from
// the revision identity fields. It returns an empty string when repo/branch
// information is unavailable so that callers can still fall back to "unknown".
func consumerDeliveryIDFromRevision(record EventRecord) string {
	rev := record.Revision
	if rev.RepoID == "" && rev.Branch == "" {
		return ""
	}
	return fmt.Sprintf("fallback:%s:%s:%s:%s", rev.RepoID, rev.Branch, rev.Before, rev.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 providerWebhookEnvelope(record EventRecord) protosocket.Envelope {
	payload := map[string]any{
		"id":          record.ID,
		"type":        record.Type,
		"provider":    record.Provider,
		"delivery_id": record.DeliveryID,
		"created_at":  record.CreatedAt.UTC().Format(time.RFC3339Nano),
	}
	if record.Webhook != nil {
		payload["event_type"] = record.Webhook.EventType
		payload["payload_size"] = record.Webhook.PayloadSize
		payload["received_at"] = record.Webhook.ReceivedAt.UTC().Format(time.RFC3339Nano)
	}
	return protosocket.NewEventEnvelope(events.ProviderWebhook, payload)
}

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) {
	before := strings.TrimSpace(req.BeforeRevision)
	after := strings.TrimSpace(req.AfterRevision)

	if before != "" && after != "" {
		// Explicit pair: bypass cursor look-up and emit immediately if watched.
		// This covers cursor-free first push of a watched branch after agent_run.
		record, matched, err := r.scanExplicitRevisionPair(ctx, req.Provider, req.RepoID, req.Branch, before, after, req.WorkDir, req.Runner, req.ObservedAt)
		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
	}

	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
}

// scanExplicitRevisionPair emits a branch.updated event for a known before/after
// pair, regardless of cursor state. changed_files are computed from workdir using
// runner. On publish success the cursor advances to after; on failure the cursor
// is not advanced, preserving retry semantics.
func (r *Runtime) scanExplicitRevisionPair(ctx context.Context, provider, repoID, branch, before, after, workdir string, runner gitengine.CommandRunner, observedAt time.Time) (EventRecord, bool, error) {
	provider = strings.TrimSpace(provider)
	if provider == "" {
		provider = "forgejo"
	}
	if strings.TrimSpace(workdir) == "" {
		return EventRecord{}, false, fmt.Errorf("workdir is required for explicit revision pair scan")
	}
	if runner == nil {
		runner = gitengine.CLI{}
	}
	if observedAt.IsZero() {
		observedAt = time.Now().UTC()
	}

	changedFiles, err := gitengine.ChangedFilesWithStatus(runner, workdir, before, after)
	if err != nil {
		return EventRecord{}, false, fmt.Errorf("changed files for explicit revision pair: %w", err)
	}

	revEvent := revisionEventFromScan(repoID, branch, before, after, changedFiles, observedAt)

	record, matched, err := r.HandleRevision(ctx, provider, "", revEvent)
	if err != nil {
		return record, matched, err
	}

	return record, matched, nil
}

type outboxBroadcaster struct {
	store storage.OperationEventStore
}

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

// DryRunReconcile validates the reconcile request and returns a proposal without
// applying any changes. It does not append events, create deliveries, or broadcast.
func (r *Runtime) DryRunReconcile(ctx context.Context, req ReconcileRequest) ReconcileResult {
	req.Provider = strings.TrimSpace(req.Provider)
	if req.Provider == "" {
		req.Provider = "forgejo"
	}
	req.RepoID = strings.TrimSpace(req.RepoID)
	req.Branch = strings.TrimSpace(req.Branch)
	if req.RepoID == "" || req.Branch == "" {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Reason:   "repo_id and branch are required",
		}
	}

	// Check expected revision mismatch first
	if req.ExpectedRevision != "" && req.CurrentRevision != "" && req.ExpectedRevision != req.CurrentRevision {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Applied:  false,
			Reason:   fmt.Sprintf("expected revision %q does not match current revision %q", req.ExpectedRevision, req.CurrentRevision),
		}
	}

	// No drift when current matches target
	if req.CurrentRevision == req.TargetRevision {
		return ReconcileResult{
			Status:  ReconcileStatusNoop,
			Drift:   false,
			Applied: false,
			Reason:  "current revision already matches target",
		}
	}

	// Drift proposal: current differs from target
	return ReconcileResult{
		Status:  ReconcileStatusProposal,
		Drift:   true,
		Applied: false,
		Proposal: &ReconcileProposal{
			Before:  req.CurrentRevision,
			After:   req.TargetRevision,
			Changed: req.ChangedFiles,
		},
		Reason: "drift detected: current differs from target",
	}
}

// ApplyReconcile validates the reconcile request, runs dry-run, and applies the
// change using HandleRevision with an idempotency key. Side effects (webhook
// delivery, broadcast) happen only on first apply; duplicate keys return
// ReconcileStatusDuplicate without additional side effects.
// Apply idempotency is delegated to HandleRevision's delivery dedupe path, so a
// failure during HandleRevision rolls back the dedupe record and allows retry.
// IdempotencyKey and durable ProviderDeliveries store are required; missing
// either returns conflict without side effects.
func (r *Runtime) ApplyReconcile(ctx context.Context, req ReconcileRequest) ReconcileResult {
	req.Provider = strings.TrimSpace(req.Provider)
	if req.Provider == "" {
		req.Provider = "forgejo"
	}
	req.RepoID = strings.TrimSpace(req.RepoID)
	req.Branch = strings.TrimSpace(req.Branch)
	if req.RepoID == "" || req.Branch == "" {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Reason:   "repo_id and branch are required",
		}
	}

	// Dry-run first: conflict blocks apply
	dryResult := r.DryRunReconcile(ctx, req)
	if dryResult.Conflict {
		return dryResult
	}

	// Idempotency key is required for apply
	if req.IdempotencyKey == "" {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Reason:   "idempotency_key is required for apply",
		}
	}

	// Durable dedupe store is required for apply
	if r.store == nil || r.store.ProviderDeliveries() == nil {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Reason:   "durable ProviderDeliveries store is required for apply",
		}
	}

	// Apply via HandleRevision. The existing delivery dedupe path inside
	// HandleRevision acts as both the delivery dedupe and the apply idempotency
	// key. When HandleRevision fails it rolls back its delivery dedupe record,
	// so the same idempotency key can be retried successfully.
	revision := core.RevisionEvent{
		RepoID:       req.RepoID,
		Branch:       req.Branch,
		Before:       req.CurrentRevision,
		After:        req.TargetRevision,
		ChangedFiles: req.ChangedFiles,
		ObservedAt:   req.ObservedAt,
	}
	record, matched, err := r.HandleRevision(ctx, req.Provider, "reconcile:"+req.IdempotencyKey, revision)
	if err != nil {
		return ReconcileResult{
			Status:   ReconcileStatusConflict,
			Conflict: true,
			Reason:   fmt.Sprintf("handle revision: %v", err),
		}
	}
	if record.Duplicate {
		// HandleRevision returned an idempotent duplicate result. This means
		// the previous invocation with the same delivery id already completed
		// side effects (webhook delivery, broadcast, cursor upsert).
		return ReconcileResult{
			Status:         ReconcileStatusDuplicate,
			Duplicate:      true,
			Applied:        false,
			IdempotencyKey: req.IdempotencyKey,
			Reason:         "duplicate reconcile apply by idempotency key",
		}
	}
	if !matched {
		return ReconcileResult{
			Status:         ReconcileStatusApplied,
			Applied:        true,
			IdempotencyKey: req.IdempotencyKey,
			Reason:         "applied but event did not match any watch",
		}
	}
	return ReconcileResult{
		Status:         ReconcileStatusApplied,
		Applied:        true,
		Duplicate:      false,
		IdempotencyKey: req.IdempotencyKey,
		Reason:         "reconcile apply succeeded",
	}
}

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
}