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

package controlplane

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.toki-labs.com/toki/gito/services/core/internal/config"
	"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"
)

// fakeStore is a shared in-memory implementation of storage.Store for tests.
type fakeStore struct {
	watches           *fakeBranchWatchStore
	cursors           *fakeRevisionCursorStore
	deliveries        *fakeProviderDeliveryStore
	webhookDeliveries *fakeWebhookDeliveryStore
	providerConfigs   storage.ProviderConfigStore
}

func newFakeStore() *fakeStore {
	return &fakeStore{
		watches:    &fakeBranchWatchStore{},
		cursors:    &fakeRevisionCursorStore{},
		deliveries: &fakeProviderDeliveryStore{records: make(map[string]core.ProviderDelivery)},
	}
}

func (f *fakeStore) Ping(_ context.Context) error                      { return nil }
func (f *fakeStore) Repos() storage.RepoStore                          { return nil }
func (f *fakeStore) WorkspaceLeases() storage.WorkspaceLeaseStore      { return nil }
func (f *fakeStore) Operations() storage.OperationStore                { return nil }
func (f *fakeStore) OperationEvents() storage.OperationEventStore      { return nil }
func (f *fakeStore) BranchWatches() storage.BranchWatchStore           { return f.watches }
func (f *fakeStore) RevisionCursors() storage.RevisionCursorStore      { return f.cursors }
func (f *fakeStore) ProviderDeliveries() storage.ProviderDeliveryStore { return f.deliveries }
func (f *fakeStore) AgentRunInputs() storage.AgentRunInputStore        { return nil }
func (f *fakeStore) WebhookDeliveries() storage.WebhookDeliveryStore {
	if f.webhookDeliveries == nil {
		return nil
	}
	return f.webhookDeliveries
}

func (f *fakeStore) ProviderConfigs() storage.ProviderConfigStore {
	return f.providerConfigs
}

var _ storage.Store = (*fakeStore)(nil)

// fakeBranchWatchStore

type fakeBranchWatchStore struct {
	mu      sync.Mutex
	watches []core.BranchWatch
}

func (f *fakeBranchWatchStore) UpsertBranchWatch(_ context.Context, watch core.BranchWatch) (core.BranchWatch, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, w := range f.watches {
		if w.Provider == watch.Provider && w.RepoID == watch.RepoID && w.Branch == watch.Branch {
			return w, nil
		}
	}
	f.watches = append(f.watches, watch)
	return watch, nil
}

func (f *fakeBranchWatchStore) FindBranchWatch(_ context.Context, provider, repoID, branch string) (core.BranchWatch, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, w := range f.watches {
		if w.Provider == provider && w.RepoID == repoID && w.Branch == branch {
			return w, true, nil
		}
	}
	return core.BranchWatch{}, false, nil
}

func (f *fakeBranchWatchStore) ListBranchWatches(_ context.Context) ([]core.BranchWatch, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]core.BranchWatch(nil), f.watches...), nil
}

// fakeRevisionCursorStore

type fakeRevisionCursorStore struct {
	mu      sync.Mutex
	cursors map[string]core.RevisionCursor
}

func (f *fakeRevisionCursorStore) UpsertRevisionCursor(_ context.Context, cursor core.RevisionCursor) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.cursors == nil {
		f.cursors = make(map[string]core.RevisionCursor)
	}
	f.cursors[cursor.RepoID+":"+cursor.Branch] = cursor
	return nil
}

func (f *fakeRevisionCursorStore) GetRevisionCursor(_ context.Context, repoID, branch string) (core.RevisionCursor, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	c, ok := f.cursors[repoID+":"+branch]
	return c, ok, nil
}

// fakeProviderDeliveryStore

type fakeProviderDeliveryStore struct {
	mu      sync.Mutex
	records map[string]core.ProviderDelivery
}

func (f *fakeProviderDeliveryStore) RecordOnce(_ context.Context, delivery core.ProviderDelivery) (storage.DeliveryResult, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if existing, ok := f.records[delivery.DedupeKey]; ok {
		return storage.DeliveryResult{
			First:             false,
			ExistingEventID:   existing.EventID,
			ExistingCreatedAt: existing.CreatedAt.Truncate(time.Microsecond),
		}, nil
	}
	f.records[delivery.DedupeKey] = delivery
	return storage.DeliveryResult{First: true}, nil
}

func (f *fakeProviderDeliveryStore) DeleteDelivery(_ context.Context, provider, dedupeKey string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	delete(f.records, dedupeKey)
	return nil
}

// fakeWebhookDeliveryStore

type fakeWebhookDeliveryStore struct {
	mu         sync.Mutex
	records    map[string]core.WebhookDelivery
	outcomeErr error
}

func (f *fakeWebhookDeliveryStore) SetOutcomeErr(err error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.outcomeErr = err
}

func (f *fakeWebhookDeliveryStore) EnqueueDelivery(_ context.Context, delivery core.WebhookDelivery) (storage.WebhookDeliveryCreateResult, error) {
	f.mu.Lock()
	defer f.mu.Unlock()

	if delivery.ID == "" {
		delivery.ID = "fake-delivery-id-" + fmt.Sprint(len(f.records))
	}
	if delivery.Status == "" {
		delivery.Status = core.WebhookDeliveryPending
	}

	key := delivery.EventID + ":" + delivery.SubscriptionID
	if existing, ok := f.records[key]; ok {
		return storage.WebhookDeliveryCreateResult{Delivery: existing, Created: false}, nil
	}

	now := time.Now().UTC()
	if delivery.CreatedAt.IsZero() {
		delivery.CreatedAt = now
	}
	if delivery.UpdatedAt.IsZero() {
		delivery.UpdatedAt = now
	}
	if delivery.NextAttemptAt.IsZero() {
		delivery.NextAttemptAt = now
	}
	f.records[key] = delivery
	return storage.WebhookDeliveryCreateResult{Delivery: delivery, Created: true}, nil
}

func (f *fakeWebhookDeliveryStore) GetDelivery(_ context.Context, id string) (core.WebhookDelivery, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for _, r := range f.records {
		if r.ID == id {
			return r, nil
		}
	}
	return core.WebhookDelivery{}, fmt.Errorf("not found")
}

func (f *fakeWebhookDeliveryStore) PickPendingDeliveries(_ context.Context, limit int, now time.Time) ([]core.WebhookDelivery, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var list []core.WebhookDelivery
	for key, d := range f.records {
		if (d.Status == core.WebhookDeliveryPending || d.Status == core.WebhookDeliveryRetryable) && (d.NextAttemptAt.Before(now) || d.NextAttemptAt.Equal(now)) {
			d.Status = core.WebhookDeliverySending
			d.UpdatedAt = now
			f.records[key] = d
			list = append(list, d)
			if len(list) >= limit {
				break
			}
		}
	}
	return list, nil
}

func (f *fakeWebhookDeliveryStore) StartDelivery(_ context.Context, id string, now time.Time) (core.WebhookDelivery, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for key, d := range f.records {
		if d.ID == id && (d.Status == core.WebhookDeliveryPending || d.Status == core.WebhookDeliveryRetryable) {
			d.Status = core.WebhookDeliverySending
			d.UpdatedAt = now
			f.records[key] = d
			return d, true, nil
		}
	}
	return core.WebhookDelivery{}, false, nil
}

func (f *fakeWebhookDeliveryStore) RecordOutcome(_ context.Context, id string, status core.WebhookDeliveryStatus, statusCode int, lastErr string, nextAttemptAt time.Time, now time.Time) (core.WebhookDelivery, error) {
	f.mu.Lock()
	if f.outcomeErr != nil {
		err := f.outcomeErr
		f.mu.Unlock()
		return core.WebhookDelivery{}, err
	}
	f.mu.Unlock()
	for key, d := range f.records {
		if d.ID == id {
			d.Status = status
			d.LastStatusCode = statusCode
			d.LastError = lastErr
			d.NextAttemptAt = nextAttemptAt
			d.AttemptCount++
			d.UpdatedAt = now
			f.records[key] = d
			return d, nil
		}
	}
	return core.WebhookDelivery{}, fmt.Errorf("not found")
}

// Tests

func TestRuntimeRestartPersistsWatchAndCursor(t *testing.T) {
	store := newFakeStore()

	// Runtime A: register a watch
	runtimeA := NewRuntimeWithStore(nil, store)
	_, err := runtimeA.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Runtime B: same store, simulates process restart
	runtimeB := NewRuntimeWithStore(nil, store)

	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa",
		After:      "bbb",
		ObservedAt: time.Now().UTC(),
	}
	record, matched, err := runtimeB.HandleRevision(context.Background(), "forgejo", "delivery-restart-1", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatalf("expected matched=true after restart, got false; record=%+v", record)
	}
	if record.Duplicate {
		t.Fatalf("expected first delivery not to be duplicate")
	}

	// cursor must be persisted
	cursor, ok, err := store.cursors.GetRevisionCursor(context.Background(), "nomadcode", "develop")
	if err != nil {
		t.Fatalf("get cursor: %v", err)
	}
	if !ok {
		t.Fatal("expected revision cursor to be stored after first delivery")
	}
	if cursor.Revision != "bbb" {
		t.Fatalf("cursor revision: got %q want %q", cursor.Revision, "bbb")
	}
}

func TestRuntimeDuplicateDeliveryIsIdempotent(t *testing.T) {
	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)

	_, err := runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa",
		After:      "bbb",
		ObservedAt: time.Now().UTC(),
	}

	record1, matched1, err := runtime.HandleRevision(context.Background(), "forgejo", "delivery-dup-1", revision)
	if err != nil {
		t.Fatalf("first handle: %v", err)
	}
	if !matched1 || record1.Duplicate {
		t.Fatalf("first: matched=%v duplicate=%v", matched1, record1.Duplicate)
	}

	record2, matched2, err := runtime.HandleRevision(context.Background(), "forgejo", "delivery-dup-1", revision)
	if err != nil {
		t.Fatalf("second handle: %v", err)
	}
	if !matched2 {
		t.Fatalf("duplicate delivery should still return matched=true")
	}
	if !record2.Duplicate {
		t.Fatalf("expected second delivery to be marked as duplicate")
	}

	// duplicate response must return same event id and created_at as the first
	if record2.ID != record1.ID {
		t.Fatalf("duplicate event id: got %q want %q", record2.ID, record1.ID)
	}
	if !record2.CreatedAt.Equal(record1.CreatedAt) {
		t.Fatalf("duplicate created_at: got %v want %v", record2.CreatedAt, record1.CreatedAt)
	}

	// in-memory event log must have exactly one record
	evList := runtime.ListEvents()
	if len(evList) != 1 {
		t.Fatalf("event list count: got %d want 1", len(evList))
	}
	if evList[0].Type != events.BranchUpdated {
		t.Fatalf("event type: got %q", evList[0].Type)
	}
}

func TestForgejoPushDuplicateDeliveryIsIdempotent(t *testing.T) {
	store := newFakeStore()
	router := newRouterWithStore(config.Config{AppEnv: "test", ProtoSocketPath: "/proto-socket"}, nil, store)

	registerBranchListener(t, router, `{"repo_id":"nomadcode","branch":"develop","provider":"forgejo"}`)

	pushBody := `{
		"ref": "refs/heads/develop",
		"before": "111",
		"after": "222",
		"repository": {"name": "nomadcode", "full_name": "toki/nomadcode"},
		"commits": [{"id": "222", "modified": ["README.md"]}]
	}`

	sendPush := func(deliveryID string) map[string]any {
		req := httptest.NewRequest(http.MethodPost, "/callbacks/forgejo/push?repo_id=nomadcode", bytes.NewBufferString(pushBody))
		req.Header.Set("X-Forgejo-Event", "push")
		req.Header.Set("X-Forgejo-Delivery", deliveryID)
		rec := httptest.NewRecorder()
		router.ServeHTTP(rec, req)
		if rec.Code != http.StatusAccepted {
			t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String())
		}
		var body map[string]any
		if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
			t.Fatalf("decode: %v", err)
		}
		return body
	}

	first := sendPush("dup-delivery-1")
	if first["matched"] != true || first["duplicate"] == true {
		t.Fatalf("first push: matched=%v duplicate=%v", first["matched"], first["duplicate"])
	}

	second := sendPush("dup-delivery-1")
	if second["matched"] != true {
		t.Fatalf("second push: expected matched=true")
	}
	if second["duplicate"] != true {
		t.Fatalf("second push: expected duplicate=true, got %v", second["duplicate"])
	}

	// duplicate response event id and created_at must match the first delivery
	firstEvent, _ := first["event"].(map[string]any)
	secondEvent, _ := second["event"].(map[string]any)
	if firstEvent == nil || secondEvent == nil {
		t.Fatalf("event missing: first=%v second=%v", first["event"], second["event"])
	}
	if firstEvent["id"] != secondEvent["id"] {
		t.Fatalf("duplicate event id mismatch: first=%v second=%v", firstEvent["id"], secondEvent["id"])
	}
	if firstEvent["created_at"] != secondEvent["created_at"] {
		t.Fatalf("duplicate created_at mismatch: first=%v second=%v", firstEvent["created_at"], secondEvent["created_at"])
	}

	// /api/events count must not increase on duplicate
	evReq := httptest.NewRequest(http.MethodGet, "/api/events", nil)
	evRec := httptest.NewRecorder()
	router.ServeHTTP(evRec, evReq)
	var evBody map[string][]map[string]any
	if err := json.NewDecoder(evRec.Body).Decode(&evBody); err != nil {
		t.Fatalf("decode events: %v", err)
	}
	if len(evBody["events"]) != 1 {
		t.Fatalf("events count: got %d want 1", len(evBody["events"]))
	}
}

func setupGitRepo(t *testing.T) (workdir string, initialRev string) {
	tmpDir := t.TempDir()
	workdir = filepath.Join(tmpDir, "repo")
	if err := os.MkdirAll(workdir, 0755); err != nil {
		t.Fatalf("failed to create repo dir: %v", err)
	}
	runGit(t, workdir, "init")
	runGit(t, workdir, "config", "user.name", "Test User")
	runGit(t, workdir, "config", "user.email", "test@example.com")
	runGit(t, workdir, "config", "commit.gpgsign", "false")
	runGit(t, workdir, "branch", "-M", "main")

	writeFile(t, filepath.Join(workdir, "README.md"), "# Seed Repository\n")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "initial commit")

	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = workdir
	out, err := cmd.Output()
	if err != nil {
		t.Fatalf("failed to get head rev: %v", err)
	}
	initialRev = strings.TrimSpace(string(out))
	return workdir, initialRev
}

func runGit(t *testing.T, dir string, args ...string) {
	cmd := exec.Command("git", args...)
	cmd.Dir = dir
	if out, err := cmd.CombinedOutput(); err != nil {
		t.Fatalf("git %s failed: %v, output: %s", strings.Join(args, " "), err, string(out))
	}
}

func writeFile(t *testing.T, path, content string) {
	if err := os.WriteFile(path, []byte(content), 0644); err != nil {
		t.Fatalf("WriteFile %s failed: %v", path, err)
	}
}

func TestRuntimeScanBranchRevisionCursorStates(t *testing.T) {
	store := newFakeStore()
	r := NewRuntimeWithStore(nil, store)
	workdir, initialRev := setupGitRepo(t)

	opts := ScanRevisionOptions{
		RepoID:  "myrepo",
		Branch:  "main",
		Workdir: workdir,
		Runner:  gitengine.CLI{},
	}

	// 1. First observation: stores cursor, returns no event, matched=false
	record, matched, err := r.ScanBranchRevision(context.Background(), opts)
	if err != nil {
		t.Fatalf("first scan: %v", err)
	}
	if matched {
		t.Fatal("first scan: expected matched=false")
	}
	if record.ID != "" {
		t.Fatalf("first scan: expected empty record, got %+v", record)
	}

	cursor, ok, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil {
		t.Fatalf("get cursor: %v", err)
	}
	if !ok {
		t.Fatal("first scan: cursor not stored")
	}
	if cursor.Revision != initialRev {
		t.Fatalf("first scan cursor revision: got %q want %q", cursor.Revision, initialRev)
	}

	// 2. Unchanged revision: no event, updates observed time (or stays unchanged), matched=false
	record2, matched2, err := r.ScanBranchRevision(context.Background(), opts)
	if err != nil {
		t.Fatalf("second scan: %v", err)
	}
	if matched2 {
		t.Fatal("second scan: expected matched=false")
	}
	if record2.ID != "" {
		t.Fatalf("second scan: expected empty record, got %+v", record2)
	}

	cursor2, ok2, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil {
		t.Fatalf("get cursor2: %v", err)
	}
	if !ok2 {
		t.Fatal("second scan: cursor not found")
	}
	if cursor2.Revision != initialRev {
		t.Fatalf("second scan cursor revision: got %q want %q", cursor2.Revision, initialRev)
	}

	// 3. Changed revision with watch: stores cursor with new revision, returns event, matched=true
	_, err = r.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	writeFile(t, filepath.Join(workdir, "README.md"), "# Seed Repository\n\nupdated\n")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "update readme")

	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = workdir
	out, err := cmd.Output()
	if err != nil {
		t.Fatalf("get head: %v", err)
	}
	afterRev := strings.TrimSpace(string(out))

	record3, matched3, err := r.ScanBranchRevision(context.Background(), opts)
	if err != nil {
		t.Fatalf("third scan: %v", err)
	}
	if !matched3 {
		t.Fatal("third scan: expected matched=true")
	}
	if record3.Revision.Before != initialRev {
		t.Fatalf("third scan record before: got %q want %q", record3.Revision.Before, initialRev)
	}
	if record3.Revision.After != afterRev {
		t.Fatalf("third scan record after: got %q want %q", record3.Revision.After, afterRev)
	}

	cursor3, ok3, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil {
		t.Fatalf("get cursor3: %v", err)
	}
	if !ok3 {
		t.Fatal("third scan: cursor not found")
	}
	if cursor3.Revision != afterRev {
		t.Fatalf("third scan cursor revision: got %q want %q", cursor3.Revision, afterRev)
	}
}

func TestRuntimeScanBranchRevisionBuildsRevisionEvent(t *testing.T) {
	store := newFakeStore()
	r := NewRuntimeWithStore(nil, store)
	workdir, initialRev := setupGitRepo(t)

	// Register watch
	_, err := r.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	opts := ScanRevisionOptions{
		RepoID:  "myrepo",
		Branch:  "main",
		Workdir: workdir,
		Runner:  gitengine.CLI{},
	}

	// First scan (observation)
	_, _, err = r.ScanBranchRevision(context.Background(), opts)
	if err != nil {
		t.Fatalf("first scan: %v", err)
	}

	// Make changes
	writeFile(t, filepath.Join(workdir, "newfile.txt"), "new file")
	writeFile(t, filepath.Join(workdir, "README.md"), "# Seed Repository\n\nupdated\n")
	runGit(t, workdir, "add", "newfile.txt", "README.md")
	runGit(t, workdir, "commit", "-m", "add newfile and update readme")

	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = workdir
	out, err := cmd.Output()
	if err != nil {
		t.Fatalf("get head: %v", err)
	}
	afterRev := strings.TrimSpace(string(out))

	// Second scan (changes)
	record, matched, err := r.ScanBranchRevision(context.Background(), opts)
	if err != nil {
		t.Fatalf("second scan: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}
	if record.Revision.Before != initialRev {
		t.Fatalf("before: got %q want %q", record.Revision.Before, initialRev)
	}
	if record.Revision.After != afterRev {
		t.Fatalf("after: got %q want %q", record.Revision.After, afterRev)
	}

	// Verify changed files
	if len(record.Revision.ChangedFiles) != 2 {
		t.Fatalf("expected 2 changed files, got %d", len(record.Revision.ChangedFiles))
	}
	filesMap := make(map[string]string)
	for _, f := range record.Revision.ChangedFiles {
		filesMap[f.Path] = f.ChangeType
	}
	if filesMap["README.md"] != "modified" {
		t.Fatalf("README.md change type: got %q want 'modified'", filesMap["README.md"])
	}
	if filesMap["newfile.txt"] != "added" {
		t.Fatalf("newfile.txt change type: got %q want 'added'", filesMap["newfile.txt"])
	}

	// Check cursor updated
	cursor, ok, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil || !ok {
		t.Fatalf("get cursor: err=%v ok=%v", err, ok)
	}
	if cursor.Revision != afterRev {
		t.Fatalf("cursor: got %q want %q", cursor.Revision, afterRev)
	}

	// Test no watch event emit
	optsNoWatch := ScanRevisionOptions{
		RepoID:  "nowatch",
		Branch:  "main",
		Workdir: workdir,
		Runner:  gitengine.CLI{},
	}
	// Initial scan for nowatch (stores cursor)
	_, _, err = r.ScanBranchRevision(context.Background(), optsNoWatch)
	if err != nil {
		t.Fatalf("first scan nowatch: %v", err)
	}

	// Add another commit
	writeFile(t, filepath.Join(workdir, "another.txt"), "another")
	runGit(t, workdir, "add", "another.txt")
	runGit(t, workdir, "commit", "-m", "add another")

	// Second scan for nowatch (no watch registered)
	beforeEvents := len(r.ListEvents())
	_, matchedNoWatch, err := r.ScanBranchRevision(context.Background(), optsNoWatch)
	if err != nil {
		t.Fatalf("second scan nowatch: %v", err)
	}
	if matchedNoWatch {
		t.Fatal("expected matched=false for unwatched repo")
	}
	afterEvents := len(r.ListEvents())
	if afterEvents != beforeEvents {
		t.Fatalf("expected events count to remain %d, got %d", beforeEvents, afterEvents)
	}

	// Verify unwatched cursor advanced to latest head revision
	latestHead, err := gitengine.HeadRevision(gitengine.CLI{}, workdir)
	if err != nil {
		t.Fatalf("failed to get head rev for verify: %v", err)
	}
	cursorNoWatch, okNoWatch, err := store.cursors.GetRevisionCursor(context.Background(), "nowatch", "main")
	if err != nil || !okNoWatch {
		t.Fatalf("get nowatch cursor: err=%v ok=%v", err, okNoWatch)
	}
	if cursorNoWatch.Revision != latestHead {
		t.Fatalf("unwatched cursor did not advance: got %q want %q", cursorNoWatch.Revision, latestHead)
	}
}

func TestRuntimeHandleProviderPollRevisionUsesRevisionIdentity(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	r := NewRuntimeWithStore(broadcaster, store)

	_, err := r.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	observedAt := time.Date(2026, 6, 18, 9, 30, 0, 0, time.UTC)
	poll := ProviderPollRevision{
		Provider: "forgejo",
		RepoID:   "nomadcode",
		Branch:   "develop",
		Before:   "rev-before",
		After:    "rev-after",
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
		ObservedAt: observedAt,
	}

	record, matched, err := r.HandleProviderPollRevision(context.Background(), poll)
	if err != nil {
		t.Fatalf("handle provider poll revision: %v", err)
	}
	if !matched {
		t.Fatal("expected provider poll revision to match branch watch")
	}
	if record.Type != events.BranchUpdated {
		t.Fatalf("event type: got %q want %q", record.Type, events.BranchUpdated)
	}
	if record.Provider != "forgejo" {
		t.Fatalf("provider: got %q want forgejo", record.Provider)
	}
	if record.DeliveryID != "" {
		t.Fatalf("provider poll source must not require provider delivery id, got %q", record.DeliveryID)
	}
	if record.Revision.RepoID != "nomadcode" ||
		record.Revision.Branch != "develop" ||
		record.Revision.Before != "rev-before" ||
		record.Revision.After != "rev-after" {
		t.Fatalf("revision identity mismatch: %+v", record.Revision)
	}
	if !record.Revision.ObservedAt.Equal(observedAt) {
		t.Fatalf("observed_at: got %s want %s", record.Revision.ObservedAt, observedAt)
	}

	dedupeKey := "revision:nomadcode:develop:rev-before:rev-after"
	if got := store.deliveries.records[dedupeKey]; got.EventID != record.ID {
		t.Fatalf("poll source dedupe key mismatch: got %+v want event_id %q", got, record.ID)
	}
	if len(broadcaster.envelopes) != 1 {
		t.Fatalf("expected one branch.updated envelope, got %d", len(broadcaster.envelopes))
	}
	payload := broadcaster.envelopes[0].Payload
	if payload["repo_id"] != "nomadcode" || payload["branch"] != "develop" ||
		payload["before"] != "rev-before" || payload["after"] != "rev-after" {
		t.Fatalf("poll source payload identity mismatch: %#v", payload)
	}

	duplicate, matched, err := r.HandleProviderPollRevision(context.Background(), poll)
	if err != nil {
		t.Fatalf("handle duplicate provider poll revision: %v", err)
	}
	if !matched {
		t.Fatal("expected duplicate provider poll revision to still match branch watch")
	}
	if !duplicate.Duplicate {
		t.Fatal("expected duplicate provider poll revision to be marked duplicate")
	}
	if duplicate.ID != record.ID {
		t.Fatalf("duplicate event id: got %q want %q", duplicate.ID, record.ID)
	}
	if len(r.ListEvents()) != 1 {
		t.Fatalf("duplicate poll source must not append another event, got %d", len(r.ListEvents()))
	}
}

type fakeBroadcaster struct {
	envelopes []protosocket.Envelope
	err       error
}

func (f *fakeBroadcaster) BroadcastEnvelope(ctx context.Context, env protosocket.Envelope) error {
	f.envelopes = append(f.envelopes, env)
	return f.err
}

func TestRuntimeScanAgentRunRevision(t *testing.T) {
	store := newFakeStore()

	// Register watch
	rWatch := NewRuntimeWithStore(nil, store)
	_, err := rWatch.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// 1. Broadcaster is nil -> ScanAgentRunRevision fails on match (not published)
	rNilBroadcaster := NewRuntimeWithStore(nil, store)

	// Create mock git workspace
	workdir := t.TempDir()
	runGit(t, workdir, "init")
	runGit(t, workdir, "config", "user.name", "test")
	runGit(t, workdir, "config", "user.email", "test@example.com")

	// First commit to set HEAD
	writeFile(t, filepath.Join(workdir, "README.md"), "hello")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "first commit")

	firstRev, _ := gitengine.HeadRevision(gitengine.CLI{}, workdir)

	// Setup initial cursor
	err = store.cursors.UpsertRevisionCursor(context.Background(), core.RevisionCursor{
		RepoID:   "myrepo",
		Branch:   "main",
		Revision: firstRev,
	})
	if err != nil {
		t.Fatalf("upsert cursor: %v", err)
	}

	// Add second commit
	writeFile(t, filepath.Join(workdir, "README.md"), "hello world")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "second commit")

	req := worker.RevisionScanRequest{
		Provider:   "forgejo",
		RepoID:     "myrepo",
		Branch:     "main",
		WorkDir:    workdir,
		Runner:     gitengine.CLI{},
		ObservedAt: time.Now(),
	}

	// Scanner with nil broadcaster should return error when matched because event isn't published
	_, err = rNilBroadcaster.ScanAgentRunRevision(context.Background(), req)
	if err == nil {
		t.Fatal("expected error on matched scan with nil broadcaster")
	}
	if !strings.Contains(err.Error(), "broadcaster is nil") {
		t.Fatalf("expected broadcaster is nil error, got %v", err)
	}

	// 2. Broadcaster is non-nil and succeeds -> ScanAgentRunRevision succeeds
	broadcaster := &fakeBroadcaster{}
	store2 := newFakeStore()
	rWithBroadcaster := NewRuntimeWithStore(broadcaster, store2)
	_, err = rWithBroadcaster.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// We need to reset the cursor back to firstRev to detect changes again
	err = store2.cursors.UpsertRevisionCursor(context.Background(), core.RevisionCursor{
		RepoID:   "myrepo",
		Branch:   "main",
		Revision: firstRev,
	})
	if err != nil {
		t.Fatalf("upsert cursor: %v", err)
	}

	res, err := rWithBroadcaster.ScanAgentRunRevision(context.Background(), req)
	if err != nil {
		t.Fatalf("unexpected ScanAgentRunRevision error: %v", err)
	}
	if !res.Matched {
		t.Fatal("expected matched=true")
	}
	if res.EventID == "" {
		t.Fatal("expected non-empty EventID")
	}
	if len(broadcaster.envelopes) != 1 {
		t.Fatalf("expected 1 broadcasted envelope, got %d", len(broadcaster.envelopes))
	}
}

type fakeOpEvents struct {
	events []storage.OperationEvent
	err    error
}

func (f *fakeOpEvents) AppendEvent(ctx context.Context, ev storage.OperationEvent) error {
	f.events = append(f.events, ev)
	return f.err
}
func (f *fakeOpEvents) ListEvents(ctx context.Context, operationID string) ([]storage.OperationEvent, error) {
	return nil, nil
}
func (f *fakeOpEvents) ListPendingEvents(ctx context.Context, limit int) ([]storage.OperationEvent, error) {
	return nil, nil
}
func (f *fakeOpEvents) MarkPublished(ctx context.Context, eventID string, now time.Time) (storage.OperationEvent, error) {
	return storage.OperationEvent{}, nil
}

func TestOutboxBroadcaster(t *testing.T) {
	eventsStore := &fakeOpEvents{}
	broadcaster := NewOutboxBroadcaster(eventsStore)

	// envelope with different channel/action should be skipped
	err := broadcaster.BroadcastEnvelope(context.Background(), protosocket.Envelope{
		Channel: "repo",
		Action:  "register",
		Payload: map[string]any{},
	})
	if err != nil {
		t.Fatalf("unexpected error on non-matched envelope: %v", err)
	}
	if len(eventsStore.events) != 0 {
		t.Fatalf("expected 0 events appended, got %d", len(eventsStore.events))
	}

	// envelope with branch.updated channel/action should append to outbox
	payload := map[string]any{
		"id":      "evt-123",
		"repo_id": "myrepo",
		"branch":  "main",
		"before":  "rev-before",
		"after":   "rev-after",
	}
	err = broadcaster.BroadcastEnvelope(context.Background(), protosocket.Envelope{
		Channel: "event",
		Action:  "branch.updated",
		Payload: payload,
	})
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(eventsStore.events) != 1 {
		t.Fatalf("expected 1 event appended, got %d", len(eventsStore.events))
	}

	ev := eventsStore.events[0]
	if ev.OperationID != "" {
		t.Errorf("expected empty OperationID for generic event, got %q", ev.OperationID)
	}
	if ev.Event.ID != "evt-123" {
		t.Errorf("expected Event ID 'evt-123', got %q", ev.Event.ID)
	}
	if ev.Event.Type != events.BranchUpdated {
		t.Errorf("expected Event Type %q, got %q", events.BranchUpdated, ev.Event.Type)
	}
	if ev.Event.Subject != "repo:myrepo" {
		t.Errorf("expected Subject 'repo:myrepo', got %q", ev.Event.Subject)
	}
}

func TestRuntimeScanAgentRunRevisionPublishFailureDoesNotAdvanceDurableState(t *testing.T) {
	store := newFakeStore()

	// Register watch
	rWatch := NewRuntimeWithStore(nil, store)
	_, err := rWatch.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Create mock git workspace
	workdir := t.TempDir()
	runGit(t, workdir, "init")
	runGit(t, workdir, "config", "user.name", "test")
	runGit(t, workdir, "config", "user.email", "test@example.com")

	// First commit to set HEAD
	writeFile(t, filepath.Join(workdir, "README.md"), "hello")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "first commit")

	firstRev, _ := gitengine.HeadRevision(gitengine.CLI{}, workdir)

	// Setup initial cursor
	err = store.cursors.UpsertRevisionCursor(context.Background(), core.RevisionCursor{
		RepoID:   "myrepo",
		Branch:   "main",
		Revision: firstRev,
	})
	if err != nil {
		t.Fatalf("upsert cursor: %v", err)
	}

	// Add second commit
	writeFile(t, filepath.Join(workdir, "README.md"), "hello world")
	runGit(t, workdir, "add", "README.md")
	runGit(t, workdir, "commit", "-m", "second commit")

	secondRev, _ := gitengine.HeadRevision(gitengine.CLI{}, workdir)

	req := worker.RevisionScanRequest{
		Provider:   "forgejo",
		RepoID:     "myrepo",
		Branch:     "main",
		WorkDir:    workdir,
		Runner:     gitengine.CLI{},
		ObservedAt: time.Now(),
	}

	// 1. Failing broadcaster
	failingBroadcaster := &fakeBroadcaster{err: fmt.Errorf("broadcast failed")}
	rFail := NewRuntimeWithStore(failingBroadcaster, store)

	_, err = rFail.ScanAgentRunRevision(context.Background(), req)
	if err == nil {
		t.Fatal("expected error on matched scan with failing broadcaster")
	}

	// Check that cursor was NOT advanced (remains firstRev)
	cur, found, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil || !found {
		t.Fatalf("failed to get cursor: %v, found: %t", err, found)
	}
	if cur.Revision != firstRev {
		t.Errorf("expected cursor to remain at %s, got %s", firstRev, cur.Revision)
	}

	// Check that dedupe record was deleted/not present so retry is allowed
	// dedupeKeyFor(deliveryID, revision) -> revision:myrepo:main:firstRev:secondRev
	dedupeKey := fmt.Sprintf("revision:myrepo:main:%s:%s", firstRev, secondRev)
	if _, ok := store.deliveries.records[dedupeKey]; ok {
		t.Error("expected dedupe delivery record to be deleted after failure")
	}

	// 2. Succeeding broadcaster (Retry)
	successBroadcaster := &fakeBroadcaster{}
	rSuccess := NewRuntimeWithStore(successBroadcaster, store)

	res, err := rSuccess.ScanAgentRunRevision(context.Background(), req)
	if err != nil {
		t.Fatalf("unexpected retry error: %v", err)
	}
	if !res.Matched {
		t.Fatal("expected matched=true on retry")
	}

	// Check that cursor advanced to secondRev
	curAfter, found, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil || !found {
		t.Fatalf("failed to get cursor after success: %v, found: %t", err, found)
	}
	if curAfter.Revision != secondRev {
		t.Errorf("expected cursor to advance to %s, got %s", secondRev, curAfter.Revision)
	}

	// Check that one event was successfully broadcasted
	if len(successBroadcaster.envelopes) != 1 {
		t.Errorf("expected 1 broadcasted envelope, got %d", len(successBroadcaster.envelopes))
	}
}

// TestRuntimeScanAgentRunRevisionEmitsWithoutExistingCursor verifies that when
// an explicit before/after pair is provided, a branch.updated event is emitted
// even if no cursor exists yet for the branch. changed_files must be populated
// from the real git diff between before and after.
func TestRuntimeScanAgentRunRevisionEmitsWithoutExistingCursor(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	r := NewRuntimeWithStore(broadcaster, store)

	_, err := r.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Set up a real git repo with two commits so changed_files can be computed.
	workdir, beforeRev := setupGitRepo(t)
	writeFile(t, filepath.Join(workdir, "agent.txt"), "agent output\n")
	runGit(t, workdir, "add", "agent.txt")
	runGit(t, workdir, "config", "commit.gpgsign", "false")
	runGit(t, workdir, "commit", "-m", "agent run result")
	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = workdir
	out, err := cmd.Output()
	if err != nil {
		t.Fatalf("get after rev: %v", err)
	}
	afterRev := strings.TrimSpace(string(out))

	// No cursor seeded — branch has never been observed.
	req := worker.RevisionScanRequest{
		Provider:       "forgejo",
		RepoID:         "myrepo",
		Branch:         "main",
		WorkDir:        workdir,
		Runner:         gitengine.CLI{},
		BeforeRevision: beforeRev,
		AfterRevision:  afterRev,
		ObservedAt:     time.Now().UTC(),
	}

	res, err := r.ScanAgentRunRevision(context.Background(), req)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if !res.Attempted {
		t.Fatal("expected attempted=true")
	}
	if !res.Matched {
		t.Fatal("expected matched=true when watch exists")
	}
	if res.EventID == "" {
		t.Fatal("expected non-empty event_id")
	}

	// Cursor must be advanced to after-revision.
	cursor, ok, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "main")
	if err != nil || !ok {
		t.Fatalf("cursor not stored after explicit scan: err=%v ok=%v", err, ok)
	}
	if cursor.Revision != afterRev {
		t.Fatalf("cursor revision: got %q want %q", cursor.Revision, afterRev)
	}

	// Broadcaster must have received the event with changed_files.
	if len(broadcaster.envelopes) != 1 {
		t.Fatalf("expected 1 broadcasted envelope, got %d", len(broadcaster.envelopes))
	}
	payload := broadcaster.envelopes[0].Payload
	files, _ := payload["changed_files"].([]any)
	if len(files) == 0 {
		t.Errorf("changed_files must be non-empty in branch.updated event; payload=%v", payload)
	}
}

// TestRuntimeScanAgentRunRevisionPublishFailureNoCursorAdvance verifies that
// when the broadcaster fails, the durable cursor is not advanced — even with
// an explicit before/after pair.
func TestRuntimeScanAgentRunRevisionPublishFailureNoCursorAdvance(t *testing.T) {
	store := newFakeStore()
	failBroadcaster := &fakeBroadcaster{err: fmt.Errorf("publish failed")}
	r := NewRuntimeWithStore(failBroadcaster, store)

	_, err := r.RegisterBranchWatch("myrepo", "feature", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Set up a real git repo with two commits.
	workdir, beforeRev := setupGitRepo(t)
	writeFile(t, filepath.Join(workdir, "feature.txt"), "feature work\n")
	runGit(t, workdir, "add", "feature.txt")
	runGit(t, workdir, "config", "commit.gpgsign", "false")
	runGit(t, workdir, "commit", "-m", "feature commit")
	cmd := exec.Command("git", "rev-parse", "HEAD")
	cmd.Dir = workdir
	out, err := cmd.Output()
	if err != nil {
		t.Fatalf("get after rev: %v", err)
	}
	afterRev := strings.TrimSpace(string(out))

	req := worker.RevisionScanRequest{
		Provider:       "forgejo",
		RepoID:         "myrepo",
		Branch:         "feature",
		WorkDir:        workdir,
		Runner:         gitengine.CLI{},
		BeforeRevision: beforeRev,
		AfterRevision:  afterRev,
		ObservedAt:     time.Now().UTC(),
	}

	_, scanErr := r.ScanAgentRunRevision(context.Background(), req)
	if scanErr == nil {
		t.Fatal("expected error when broadcaster fails, got nil")
	}

	// Cursor must NOT be advanced because publish failed.
	_, found, err := store.cursors.GetRevisionCursor(context.Background(), "myrepo", "feature")
	if err != nil {
		t.Fatalf("get cursor: %v", err)
	}
	if found {
		t.Fatal("cursor must not be stored when publish fails")
	}
}

// --- Webhook Delivery Tests ---

// fakeWebhookConsumer captures received HTTP requests for testing.
type fakeWebhookConsumer struct {
	mu       sync.Mutex
	received []webhookReception
}

type webhookReception struct {
	url        string
	headers    map[string]string
	body       map[string]any
	bodyRaw    []byte
	eventType  string
	deliveryID string
	signature  string
}

func (c *fakeWebhookConsumer) Record(req *http.Request, body []byte) error {
	rec := webhookReception{
		url:     req.URL.String(),
		bodyRaw: body,
	}

	// Extract headers we care about
	rec.eventType = req.Header.Get("X-Gito-Event")
	rec.deliveryID = req.Header.Get("X-Gito-Delivery")
	rec.signature = req.Header.Get("X-Gito-Signature")

	// Parse body
	if len(body) > 0 {
		var m map[string]any
		if err := json.Unmarshal(body, &m); err == nil {
			rec.body = m
		}
	}
	c.mu.Lock()
	c.received = append(c.received, rec)
	c.mu.Unlock()
	return nil
}

// fakeWebhookServer creates an httptest.Server that records incoming webhook deliveries.
func fakeWebhookServer(t *testing.T, consumer *fakeWebhookConsumer) *httptest.Server {
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(r.Body)
		if err != nil {
			t.Fatalf("read body: %v", err)
		}
		w.WriteHeader(http.StatusOK)
		_ = consumer.Record(r, body)
	}))
}

// TestRuntime_WebhookDeliveryBranchUpdated verifies that when a branch.updated
// event matches a webhook subscription, an HTTP POST is sent to the target URL
// with correct headers and payload (S02).
func TestRuntime_WebhookDeliveryBranchUpdated(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)
	runtime.WithWebhookClient(server.Client())

	// Register a webhook subscription that watches branch.updated events
	_, err := runtime.RegisterWebhookSubscription(
		"test-webhook", server.URL,
		[]string{"branch.updated"},
		"nomadcode", "", // empty repo_id/branch = match all
		"", // no secret
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	// Register a branch watch
	_, err = runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Handle a revision event
	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa111",
		After:      "bbb222",
		ObservedAt: time.Now().UTC(),
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "test-delivery-1", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	// Verify the fake consumer received exactly one POST
	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()

	if len(received) != 1 {
		t.Fatalf("expected 1 webhook reception, got %d", len(received))
	}

	rec := received[0]

	// Check headers
	if rec.eventType != "branch.updated" {
		t.Errorf("X-Gito-Event: got %q want %q", rec.eventType, "branch.updated")
	}
	if rec.deliveryID != "test-delivery-1" {
		t.Errorf("X-Gito-Delivery: got %q want %q", rec.deliveryID, "test-delivery-1")
	}
	if rec.signature != "" {
		t.Error("expected no X-Gito-Signature when no secret is set")
	}

	// Check body fields
	body := rec.body
	if body == nil {
		t.Fatal("expected non-nil body")
	}
	expectedFields := []string{"id", "type", "provider", "delivery_id", "repo_id", "branch", "before", "after", "changed_files", "observed_at", "created_at"}
	for _, field := range expectedFields {
		if _, ok := body[field]; !ok {
			t.Errorf("expected field %q in webhook payload", field)
		}
	}

	// Check specific values
	if body["type"] != "branch.updated" {
		t.Errorf("type: got %v want %q", body["type"], "branch.updated")
	}
	if body["provider"] != "forgejo" {
		t.Errorf("provider: got %v want %q", body["provider"], "forgejo")
	}
	if body["repo_id"] != "nomadcode" {
		t.Errorf("repo_id: got %v want %q", body["repo_id"], "nomadcode")
	}
	if body["branch"] != "develop" {
		t.Errorf("branch: got %v want %q", body["branch"], "develop")
	}
	if body["before"] != "aaa111" {
		t.Errorf("before: got %v want %q", body["before"], "aaa111")
	}
	if body["after"] != "bbb222" {
		t.Errorf("after: got %v want %q", body["after"], "bbb222")
	}

	// Check that X-Gito-Delivery is non-empty (already checked above but explicit)
	if rec.deliveryID == "" {
		t.Error("X-Gito-Delivery must be non-empty")
	}

	// Verify no nomadcode-specific or plane-specific fields in payload
	for key := range body {
		if strings.HasPrefix(key, "nomadcode_") || strings.HasPrefix(key, "plane_") {
			t.Errorf("unexpected consumer-specific field %q in payload", key)
		}
	}

	// Verify broadcast still happened
	if len(broadcaster.envelopes) != 1 {
		t.Errorf("expected 1 broadcasted envelope, got %d", len(broadcaster.envelopes))
	}
}

// TestRuntime_WebhookDeliveryWithSecret verifies that when a subscription has a
// secret_ref, the X-Gito-Signature header is set with HMAC-SHA256 signature.
func TestRuntime_WebhookDeliveryWithSecret(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)
	runtime.WithWebhookClient(server.Client())

	testSecret := "my-secret-key"
	runtime.WithSecretResolver(&fakeSecretResolver{secret: testSecret, resolved: true})

	_, err := runtime.RegisterWebhookSubscription(
		"secured-webhook", server.URL,
		[]string{"branch.updated"},
		"myrepo", "",
		"secret-ref-1",
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "myrepo",
		Branch:     "main",
		Before:     "abc",
		After:      "def",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "sec-delivery-1", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()

	if len(received) != 1 {
		t.Fatalf("expected 1 webhook reception, got %d", len(received))
	}

	rec := received[0]
	if rec.signature == "" {
		t.Fatal("expected X-Gito-Signature when secret is set")
	}

	// Verify signature: sha256=HMAC-SHA256(secret, body)
	mac := hmac.New(sha256.New, []byte(testSecret))
	mac.Write(rec.bodyRaw)
	expectedSig := "sha256=" + hex.EncodeToString(mac.Sum(nil))
	if rec.signature != expectedSig {
		t.Errorf("X-Gito-Signature: got %q want %q", rec.signature, expectedSig)
	}
}

// TestRuntime_WebhookDeliveryNoSecretExposure verifies that the secret value is
// never exposed in the response body or logs.
func TestRuntime_WebhookDeliveryNoSecretExposure(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)
	runtime.WithWebhookClient(server.Client())

	runtime.WithSecretResolver(&fakeSecretResolver{secret: "super-secret", resolved: true})

	_, err := runtime.RegisterWebhookSubscription(
		"no-exposure-webhook", server.URL,
		[]string{"branch.updated"},
		"secure-repo", "",
		"secret-abc",
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("secure-repo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "secure-repo",
		Branch:     "main",
		Before:     "111",
		After:      "222",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "expose-test", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()

	if len(received) != 1 {
		t.Fatalf("expected 1 webhook reception, got %d", len(received))
	}

	body := received[0].body
	if body == nil {
		t.Fatal("expected non-nil body")
	}

	// Check that the secret value does not appear in the payload body
	bodyStr, err := json.Marshal(body)
	if err != nil {
		t.Fatalf("marshal body: %v", err)
	}
	if strings.Contains(string(bodyStr), "super-secret") {
		t.Error("secret value must not appear in webhook payload body")
	}

	// Check that secret_ref is not exposed
	if strings.Contains(string(bodyStr), "secret-abc") {
		t.Error("secret_ref must not appear in webhook payload body")
	}
}

// TestRuntime_WebhookDeliveryFiltering verifies that webhooks are only sent
// to subscriptions that match the event type, repo_id, and branch.
func TestRuntime_WebhookDeliveryFiltering(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)
	runtime.WithWebhookClient(server.Client())

	// Register multiple subscriptions with different filters
	_, err := runtime.RegisterWebhookSubscription(
		"all-branch-updated", server.URL,
		[]string{"branch.updated", "push"},
		"", "", "", // match all
	)
	if err != nil {
		t.Fatalf("register webhook all: %v", err)
	}

	_, err = runtime.RegisterWebhookSubscription(
		"nomadcode-only", server.URL,
		[]string{"branch.updated"},
		"nomadcode", "", "", // only nomadcode repo
	)
	if err != nil {
		t.Fatalf("register webhook nomadcode: %v", err)
	}

	_, err = runtime.RegisterWebhookSubscription(
		"develop-branch-only", server.URL,
		[]string{"branch.updated"},
		"", "main", "", // only main branch filter
	)
	if err != nil {
		t.Fatalf("register webhook develop: %v", err)
	}

	_, err = runtime.RegisterWebhookSubscription(
		"push-only", server.URL,
		[]string{"push"},
		"", "", "", // only push events
	)
	if err != nil {
		t.Fatalf("register webhook push: %v", err)
	}

	// Register branch watch
	_, err = runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Handle a branch.updated event for nomadcode/develop
	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa",
		After:      "bbb",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "filter-test", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()

	// Expected: all-branch-updated (matches event type, no repo/branch filter)
	//           nomadcode-only (matches event type, repo_id=nomadcode matches)
	// NOT: develop-branch-only (branch filter is "main" but event is "develop")
	// NOT: push-only (event type is "branch.updated" not "push")
	if len(received) != 2 {
		t.Fatalf("expected 2 webhook receptions, got %d", len(received))
	}
}

// fakeSecretResolver implements WebhookSecretResolver for testing.
type fakeSecretResolver struct {
	secret   string
	resolved bool
	err      error
}

func (f *fakeSecretResolver) ResolveWebhookSecret(_ context.Context, secretRef string) (string, bool, error) {
	if f.err != nil {
		return "", false, f.err
	}
	if f.resolved {
		return f.secret, true, nil
	}
	return "", false, nil
}

var _ WebhookSecretResolver = (*fakeSecretResolver)(nil)

// TestRuntime_WebhookDeliveryUnsignedSecretRefFails verifies that when a subscription
// has secret_ref but the resolver is nil or returns unresolved, the delivery fails
// with an error (no unsigned delivery is sent).
func TestRuntime_WebhookDeliveryUnsignedSecretRefFails(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)
	runtime.WithWebhookClient(server.Client())
	// No secret resolver set — resolver is nil

	_, err := runtime.RegisterWebhookSubscription(
		"secured-webhook", server.URL,
		[]string{"branch.updated"},
		"myrepo", "",
		"secret-ref-1",
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "myrepo",
		Branch:     "main",
		Before:     "abc",
		After:      "def",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "unsigned-test-1", revision)
	if err == nil {
		t.Fatal("expected error when secret_ref is set but resolver is nil")
	}
	if !matched {
		t.Fatal("expected matched=true")
	}
	if !strings.Contains(err.Error(), "secret_ref") {
		t.Errorf("expected error to mention secret_ref, got: %v", err)
	}

	// Verify no request was sent to the consumer
	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()
	if len(received) != 0 {
		t.Fatalf("expected 0 receptions (unsigned delivery blocked), got %d", len(received))
	}
}

// TestRuntime_WebhookDeliveryUnsignedSecretRefFails_Unresolved verifies that when
// the resolver returns resolved=false, the delivery fails with an error.
func TestRuntime_WebhookDeliveryUnsignedSecretRefFails_Unresolved(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	runtime := NewRuntimeWithStore(nil, store)
	runtime.WithWebhookClient(server.Client())
	runtime.WithSecretResolver(&fakeSecretResolver{secret: "", resolved: false})

	_, err := runtime.RegisterWebhookSubscription(
		"secured-webhook", server.URL,
		[]string{"branch.updated"},
		"myrepo", "",
		"secret-ref-1",
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("myrepo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "myrepo",
		Branch:     "main",
		Before:     "abc",
		After:      "def",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "unresolved-test-1", revision)
	if err == nil {
		t.Fatal("expected error when secret_ref is unresolved")
	}
	if !matched {
		t.Fatal("expected matched=true")
	}
	if !strings.Contains(err.Error(), "unresolved") {
		t.Errorf("expected error to mention unresolved, got: %v", err)
	}

	// Verify no request was sent
	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()
	if len(received) != 0 {
		t.Fatalf("expected 0 receptions (unresolved delivery blocked), got %d", len(received))
	}
}

// TestRuntime_WebhookDeliveryFallbackDeliveryID verifies that when deliveryID is empty,
// the consumer still receives a stable non-empty X-Gito-Delivery header and delivery_id.
func TestRuntime_WebhookDeliveryFallbackDeliveryID(t *testing.T) {
	consumer := &fakeWebhookConsumer{}
	server := fakeWebhookServer(t, consumer)
	defer server.Close()

	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)
	runtime.WithWebhookClient(server.Client())

	_, err := runtime.RegisterWebhookSubscription(
		"test-webhook", server.URL,
		[]string{"branch.updated"},
		"nomadcode", "",
		"", // no secret
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Handle with empty deliveryID (simulates scan path)
	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa111",
		After:      "bbb222",
		ObservedAt: time.Now().UTC(),
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "", revision)
	if err != nil {
		t.Fatalf("handle revision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	consumer.mu.Lock()
	received := consumer.received
	consumer.mu.Unlock()

	if len(received) != 1 {
		t.Fatalf("expected 1 webhook reception, got %d", len(received))
	}

	rec := received[0]
	if rec.deliveryID == "" {
		t.Fatal("X-Gito-Delivery must be non-empty even when provider deliveryID is empty")
	}
	if rec.body == nil {
		t.Fatal("expected non-nil body")
	}
	bodyDeliveryID, ok := rec.body["delivery_id"].(string)
	if !ok || bodyDeliveryID == "" {
		t.Errorf("body delivery_id must be non-empty string, got: %v (ok=%v)", rec.body["delivery_id"], ok)
	}
}

// TestRuntime_WebhookDeliveryFailureAllowsRetry verifies that when webhook delivery fails,
// the provider delivery dedupe record is rolled back so that the same delivery can be
// retried successfully, and the same stable consumer delivery ID is preserved across
// retries even when provider deliveryID is empty.
func TestRuntime_WebhookDeliveryFailureAllowsRetry(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)

	// Use a simpler approach: track receptions in order with a custom server
	mu := sync.Mutex{}
	callNum := 0
	var receptions []webhookReception
	serverWithRetry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		mu.Lock()
		callNum++
		n := callNum
		mu.Unlock()

		body, _ := io.ReadAll(r.Body)
		rec := webhookReception{
			url:        r.URL.String(),
			eventType:  r.Header.Get("X-Gito-Event"),
			deliveryID: r.Header.Get("X-Gito-Delivery"),
			signature:  r.Header.Get("X-Gito-Signature"),
		}
		if len(body) > 0 {
			var m map[string]any
			if json.Unmarshal(body, &m) == nil {
				rec.body = m
			}
		}
		receptions = append(receptions, rec)

		if n == 1 {
			// First call: fail
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		// Second call: succeed
		w.WriteHeader(http.StatusOK)
	}))
	defer serverWithRetry.Close()

	runtime.WithWebhookClient(serverWithRetry.Client())

	_, err := runtime.RegisterWebhookSubscription(
		"test-webhook", serverWithRetry.URL,
		[]string{"branch.updated"},
		"nomadcode", "",
		"", // no secret
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa111",
		After:      "bbb222",
		ObservedAt: time.Now().UTC(),
	}

	// First call: should fail (call 1 in server)
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "", revision)
	if err == nil {
		t.Fatal("expected error on first delivery failure")
	}
	if !matched {
		t.Fatal("expected matched=true on first call")
	}

	mu.Lock()
	firstCallReceptions := len(receptions)
	mu.Unlock()
	if firstCallReceptions != 1 {
		t.Fatalf("expected 1 reception on first call, got %d", firstCallReceptions)
	}

	// Second call with same revision: should succeed (call 2 in server)
	_, matched, err = runtime.HandleRevision(context.Background(), "forgejo", "", revision)
	if err != nil {
		t.Fatalf("expected success on retry, got: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true on retry")
	}

	mu.Lock()
	finalReceptions := len(receptions)
	mu.Unlock()
	if finalReceptions != 2 {
		t.Fatalf("expected 2 receptions total, got %d", finalReceptions)
	}

	// Verify both receptions have the same X-Gito-Delivery header
	mu.Lock()
	firstDeliveryID := receptions[0].deliveryID
	secondDeliveryID := receptions[1].deliveryID
	mu.Unlock()

	if firstDeliveryID == "" {
		t.Fatal("first reception X-Gito-Delivery must be non-empty")
	}
	if secondDeliveryID == "" {
		t.Fatal("second reception X-Gito-Delivery must be non-empty")
	}
	if firstDeliveryID != secondDeliveryID {
		t.Errorf("X-Gito-Delivery mismatch across retry: first=%q second=%q (expected same stable ID)", firstDeliveryID, secondDeliveryID)
	}

	// Verify body delivery_id also matches
	mu.Lock()
	firstBodyDeliveryID, _ := receptions[0].body["delivery_id"].(string)
	secondBodyDeliveryID, _ := receptions[1].body["delivery_id"].(string)
	mu.Unlock()

	if firstBodyDeliveryID == "" {
		t.Fatal("first reception body delivery_id must be non-empty")
	}
	if secondBodyDeliveryID == "" {
		t.Fatal("second reception body delivery_id must be non-empty")
	}
	if firstBodyDeliveryID != secondBodyDeliveryID {
		t.Errorf("body delivery_id mismatch across retry: first=%q second=%q (expected same stable ID)", firstBodyDeliveryID, secondBodyDeliveryID)
	}

	// Verify the delivery ID is the deterministic fallback format
	expectedFallback := fmt.Sprintf("fallback:%s:%s:%s:%s", revision.RepoID, revision.Branch, revision.Before, revision.After)
	if firstDeliveryID != expectedFallback {
		t.Errorf("X-Gito-Delivery: got %q want %q", firstDeliveryID, expectedFallback)
	}
	if firstBodyDeliveryID != expectedFallback {
		t.Errorf("body delivery_id: got %q want %q", firstBodyDeliveryID, expectedFallback)
	}
}

// TestWebhookDeliveryBranchUpdated is an alias for TestRuntime_WebhookDeliveryBranchUpdated
// to match the SDD test evidence command: go test -run TestWebhookDeliveryBranchUpdated
// --- Reconcile Tests ---

func TestRuntimeReconcileExpectedRevisionConflict(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)

	// Register branch watch so events could match if emitted
	_, err := runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Dry-run with expected/current mismatch must return conflict
	req := ReconcileRequest{
		Provider:         "forgejo",
		RepoID:           "nomadcode",
		Branch:           "develop",
		ExpectedRevision: "rev-expected",
		CurrentRevision:  "rev-current",
		TargetRevision:   "rev-target",
	}
	result := runtime.DryRunReconcile(context.Background(), req)

	if !result.Conflict {
		t.Fatal("expected conflict for expected/current revision mismatch")
	}
	if result.Status != ReconcileStatusConflict {
		t.Fatalf("status: got %q want %q", result.Status, ReconcileStatusConflict)
	}
	if result.Applied {
		t.Fatal("expected applied=false for conflict")
	}
	if result.Reason == "" {
		t.Fatal("expected reason to be set")
	}

	// Events and broadcaster must not be affected by dry-run
	if len(runtime.ListEvents()) != 0 {
		t.Fatalf("expected 0 events after dry-run, got %d", len(runtime.ListEvents()))
	}
	if len(broadcaster.envelopes) != 0 {
		t.Fatalf("expected 0 envelopes after dry-run, got %d", len(broadcaster.envelopes))
	}

	// Apply with expected/current mismatch must also return conflict without side effects
	applyReq := req
	applyReq.IdempotencyKey = "conflict-apply-key"
	applyResult := runtime.ApplyReconcile(context.Background(), applyReq)
	if !applyResult.Conflict {
		t.Fatal("expected conflict for apply with expected/current mismatch")
	}
	if applyResult.Applied {
		t.Fatal("expected applied=false for conflict apply")
	}
	if len(runtime.ListEvents()) != 0 {
		t.Fatal("apply conflict must not create events")
	}
	if len(broadcaster.envelopes) != 0 {
		t.Fatal("apply conflict must not broadcast")
	}
}

func TestRuntimeDryRunReconcileReportsDriftConflictProposal(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)

	// Register watch
	_, err := runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Conflict case: expected != current
	conflictReq := ReconcileRequest{
		Provider:         "forgejo",
		RepoID:           "nomadcode",
		Branch:           "develop",
		ExpectedRevision: "rev-a",
		CurrentRevision:  "rev-b",
		TargetRevision:   "rev-c",
	}
	conflictResult := runtime.DryRunReconcile(context.Background(), conflictReq)
	if !conflictResult.Conflict {
		t.Fatal("expected conflict status")
	}
	if conflictResult.Status != ReconcileStatusConflict {
		t.Fatalf("conflict status: got %q", conflictResult.Status)
	}

	// Noop case: current == target
	noopReq := ReconcileRequest{
		Provider:        "forgejo",
		RepoID:          "nomadcode",
		Branch:          "develop",
		CurrentRevision: "rev-same",
		TargetRevision:  "rev-same",
	}
	noopResult := runtime.DryRunReconcile(context.Background(), noopReq)
	if noopResult.Status != ReconcileStatusNoop {
		t.Fatalf("noop status: got %q want %q", noopResult.Status, ReconcileStatusNoop)
	}
	if noopResult.Drift {
		t.Fatal("expected drift=false for noop")
	}
	if noopResult.Applied {
		t.Fatal("expected applied=false for dry-run noop")
	}

	// Proposal/drift case: current != target, no expected mismatch
	proposalReq := ReconcileRequest{
		Provider:        "forgejo",
		RepoID:          "nomadcode",
		Branch:          "develop",
		CurrentRevision: "rev-current",
		TargetRevision:  "rev-target",
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
	}
	proposalResult := runtime.DryRunReconcile(context.Background(), proposalReq)
	if proposalResult.Status != ReconcileStatusProposal {
		t.Fatalf("proposal status: got %q want %q", proposalResult.Status, ReconcileStatusProposal)
	}
	if !proposalResult.Drift {
		t.Fatal("expected drift=true for proposal")
	}
	if proposalResult.Applied {
		t.Fatal("expected applied=false for dry-run proposal")
	}
	if proposalResult.Proposal == nil {
		t.Fatal("expected proposal to be set")
	} else {
		if proposalResult.Proposal.Before != "rev-current" {
			t.Errorf("proposal before: got %q want %q", proposalResult.Proposal.Before, "rev-current")
		}
		if proposalResult.Proposal.After != "rev-target" {
			t.Errorf("proposal after: got %q want %q", proposalResult.Proposal.After, "rev-target")
		}
		if len(proposalResult.Proposal.Changed) != 1 {
			t.Fatalf("expected 1 changed file in proposal, got %d", len(proposalResult.Proposal.Changed))
		}
	}

	// Dry-run must never create events or broadcasts
	if len(runtime.ListEvents()) != 0 {
		t.Fatalf("expected 0 events after dry-run, got %d", len(runtime.ListEvents()))
	}
	if len(broadcaster.envelopes) != 0 {
		t.Fatalf("expected 0 envelopes after dry-run, got %d", len(broadcaster.envelopes))
	}
}

func TestRuntimeApplyReconcileIsIdempotent(t *testing.T) {
	store := newFakeStore()
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)

	// Register watch
	_, err := runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	req := ReconcileRequest{
		Provider:        "forgejo",
		RepoID:          "nomadcode",
		Branch:          "develop",
		CurrentRevision: "rev-before",
		TargetRevision:  "rev-after",
		IdempotencyKey:  "idempotent-key-1",
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
		ObservedAt: time.Now().UTC(),
	}

	// First apply: should succeed and create 1 event + 1 envelope
	firstResult := runtime.ApplyReconcile(context.Background(), req)
	if firstResult.Status != ReconcileStatusApplied {
		t.Fatalf("first apply status: got %q want %q", firstResult.Status, ReconcileStatusApplied)
	}
	if !firstResult.Applied {
		t.Fatal("expected applied=true on first apply")
	}
	if firstResult.Duplicate {
		t.Fatal("expected duplicate=false on first apply")
	}
	if len(runtime.ListEvents()) != 1 {
		t.Fatalf("expected 1 event after first apply, got %d", len(runtime.ListEvents()))
	}
	if len(broadcaster.envelopes) != 1 {
		t.Fatalf("expected 1 envelope after first apply, got %d", len(broadcaster.envelopes))
	}

	// Second apply with same idempotency key: should return duplicate without side effects
	secondResult := runtime.ApplyReconcile(context.Background(), req)
	if secondResult.Status != ReconcileStatusDuplicate {
		t.Fatalf("second apply status: got %q want %q", secondResult.Status, ReconcileStatusDuplicate)
	}
	if secondResult.Applied {
		t.Fatal("expected applied=false on duplicate apply")
	}
	if !secondResult.Duplicate {
		t.Fatal("expected duplicate=true on second apply")
	}

	// Side effects must not increase on duplicate
	if len(runtime.ListEvents()) != 1 {
		t.Fatalf("expected 1 event after duplicate apply, got %d", len(runtime.ListEvents()))
	}
	if len(broadcaster.envelopes) != 1 {
		t.Fatalf("expected 1 envelope after duplicate apply, got %d", len(broadcaster.envelopes))
	}

	// Expected/current mismatch apply must return conflict without side effects
	conflictReq := req
	conflictReq.ExpectedRevision = "rev-mismatch"
	conflictResult := runtime.ApplyReconcile(context.Background(), conflictReq)
	if !conflictResult.Conflict {
		t.Fatal("expected conflict for apply with expected mismatch")
	}
	if conflictResult.Applied {
		t.Fatal("expected applied=false for conflict apply")
	}
	if len(runtime.ListEvents()) != 1 {
		t.Fatal("conflict apply must not create additional events")
	}
	if len(broadcaster.envelopes) != 1 {
		t.Fatal("conflict apply must not broadcast additional envelopes")
	}
}

func TestWebhookDeliveryBranchUpdated(t *testing.T) {
	TestRuntime_WebhookDeliveryBranchUpdated(t)
}

func TestWebhookDeliveryRetryBackoffRecordsState(t *testing.T) {
	store := newFakeStore()
	store.webhookDeliveries = &fakeWebhookDeliveryStore{records: make(map[string]core.WebhookDelivery)}
	runtime := NewRuntimeWithStore(nil, store)

	var statusCode int
	var responseError error
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if responseError != nil {
			hj, ok := w.(http.Hijacker)
			if ok {
				conn, _, _ := hj.Hijack()
				conn.Close()
				return
			}
		}
		w.WriteHeader(statusCode)
	}))
	defer server.Close()

	runtime.WithWebhookClient(server.Client())

	sub, err := runtime.RegisterWebhookSubscription(
		"test-webhook", server.URL,
		[]string{"branch.updated"},
		"nomadcode", "",
		"",
	)
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}

	_, err = runtime.RegisterBranchWatch("nomadcode", "develop", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	revision := core.RevisionEvent{
		RepoID:     "nomadcode",
		Branch:     "develop",
		Before:     "aaa111",
		After:      "bbb222",
		ObservedAt: time.Now().UTC(),
	}

	// 1. Test Retryable Outcome (HTTP 500)
	statusCode = http.StatusInternalServerError
	record, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "del-123", revision)
	if err != nil {
		t.Fatalf("HandleRevision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched to be true")
	}

	dStore := store.WebhookDeliveries().(*fakeWebhookDeliveryStore)
	key := record.ID + ":" + sub.ID
	dStore.mu.Lock()
	del, ok := dStore.records[key]
	dStore.mu.Unlock()
	if !ok {
		t.Fatalf("expected delivery record to be created in store")
	}

	if del.Status != core.WebhookDeliveryRetryable {
		t.Errorf("expected status retryable, got %s", del.Status)
	}
	if del.AttemptCount != 1 {
		t.Errorf("expected attempt count 1, got %d", del.AttemptCount)
	}
	if del.LastStatusCode != 500 {
		t.Errorf("expected last status code 500, got %d", del.LastStatusCode)
	}
	if del.NextAttemptAt.Before(time.Now()) {
		t.Errorf("expected next attempt to be in the future")
	}

	// 2. Test Success Outcome (HTTP 200)
	dStore.mu.Lock()
	del.Status = core.WebhookDeliveryPending
	del.NextAttemptAt = time.Now().Add(-time.Hour)
	dStore.records[key] = del
	dStore.mu.Unlock()

	statusCode = http.StatusOK
	err = runtime.ProcessPendingWebhooks(context.Background())
	if err != nil {
		t.Fatalf("ProcessPendingWebhooks: %v", err)
	}

	dStore.mu.Lock()
	del = dStore.records[key]
	dStore.mu.Unlock()

	if del.Status != core.WebhookDeliverySucceeded {
		t.Errorf("expected status succeeded, got %s", del.Status)
	}
	if del.AttemptCount != 2 {
		t.Errorf("expected attempt count 2, got %d", del.AttemptCount)
	}
	if del.LastStatusCode != 200 {
		t.Errorf("expected last status code 200, got %d", del.LastStatusCode)
	}

	// 3. Test Network Transport Error (should be retryable)
	dStore.mu.Lock()
	del.Status = core.WebhookDeliveryPending
	del.NextAttemptAt = time.Now().Add(-time.Hour)
	dStore.records[key] = del
	dStore.mu.Unlock()

	responseError = fmt.Errorf("network error")
	err = runtime.ProcessPendingWebhooks(context.Background())
	if err != nil {
		t.Fatalf("ProcessPendingWebhooks: %v", err)
	}

	dStore.mu.Lock()
	del = dStore.records[key]
	dStore.mu.Unlock()

	if del.Status != core.WebhookDeliveryRetryable {
		t.Errorf("expected status retryable after transport error, got %s", del.Status)
	}
	if del.AttemptCount != 3 {
		t.Errorf("expected attempt count 3, got %d", del.AttemptCount)
	}
}

// TestWebhookDeliveryDoesNotDispatchUnrelatedPendingDelivery verifies that the immediate
// dispatch path only dispatches the delivery just enqueued for the current event/subscription,
// not an unrelated pending/retryable delivery that happened to be earlier in the global queue.
func TestWebhookDeliveryDoesNotDispatchUnrelatedPendingDelivery(t *testing.T) {
	store := newFakeStore()
	store.webhookDeliveries = &fakeWebhookDeliveryStore{records: make(map[string]core.WebhookDelivery)}
	broadcaster := &fakeBroadcaster{}
	runtime := NewRuntimeWithStore(broadcaster, store)

	mu := sync.Mutex{}
	var receptions []webhookReception
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		mu.Lock()
		receptions = append(receptions, webhookReception{
			url:        r.URL.String(),
			eventType:  r.Header.Get("X-Gito-Event"),
			deliveryID: r.Header.Get("X-Gito-Delivery"),
		})
		if len(body) > 0 {
			var m map[string]any
			if json.Unmarshal(body, &m) == nil {
				receptions[len(receptions)-1].body = m
			}
		}
		mu.Unlock()
		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()

	runtime.WithWebhookClient(server.Client())

	// 1. First subscription for event type "branch.updated" on repo "repo-a".
	subA, err := runtime.RegisterWebhookSubscription("sub-a", server.URL, []string{"branch.updated"}, "repo-a", "", "")
	if err != nil {
		t.Fatalf("register sub-a: %v", err)
	}

	// 2. Second subscription for same event on repo "repo-b" that will receive the new event.
	_, err = runtime.RegisterWebhookSubscription("sub-b", server.URL, []string{"branch.updated"}, "repo-b", "", "")
	if err != nil {
		t.Fatalf("register sub-b: %v", err)
	}

	// 3. Seed an unrelated pending delivery for subA (repo-a) that has an earlier next_attempt_at.
	// This simulates an existing pending/retryable delivery that would normally be picked first
	// by a global PickPendingDeliveries(limit=1) call.
	earlyDelivery := core.WebhookDelivery{
		ID:             "early-delivery-id",
		EventID:        "old-event-id",
		SubscriptionID: subA.ID,
		Status:         core.WebhookDeliveryPending,
		NextAttemptAt:  time.Now().Add(-time.Hour),
	}
	store.webhookDeliveries.records["old-event-id:"+subA.ID] = earlyDelivery

	// Register branch watches so HandleRevision uses the store path.
	_, err = runtime.RegisterBranchWatch("repo-a", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch repo-a: %v", err)
	}
	_, err = runtime.RegisterBranchWatch("repo-b", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch repo-b: %v", err)
	}

	// 4. Handle a new revision event for repo-b. This will enqueue a new delivery for subB.
	revision := core.RevisionEvent{
		RepoID:     "repo-b",
		Branch:     "main",
		Before:     "aaa",
		After:      "bbb",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "new-delivery-id", revision)
	if err != nil {
		t.Fatalf("HandleRevision: %v", err)
	}
	if !matched {
		t.Fatal("expected matched=true")
	}

	// 5. Verify that only the new delivery (for subB/repo-b) was dispatched.
	// The early delivery for subA should remain untouched.
	mu.Lock()
	defer mu.Unlock()

	if len(receptions) != 1 {
		t.Fatalf("expected 1 webhook reception, got %d", len(receptions))
	}

	rec := receptions[0]
	if rec.body == nil {
		t.Fatal("expected non-nil body")
	}
	repoID, ok := rec.body["repo_id"].(string)
	if !ok || repoID != "repo-b" {
		t.Errorf("expected repo_id=%q, got %v", "repo-b", rec.body["repo_id"])
	}

	// The early delivery should still be pending (not dispatched by the new event).
	store.webhookDeliveries.mu.Lock()
	early, ok := store.webhookDeliveries.records["old-event-id:"+subA.ID]
	store.webhookDeliveries.mu.Unlock()
	if !ok {
		t.Fatal("early delivery record disappeared")
	}
	if early.Status != core.WebhookDeliveryPending {
		t.Errorf("early delivery should still be pending, got %s", early.Status)
	}
}

// TestRuntimeApplyReconcileRetryAfterPublishFailure verifies that when the first
// ApplyReconcile fails during webhook/broadcast side effects, the apply idempotency
// key is NOT consumed, and a retry with the same key succeeds.
func TestRuntimeApplyReconcileRetryAfterPublishFailure(t *testing.T) {
	store := newFakeStore()

	// Register branch watch so the reconcile event matches
	_, err := store.watches.UpsertBranchWatch(context.Background(), core.BranchWatch{
		ID:       "watch-nomadcode-develop",
		Provider: "forgejo",
		RepoID:   "nomadcode",
		Branch:   "develop",
	})
	if err != nil {
		t.Fatalf("upsert branch watch: %v", err)
	}

	// 1. First apply with failing broadcaster -> conflict
	failingBroadcaster := &fakeBroadcaster{err: fmt.Errorf("broadcast failed")}
	rFail := NewRuntimeWithStore(failingBroadcaster, store)

	applyReq := ReconcileRequest{
		Provider:        "forgejo",
		RepoID:          "nomadcode",
		Branch:          "develop",
		CurrentRevision: "rev-before",
		TargetRevision:  "rev-after",
		IdempotencyKey:  "retry-key-1",
		ChangedFiles: []core.ChangedFile{
			{Path: "README.md", ChangeType: "modified"},
		},
		ObservedAt: time.Now().UTC(),
	}

	firstResult := rFail.ApplyReconcile(context.Background(), applyReq)
	if firstResult.Status != ReconcileStatusConflict {
		t.Fatalf("first apply status: got %q want %q", firstResult.Status, ReconcileStatusConflict)
	}
	if !firstResult.Conflict {
		t.Fatal("first apply: expected conflict=true")
	}

	// 2. Second apply with succeeding broadcaster -> applied
	successBroadcaster := &fakeBroadcaster{}
	rSuccess := NewRuntimeWithStore(successBroadcaster, store)

	secondResult := rSuccess.ApplyReconcile(context.Background(), applyReq)
	if secondResult.Status != ReconcileStatusApplied {
		t.Fatalf("second apply status: got %q want %q", secondResult.Status, ReconcileStatusApplied)
	}
	if !secondResult.Applied {
		t.Fatal("second apply: expected applied=true")
	}
	if secondResult.Duplicate {
		t.Fatal("second apply: expected duplicate=false")
	}

	// 3. Third apply with same key -> duplicate (side effects not repeated)
	thirdResult := rSuccess.ApplyReconcile(context.Background(), applyReq)
	if thirdResult.Status != ReconcileStatusDuplicate {
		t.Fatalf("third apply status: got %q want %q", thirdResult.Status, ReconcileStatusDuplicate)
	}
	if thirdResult.Applied {
		t.Fatal("third apply: expected applied=false")
	}
	if !thirdResult.Duplicate {
		t.Fatal("third apply: expected duplicate=true")
	}

	// 4. Events/envelopes count: successful retry creates 1 event + 1 envelope; duplicate creates nothing.
	if len(rSuccess.ListEvents()) != 1 {
		t.Fatalf("expected 1 event total, got %d", len(rSuccess.ListEvents()))
	}
	if len(successBroadcaster.envelopes) != 1 {
		t.Fatalf("expected 1 envelope total, got %d", len(successBroadcaster.envelopes))
	}
}

func TestWebhookDeliveryRecordOutcomeError(t *testing.T) {
	// Create a runtime with a store that has a webhook delivery store which returns an error on RecordOutcome.
	store := newFakeStore()
	store.webhookDeliveries = &fakeWebhookDeliveryStore{
		records:    make(map[string]core.WebhookDelivery),
		outcomeErr: fmt.Errorf("database connection lost"),
	}
	runtime := NewRuntimeWithStore(nil, store)

	// The fake server that succeeds on the HTTP side — so outcome failure is the only failure.
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()
	runtime.WithWebhookClient(server.Client())

	// Register subscription and watch.
	_, err := runtime.RegisterWebhookSubscription("test-sub", server.URL, []string{"branch.updated"}, "repo", "", "")
	if err != nil {
		t.Fatalf("register webhook: %v", err)
	}
	_, err = runtime.RegisterBranchWatch("repo", "main", "forgejo")
	if err != nil {
		t.Fatalf("register watch: %v", err)
	}

	// Handle revision — enqueue + dispatch + RecordOutcome.
	revision := core.RevisionEvent{
		RepoID:     "repo",
		Branch:     "main",
		Before:     "aaa",
		After:      "bbb",
		ObservedAt: time.Now().UTC(),
	}
	_, matched, err := runtime.HandleRevision(context.Background(), "forgejo", "test-del-1", revision)
	if !matched {
		t.Fatal("expected matched=true")
	}

	// Because RecordOutcome returns an error, HandleRevision should propagate it.
	// deliverWebhooks calls dispatchAndRecordOutcome which now returns error.
	if err == nil {
		t.Fatal("expected error when RecordOutcome fails, got nil")
	}
	if !strings.Contains(err.Error(), "record webhook delivery outcome") {
		t.Errorf("expected error to contain 'record webhook delivery outcome', got: %v", err)
	}
}