package storage_test

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

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

func TestPostgresRepoStoreCreatesListsUpdatesAndValidates(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	repo := core.Repo{
		ID:            "nomadcode",
		Name:          "NomadCode",
		RemoteURL:     "ssh://git.example.invalid/toki/nomadcode.git",
		DefaultBranch: "develop",
		WorkspaceRoot: "/workspaces/nomadcode",
		CredentialRef: "credential:gito/nomadcode",
	}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}
	if err := store.Repos().CreateRepo(ctx, repo); !errors.Is(err, storage.ErrRepoAlreadyExists) {
		t.Fatalf("duplicate repo id error: got %v want %v", err, storage.ErrRepoAlreadyExists)
	}
	if err := store.Repos().CreateRepo(ctx, core.Repo{ID: "missing-remote"}); !errors.Is(err, storage.ErrInvalidRepo) {
		t.Fatalf("required field validation error: got %v want %v", err, storage.ErrInvalidRepo)
	}

	got, err := store.Repos().GetRepo(ctx, repo.ID)
	if err != nil {
		t.Fatalf("get repo: %v", err)
	}
	if got != repo {
		t.Fatalf("get repo: got %+v want %+v", got, repo)
	}

	repos, err := store.Repos().ListRepos(ctx)
	if err != nil {
		t.Fatalf("list repos: %v", err)
	}
	if len(repos) != 1 || repos[0] != repo {
		t.Fatalf("list repos: got %+v want [%+v]", repos, repo)
	}

	updated := core.Repo{
		ID:            repo.ID,
		Name:          "NomadCode Core",
		RemoteURL:     repo.RemoteURL,
		DefaultBranch: "main",
		WorkspaceRoot: "/workspaces/nomadcode-main",
	}
	if err := store.Repos().UpdateRepo(ctx, updated); err != nil {
		t.Fatalf("update repo: %v", err)
	}
	got, err = store.Repos().GetRepo(ctx, repo.ID)
	if err != nil {
		t.Fatalf("get updated repo: %v", err)
	}
	if got != updated {
		t.Fatalf("updated repo: got %+v want %+v", got, updated)
	}

	if _, err := store.Repos().GetRepo(ctx, "missing"); !errors.Is(err, storage.ErrRepoNotFound) {
		t.Fatalf("missing repo error: got %v want %v", err, storage.ErrRepoNotFound)
	}
	missingUpdate := updated
	missingUpdate.ID = "missing"
	if err := store.Repos().UpdateRepo(ctx, missingUpdate); !errors.Is(err, storage.ErrRepoNotFound) {
		t.Fatalf("missing update error: got %v want %v", err, storage.ErrRepoNotFound)
	}
}

func TestPostgresWorkspaceLeaseStoreAcquiresOneAvailableSlotConcurrently(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	repo := core.Repo{
		ID:            "lease-repo",
		Name:          "Lease Repo",
		RemoteURL:     "ssh://git.example.invalid/toki/lease-repo.git",
		DefaultBranch: "main",
		WorkspaceRoot: "/workspaces/lease-repo",
	}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	seedLeases := []core.WorkspaceLease{
		{
			ID:     "lease-001",
			RepoID: repo.ID,
			Slot:   "001",
			Path:   "/workspaces/lease-repo/001",
			State:  core.WorkspaceAvailable,
		},
		{
			ID:     "lease-002",
			RepoID: repo.ID,
			Slot:   "002",
			Path:   "/workspaces/lease-repo/002",
			State:  core.WorkspaceAvailable,
		},
	}
	for _, lease := range seedLeases {
		if err := store.WorkspaceLeases().SaveLease(ctx, lease); err != nil {
			t.Fatalf("save lease %s: %v", lease.ID, err)
		}
	}

	const workers = 8
	start := make(chan struct{})
	type acquireResult struct {
		lease core.WorkspaceLease
		ok    bool
		err   error
	}
	results := make(chan acquireResult, workers)
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			lease, ok, err := store.WorkspaceLeases().AcquireAvailableLease(ctx, repo.ID, time.Now().UTC(), time.Hour)
			results <- acquireResult{lease: lease, ok: ok, err: err}
		}()
	}
	close(start)
	wg.Wait()
	close(results)

	acquired := make(map[string]core.WorkspaceLease)
	for result := range results {
		if result.err != nil {
			t.Fatalf("acquire lease: %v", result.err)
		}
		if !result.ok {
			continue
		}
		if _, exists := acquired[result.lease.ID]; exists {
			t.Fatalf("lease %s was acquired more than once", result.lease.ID)
		}
		acquired[result.lease.ID] = result.lease
	}
	if len(acquired) != len(seedLeases) {
		t.Fatalf("acquired lease count: got %d want %d", len(acquired), len(seedLeases))
	}
	for _, lease := range acquired {
		if lease.State != core.WorkspaceLeased || lease.ExpiresAt == nil {
			t.Fatalf("acquired lease state: got %+v", lease)
		}
	}

	released, err := store.WorkspaceLeases().ReleaseLease(ctx, "lease-001", storage.LeaseRelease{
		OperationState: core.OperationSucceeded,
	})
	if err != nil {
		t.Fatalf("release successful lease: %v", err)
	}
	if released.State != core.WorkspaceAvailable || released.ExpiresAt != nil {
		t.Fatalf("released clean lease: got %+v", released)
	}

	errored, err := store.WorkspaceLeases().ReleaseLease(ctx, "lease-002", storage.LeaseRelease{
		OperationState: core.OperationFailed,
	})
	if err != nil {
		t.Fatalf("release failed lease: %v", err)
	}
	if errored.State != core.WorkspaceError || errored.ExpiresAt != nil {
		t.Fatalf("released failed lease: got %+v", errored)
	}

	reacquired, ok, err := store.WorkspaceLeases().AcquireAvailableLease(ctx, repo.ID, time.Now().UTC(), time.Hour)
	if err != nil {
		t.Fatalf("reacquire lease: %v", err)
	}
	if !ok || reacquired.ID != "lease-001" {
		t.Fatalf("reacquired lease: got %+v ok=%v want lease-001", reacquired, ok)
	}
	dirty, err := store.WorkspaceLeases().ReleaseLease(ctx, reacquired.ID, storage.LeaseRelease{
		OperationState: core.OperationSucceeded,
		Dirty:          true,
	})
	if err != nil {
		t.Fatalf("release dirty lease: %v", err)
	}
	if dirty.State != core.WorkspaceDirty || dirty.ExpiresAt != nil {
		t.Fatalf("released dirty lease: got %+v", dirty)
	}

	listed, err := store.WorkspaceLeases().ListLeases(ctx, repo.ID)
	if err != nil {
		t.Fatalf("list leases: %v", err)
	}
	if len(listed) != len(seedLeases) {
		t.Fatalf("listed leases: got %d want %d", len(listed), len(seedLeases))
	}
}

func TestPostgresBranchEventStorePersistsWatchCursorAndDedupesDelivery(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	// Store instance A: apply migration and register a branch watch
	storeA, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open storeA: %v", err)
	}
	defer storeA.Close()

	testID := time.Now().UnixNano()
	suffix := strconv.FormatInt(testID, 10)

	repoID := "pg-test-repo"
	branch := "main"
	provider := "forgejo"
	watchID := "watch-pg-test"

	watch := core.BranchWatch{
		ID:        watchID,
		Provider:  provider,
		RepoID:    repoID,
		Branch:    branch,
		CreatedAt: time.Now().UTC(),
	}
	upserted, err := storeA.BranchWatches().UpsertBranchWatch(ctx, watch)
	if err != nil {
		t.Fatalf("upsert watch: %v", err)
	}
	if upserted.ID != watchID {
		t.Fatalf("upserted watch id: got %q want %q", upserted.ID, watchID)
	}

	// Store instance B: simulates restart, reads watch from same DB
	storeB, err := storage.NewPgStore(ctx, dsn, "")
	if err != nil {
		t.Fatalf("open storeB: %v", err)
	}
	defer storeB.Close()

	found, ok, err := storeB.BranchWatches().FindBranchWatch(ctx, provider, repoID, branch)
	if err != nil {
		t.Fatalf("find watch: %v", err)
	}
	if !ok {
		t.Fatal("expected watch to be found after restart")
	}
	if found.RepoID != repoID || found.Branch != branch {
		t.Fatalf("found watch: %+v", found)
	}

	// Upsert revision cursor via storeA, read back via storeB
	cursor := core.RevisionCursor{
		RepoID:     repoID,
		Branch:     branch,
		Revision:   "abc123",
		ObservedAt: time.Now().UTC(),
	}
	if err := storeA.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
		t.Fatalf("upsert cursor: %v", err)
	}

	got, ok, err := storeB.RevisionCursors().GetRevisionCursor(ctx, repoID, branch)
	if err != nil {
		t.Fatalf("get cursor: %v", err)
	}
	if !ok {
		t.Fatal("expected cursor to be found")
	}
	if got.Revision != "abc123" {
		t.Fatalf("cursor revision: got %q want %q", got.Revision, "abc123")
	}

	// Delivery deduplication: same dedupe_key should return false on second call
	dedupeKey := "delivery:test-delivery-unique"
	delivery := core.ProviderDelivery{
		ID:         "pg-delivery-unique-" + suffix,
		Provider:   provider,
		DeliveryID: "test-delivery-unique-" + suffix,
		DedupeKey:  dedupeKey + "-" + suffix,
		EventID:    "event-unique",
		RepoID:     repoID,
		Branch:     branch,
		Revision:   "abc123",
		CreatedAt:  time.Now().UTC(),
	}

	firstResult, err := storeA.ProviderDeliveries().RecordOnce(ctx, delivery)
	if err != nil {
		t.Fatalf("record first delivery: %v", err)
	}
	if !firstResult.First {
		t.Fatal("expected first delivery to return First=true")
	}

	// Second attempt with same dedupe key via storeB — must be duplicate with consistent metadata.
	secondResult, err := storeB.ProviderDeliveries().RecordOnce(ctx, delivery)
	if err != nil {
		t.Fatalf("record second delivery: %v", err)
	}
	if secondResult.First {
		t.Fatal("expected duplicate delivery to return First=false")
	}
	if secondResult.ExistingEventID != delivery.EventID {
		t.Fatalf("duplicate ExistingEventID: got %q want %q", secondResult.ExistingEventID, delivery.EventID)
	}
	if !secondResult.ExistingCreatedAt.Equal(delivery.CreatedAt.Truncate(time.Microsecond)) {
		t.Fatalf("duplicate ExistingCreatedAt: got %v want %v", secondResult.ExistingCreatedAt, delivery.CreatedAt.Truncate(time.Microsecond))
	}
}

func TestPostgresMigrationDropsLegacyRevisionCursorRepoFK(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	legacyPool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatalf("open legacy pool: %v", err)
	}
	defer legacyPool.Close()

	_, err = legacyPool.Exec(ctx, `
		CREATE TABLE repos (
			id TEXT PRIMARY KEY
		);
		CREATE TABLE revision_cursors (
			repo_id TEXT NOT NULL REFERENCES repos(id),
			branch TEXT NOT NULL,
			revision TEXT NOT NULL,
			observed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			PRIMARY KEY (repo_id, branch)
		);
	`)
	if err != nil {
		t.Fatalf("create legacy schema: %v", err)
	}

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}
	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	cursor := core.RevisionCursor{
		RepoID:     "provider-repo-without-repos-row",
		Branch:     "develop",
		Revision:   "abc123",
		ObservedAt: time.Now().UTC(),
	}
	if err := store.RevisionCursors().UpsertRevisionCursor(ctx, cursor); err != nil {
		t.Fatalf("upsert cursor after legacy FK migration: %v", err)
	}
}

func TestPostgresOperationStoreLifecycleAndIdempotency(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	if store.Operations() == nil {
		t.Fatal("Operations() must not return nil")
	}

	repo := core.Repo{
		ID:            "op-repo",
		Name:          "Op Repo",
		RemoteURL:     "ssh://git.example.invalid/toki/op-repo.git",
		DefaultBranch: "main",
		WorkspaceRoot: "/workspaces/op-repo",
	}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	ops := store.Operations()

	// Create with idempotency key.
	op := core.Operation{
		ID:             "op-001",
		RepoID:         repo.ID,
		Type:           core.OperationClone,
		State:          core.OperationQueued,
		IdempotencyKey: "idem-clone-1",
		CreatedBy:      "agent-shell",
	}
	created, err := ops.CreateOperation(ctx, op)
	if err != nil {
		t.Fatalf("create operation: %v", err)
	}
	if !created.Created {
		t.Fatal("first create must report Created=true")
	}
	if created.Operation.State != core.OperationQueued {
		t.Fatalf("created state: got %q want %q", created.Operation.State, core.OperationQueued)
	}

	// Duplicate idempotency key (different id) returns the existing operation.
	dup := op
	dup.ID = "op-001-retry"
	dupResult, err := ops.CreateOperation(ctx, dup)
	if err != nil {
		t.Fatalf("duplicate idempotency create: %v", err)
	}
	if dupResult.Created {
		t.Fatal("duplicate idempotency create must report Created=false")
	}
	if dupResult.Operation.ID != op.ID {
		t.Fatalf("duplicate idempotency returned id %q want %q", dupResult.Operation.ID, op.ID)
	}

	// id collision without idempotency key is a hard duplicate.
	noKey := core.Operation{
		ID:        "op-002",
		RepoID:    repo.ID,
		Type:      core.OperationFetch,
		State:     core.OperationQueued,
		CreatedBy: "agent-shell",
	}
	if _, err := ops.CreateOperation(ctx, noKey); err != nil {
		t.Fatalf("create no-key operation: %v", err)
	}
	if _, err := ops.CreateOperation(ctx, noKey); !errors.Is(err, storage.ErrOperationAlreadyExists) {
		t.Fatalf("id collision error: got %v want %v", err, storage.ErrOperationAlreadyExists)
	}

	// Same id as an existing operation but a different idempotency key collides on
	// the primary key, not the idempotency index, so it is a duplicate create.
	sameIDOtherKey := op
	sameIDOtherKey.IdempotencyKey = "idem-clone-2"
	if _, err := ops.CreateOperation(ctx, sameIDOtherKey); !errors.Is(err, storage.ErrOperationAlreadyExists) {
		t.Fatalf("same id different idempotency key error: got %v want %v", err, storage.ErrOperationAlreadyExists)
	}

	// queued -> running -> succeeded.
	now := time.Now().UTC()
	running, err := ops.StartOperation(ctx, op.ID, now)
	if err != nil {
		t.Fatalf("start operation: %v", err)
	}
	if running.State != core.OperationRunning {
		t.Fatalf("started state: got %q want %q", running.State, core.OperationRunning)
	}
	succeeded, err := ops.SucceedOperation(ctx, op.ID, now.Add(time.Second))
	if err != nil {
		t.Fatalf("succeed operation: %v", err)
	}
	if succeeded.State != core.OperationSucceeded {
		t.Fatalf("succeeded state: got %q want %q", succeeded.State, core.OperationSucceeded)
	}

	// succeeded -> failed is an invalid transition.
	if _, err := ops.FailOperation(ctx, op.ID, now.Add(2*time.Second)); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("invalid transition error: got %v want %v", err, storage.ErrInvalidOperation)
	}

	// starting an already-terminal operation is also invalid.
	if _, err := ops.StartOperation(ctx, op.ID, now); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("start terminal error: got %v want %v", err, storage.ErrInvalidOperation)
	}

	// GetOperation and not-found path.
	got, err := ops.GetOperation(ctx, op.ID)
	if err != nil {
		t.Fatalf("get operation: %v", err)
	}
	if got.State != core.OperationSucceeded {
		t.Fatalf("get operation state: got %q want %q", got.State, core.OperationSucceeded)
	}
	if _, err := ops.GetOperation(ctx, "missing"); !errors.Is(err, storage.ErrOperationNotFound) {
		t.Fatalf("get missing error: got %v want %v", err, storage.ErrOperationNotFound)
	}

	// List ordering is stable (created_at, id).
	listed, err := ops.ListOperations(ctx, repo.ID)
	if err != nil {
		t.Fatalf("list operations: %v", err)
	}
	if len(listed) != 2 {
		t.Fatalf("listed operations: got %d want 2", len(listed))
	}
	for i := 1; i < len(listed); i++ {
		prev, cur := listed[i-1], listed[i]
		if prev.CreatedAt.After(cur.CreatedAt) ||
			(prev.CreatedAt.Equal(cur.CreatedAt) && prev.ID > cur.ID) {
			t.Fatalf("list ordering unstable at %d: %+v then %+v", i, prev, cur)
		}
	}
}

func TestPostgresOperationStorePickQueuedOperationDoesNotDuplicate(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	ops := store.Operations()

	// 1. empty state pick returns ok=false
	emptyOp, ok, err := ops.PickQueuedOperation(ctx, time.Now())
	if err != nil {
		t.Fatalf("pick empty: %v", err)
	}
	if ok {
		t.Fatalf("expected ok=false on empty store, got operation: %+v", emptyOp)
	}

	// 2. create repo
	repo := core.Repo{
		ID:            "pick-repo",
		Name:          "Pick Repo",
		RemoteURL:     "ssh://git.example.invalid/toki/pick-repo.git",
		DefaultBranch: "main",
		WorkspaceRoot: "/workspaces/pick-repo",
	}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	// 3. Seed operations
	const seedCount = 10
	for i := 0; i < seedCount; i++ {
		op := core.Operation{
			ID:             fmt.Sprintf("op-seed-%d", i),
			RepoID:         repo.ID,
			Type:           core.OperationClone,
			State:          core.OperationQueued,
			IdempotencyKey: fmt.Sprintf("idem-seed-%d", i),
			CreatedBy:      "test-picker",
		}
		if _, err := ops.CreateOperation(ctx, op); err != nil {
			t.Fatalf("create op %d: %v", i, err)
		}
	}

	// 4. Concurrent pickers
	const workerCount = 20
	start := make(chan struct{})

	type pickResult struct {
		op  core.Operation
		ok  bool
		err error
	}
	results := make(chan pickResult, workerCount)
	var wg sync.WaitGroup

	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			op, ok, err := ops.PickQueuedOperation(ctx, time.Now())
			results <- pickResult{op: op, ok: ok, err: err}
		}()
	}

	close(start)
	wg.Wait()
	close(results)

	// 5. Verify results
	pickedIDs := make(map[string]bool)
	successCount := 0

	for res := range results {
		if res.err != nil {
			t.Errorf("PickQueuedOperation returned error: %v", res.err)
			continue
		}
		if !res.ok {
			continue
		}
		successCount++
		if pickedIDs[res.op.ID] {
			t.Errorf("operation %s was picked more than once!", res.op.ID)
		}
		pickedIDs[res.op.ID] = true

		// Also check state is changed to Running
		if res.op.State != core.OperationRunning {
			t.Errorf("picked operation state is %q, expected %q", res.op.State, core.OperationRunning)
		}
	}

	if successCount != seedCount {
		t.Errorf("success picked count: got %d want %d", successCount, seedCount)
	}
}

func assertJSONPayload(t *testing.T, actual []byte, expected string) {
	t.Helper()
	var actMap, expMap map[string]any
	if err := json.Unmarshal(actual, &actMap); err != nil {
		t.Fatalf("unmarshal actual payload %q: %v", string(actual), err)
	}
	if err := json.Unmarshal([]byte(expected), &expMap); err != nil {
		t.Fatalf("unmarshal expected payload %q: %v", expected, err)
	}
	if !reflect.DeepEqual(actMap, expMap) {
		t.Errorf("payload mismatch:\ngot:  %+v\nwant: %+v", actMap, expMap)
	}
}

func TestPostgresOperationEventStoreAppendsListsMarksPublishedAndValidates(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	if store.OperationEvents() == nil {
		t.Fatal("OperationEvents() must not return nil")
	}

	repo := core.Repo{
		ID:            "event-repo",
		Name:          "Event Repo",
		RemoteURL:     "ssh://git.example.invalid/toki/event-repo.git",
		DefaultBranch: "main",
		WorkspaceRoot: "/workspaces/event-repo",
	}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	op := core.Operation{
		ID:             "op-event-001",
		RepoID:         repo.ID,
		Type:           core.OperationClone,
		State:          core.OperationQueued,
		IdempotencyKey: "idem-event-1",
		CreatedBy:      "test-event-appender",
	}
	if _, err := store.Operations().CreateOperation(ctx, op); err != nil {
		t.Fatalf("create operation: %v", err)
	}

	eventStore := store.OperationEvents()

	created1 := time.Now().UTC().Truncate(time.Microsecond)
	created2 := created1.Add(time.Second)
	subject := "operation:" + op.ID
	startedPayload := fmt.Sprintf(
		`{"operation_id":%q,"repo_id":%q,"type":%q,"state":%q,"started_at":%q}`,
		op.ID,
		repo.ID,
		op.Type,
		core.OperationRunning,
		created1.Format(time.RFC3339Nano),
	)
	completedPayload := fmt.Sprintf(
		`{"operation_id":%q,"repo_id":%q,"type":%q,"state":%q,"completed_at":%q}`,
		op.ID,
		repo.ID,
		op.Type,
		core.OperationSucceeded,
		created2.Format(time.RFC3339Nano),
	)

	// Append two events
	ev1 := storage.OperationEvent{
		OperationID: op.ID,
		Event: events.Event{
			ID:        "evt-001",
			Type:      events.OperationStarted,
			Subject:   subject,
			Payload:   []byte(startedPayload),
			CreatedAt: created1,
		},
	}
	ev2 := storage.OperationEvent{
		OperationID: op.ID,
		Event: events.Event{
			ID:        "evt-002",
			Type:      events.OperationCompleted,
			Subject:   subject,
			Payload:   []byte(completedPayload),
			CreatedAt: created2,
		},
	}

	if err := eventStore.AppendEvent(ctx, ev1); err != nil {
		t.Fatalf("append event 1: %v", err)
	}
	if err := eventStore.AppendEvent(ctx, ev2); err != nil {
		t.Fatalf("append event 2: %v", err)
	}

	// Append third event to test empty/nil payload normalization to {}
	ev3 := storage.OperationEvent{
		OperationID: op.ID,
		Event: events.Event{
			ID:        "evt-003",
			Type:      events.OperationCompleted,
			Subject:   subject,
			Payload:   nil,
			CreatedAt: created2.Add(time.Second),
		},
	}
	if err := eventStore.AppendEvent(ctx, ev3); err != nil {
		t.Fatalf("append event 3 (nil payload): %v", err)
	}

	// List events and check ordering
	listed, err := eventStore.ListEvents(ctx, op.ID)
	if err != nil {
		t.Fatalf("list events: %v", err)
	}
	if len(listed) != 3 {
		t.Fatalf("expected 3 listed events, got %d", len(listed))
	}

	// Verify details of each event (Round-trip)
	// Event 1
	if listed[0].OperationID != op.ID {
		t.Errorf("ev1: operationID mismatch: got %q want %q", listed[0].OperationID, op.ID)
	}
	if listed[0].Event.ID != "evt-001" {
		t.Errorf("ev1: eventID mismatch: got %q want %q", listed[0].Event.ID, "evt-001")
	}
	if listed[0].Event.Type != events.OperationStarted {
		t.Errorf("ev1: type mismatch: got %q want %q", listed[0].Event.Type, events.OperationStarted)
	}
	if listed[0].Event.Subject != subject {
		t.Errorf("ev1: subject mismatch: got %q want %q", listed[0].Event.Subject, subject)
	}
	assertJSONPayload(t, listed[0].Event.Payload, startedPayload)
	if !listed[0].Event.CreatedAt.Equal(created1) {
		t.Errorf("ev1: CreatedAt mismatch: got %v want %v", listed[0].Event.CreatedAt, created1)
	}
	if listed[0].PublishedAt != nil {
		t.Errorf("ev1: PublishedAt must be nil initially")
	}

	// Event 2
	if listed[1].OperationID != op.ID {
		t.Errorf("ev2: operationID mismatch")
	}
	if listed[1].Event.ID != "evt-002" {
		t.Errorf("ev2: eventID mismatch")
	}
	if listed[1].Event.Subject != subject {
		t.Errorf("ev2: subject mismatch: got %q want %q", listed[1].Event.Subject, subject)
	}
	assertJSONPayload(t, listed[1].Event.Payload, completedPayload)
	if !listed[1].Event.CreatedAt.Equal(created2) {
		t.Errorf("ev2: CreatedAt mismatch")
	}

	// Event 3 (nil payload normalization check)
	if listed[2].Event.ID != "evt-003" {
		t.Errorf("ev3: eventID mismatch")
	}
	assertJSONPayload(t, listed[2].Event.Payload, `{}`)

	// List pending events before publish
	pending, err := eventStore.ListPendingEvents(ctx, 10)
	if err != nil {
		t.Fatalf("list pending: %v", err)
	}
	if len(pending) != 3 {
		t.Fatalf("expected 3 pending events, got %d", len(pending))
	}

	// Mark first event as published
	now := time.Now().UTC().Truncate(time.Microsecond)
	marked, err := eventStore.MarkPublished(ctx, "evt-001", now)
	if err != nil {
		t.Fatalf("mark published: %v", err)
	}
	if marked.PublishedAt == nil {
		t.Fatal("expected non-nil PublishedAt after mark published")
	}
	if !marked.PublishedAt.Equal(now) {
		t.Fatalf("expected PublishedAt to be %v, got %v", now, *marked.PublishedAt)
	}

	// List pending events after publish
	pendingAfter, err := eventStore.ListPendingEvents(ctx, 10)
	if err != nil {
		t.Fatalf("list pending after: %v", err)
	}
	if len(pendingAfter) != 2 {
		t.Fatalf("expected 2 pending events, got %d", len(pendingAfter))
	}
	if pendingAfter[0].Event.ID != "evt-002" || pendingAfter[1].Event.ID != "evt-003" {
		t.Fatalf("unexpected pending events after publish")
	}

	// Invalid input coverage
	// 1. Invalid payload (array)
	badPayloadEv := storage.OperationEvent{
		OperationID: op.ID,
		Event: events.Event{
			ID:        "evt-bad",
			Type:      events.OperationStarted,
			Subject:   "clone",
			Payload:   []byte(`[1]`),
			CreatedAt: time.Now().UTC(),
		},
	}
	if err := eventStore.AppendEvent(ctx, badPayloadEv); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for invalid JSON array payload, got %v", err)
	}

	// 2. Invalid payload (string)
	badPayloadEv2 := badPayloadEv
	badPayloadEv2.Event.Payload = []byte(`"abc"`)
	if err := eventStore.AppendEvent(ctx, badPayloadEv2); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for invalid JSON string payload, got %v", err)
	}

	// 3. Invalid payload (null)
	badPayloadEv3 := badPayloadEv
	badPayloadEv3.Event.Payload = []byte(`null`)
	if err := eventStore.AppendEvent(ctx, badPayloadEv3); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for invalid JSON null payload, got %v", err)
	}

	// 4. Missing fields
	emptyOpIDEv := ev1
	emptyOpIDEv.OperationID = ""
	emptyOpIDEv.Event.ID = "empty-op-id-test-event"
	if err := eventStore.AppendEvent(ctx, emptyOpIDEv); err != nil {
		t.Fatalf("expected success for empty operation ID, got %v", err)
	}

	// Verify the empty-op-id event is pending and can be read properly
	pendingWithEmpty, err := eventStore.ListPendingEvents(ctx, 10)
	if err != nil {
		t.Fatalf("list pending with empty op: %v", err)
	}
	foundEmpty := false
	for _, pe := range pendingWithEmpty {
		if pe.Event.ID == "empty-op-id-test-event" {
			foundEmpty = true
			if pe.OperationID != "" {
				t.Errorf("expected empty OperationID, got %q", pe.OperationID)
			}
		}
	}
	if !foundEmpty {
		t.Fatal("expected empty-op-id-test-event in pending list, but not found")
	}

	// Mark the empty-op event as published
	markedEmpty, err := eventStore.MarkPublished(ctx, "empty-op-id-test-event", now)
	if err != nil {
		t.Fatalf("mark published empty op: %v", err)
	}
	if markedEmpty.PublishedAt == nil || !markedEmpty.PublishedAt.Equal(now) {
		t.Errorf("failed to mark empty operation event as published")
	}

	emptyEventIDEv := ev1
	emptyEventIDEv.Event.ID = ""
	if err := eventStore.AppendEvent(ctx, emptyEventIDEv); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for empty event ID, got %v", err)
	}

	emptyTypeEv := ev1
	emptyTypeEv.Event.Type = ""
	if err := eventStore.AppendEvent(ctx, emptyTypeEv); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for empty type, got %v", err)
	}

	emptySubjectEv := ev1
	emptySubjectEv.Event.Subject = ""
	if err := eventStore.AppendEvent(ctx, emptySubjectEv); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for empty subject, got %v", err)
	}

	// 5. Non-positive limit for pending list
	if _, err := eventStore.ListPendingEvents(ctx, 0); !errors.Is(err, storage.ErrInvalidOperation) {
		t.Fatalf("expected ErrInvalidOperation for non-positive limit, got %v", err)
	}

	// 6. Missing mark ID
	if _, err := eventStore.MarkPublished(ctx, "evt-missing", time.Now()); !errors.Is(err, storage.ErrOperationNotFound) {
		t.Fatalf("expected ErrOperationNotFound for missing mark ID, got %v", err)
	}
}

func TestPostgresOperationStorePickQueuedAgentRunOperationFiltersType(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	repo := core.Repo{ID: "repo-picker-test", Name: "PickerTest", RemoteURL: "https://example.invalid/r.git", DefaultBranch: "main", WorkspaceRoot: "/ws", CredentialRef: "ref:x"}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	now := time.Now().UTC()

	// Enqueue a non-agent_run operation (clone).
	cloneOp := core.Operation{ID: "op-clone-1", RepoID: repo.ID, Type: core.OperationClone, State: core.OperationQueued, CreatedBy: "test"}
	if _, err := store.Operations().CreateOperation(ctx, cloneOp); err != nil {
		t.Fatalf("create clone op: %v", err)
	}

	// Typed pick should not return the clone operation.
	picked, ok, err := store.Operations().PickQueuedAgentRunOperation(ctx, now)
	if err != nil {
		t.Fatalf("pick agent_run: %v", err)
	}
	if ok {
		t.Fatalf("expected no agent_run to pick, got %+v", picked)
	}

	// Generic pick should still return the clone operation.
	genPicked, genOk, err := store.Operations().PickQueuedOperation(ctx, now)
	if err != nil {
		t.Fatalf("generic pick: %v", err)
	}
	if !genOk {
		t.Fatal("expected generic pick to return the clone op")
	}
	if genPicked.Type != core.OperationClone {
		t.Fatalf("generic pick type: got %q want %q", genPicked.Type, core.OperationClone)
	}

	// Now enqueue an agent_run operation.
	agentOp := core.Operation{ID: "op-agent-1", RepoID: repo.ID, Type: core.OperationAgentRun, State: core.OperationQueued, CreatedBy: "test"}
	if _, err := store.Operations().CreateOperation(ctx, agentOp); err != nil {
		t.Fatalf("create agent_run op: %v", err)
	}

	// Typed pick should return only the agent_run.
	agentPicked, agentOk, err := store.Operations().PickQueuedAgentRunOperation(ctx, now)
	if err != nil {
		t.Fatalf("pick agent_run after enqueue: %v", err)
	}
	if !agentOk {
		t.Fatal("expected agent_run to be picked")
	}
	if agentPicked.Type != core.OperationAgentRun {
		t.Fatalf("picked type: got %q want %q", agentPicked.Type, core.OperationAgentRun)
	}
	if agentPicked.State != core.OperationRunning {
		t.Fatalf("picked state: got %q want running", agentPicked.State)
	}

	// Second typed pick should return nothing (already running).
	_, ok2, err := store.Operations().PickQueuedAgentRunOperation(ctx, now)
	if err != nil {
		t.Fatalf("second pick: %v", err)
	}
	if ok2 {
		t.Fatal("expected no second agent_run to pick (already running)")
	}
}

func TestPostgresAgentRunInputStoreCreateAndGet(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	repo := core.Repo{ID: "repo-input-test", Name: "InputTest", RemoteURL: "https://example.invalid/r.git", DefaultBranch: "main", WorkspaceRoot: "/ws", CredentialRef: "ref:x"}
	if err := store.Repos().CreateRepo(ctx, repo); err != nil {
		t.Fatalf("create repo: %v", err)
	}

	op := core.Operation{ID: "op-input-1", RepoID: repo.ID, Type: core.OperationAgentRun, State: core.OperationQueued, CreatedBy: "test"}
	if _, err := store.Operations().CreateOperation(ctx, op); err != nil {
		t.Fatalf("create op: %v", err)
	}

	now := time.Now().UTC().Truncate(time.Millisecond)

	policy := map[string]any{"commit_push": true, "branch": "main"}

	input := core.DurableAgentRunInput{
		OperationID:      op.ID,
		RepoID:           repo.ID,
		Branch:           "main",
		WorkspacePath:    "/workspaces/repo-input-test/slot-0",
		Instruction:      "run the linter",
		PolicyContext:    policy,
		ExpectedRevision: "abc123",
		CreatedAt:        now,
	}

	if err := store.AgentRunInputs().CreateAgentRunInput(ctx, input); err != nil {
		t.Fatalf("create agent run input: %v", err)
	}

	got, err := store.AgentRunInputs().GetAgentRunInput(ctx, op.ID)
	if err != nil {
		t.Fatalf("get agent run input: %v", err)
	}

	if got.OperationID != input.OperationID {
		t.Errorf("operation_id: got %q want %q", got.OperationID, input.OperationID)
	}
	if got.RepoID != input.RepoID {
		t.Errorf("repo_id: got %q want %q", got.RepoID, input.RepoID)
	}
	if got.Branch != input.Branch {
		t.Errorf("branch: got %q want %q", got.Branch, input.Branch)
	}
	if got.WorkspacePath != input.WorkspacePath {
		t.Errorf("workspace_path: got %q want %q", got.WorkspacePath, input.WorkspacePath)
	}
	if got.Instruction != input.Instruction {
		t.Errorf("instruction: got %q want %q", got.Instruction, input.Instruction)
	}
	if got.ExpectedRevision != input.ExpectedRevision {
		t.Errorf("expected_revision: got %q want %q", got.ExpectedRevision, input.ExpectedRevision)
	}
	gotJSON, _ := json.Marshal(got.PolicyContext)
	wantJSON, _ := json.Marshal(input.PolicyContext)
	if string(gotJSON) != string(wantJSON) {
		t.Errorf("policy_context: got %s want %s", gotJSON, wantJSON)
	}

	// Validation: missing required fields (empty struct).
	if err := store.AgentRunInputs().CreateAgentRunInput(ctx, core.DurableAgentRunInput{}); !errors.Is(err, storage.ErrInvalidAgentRunInput) {
		t.Fatalf("expected ErrInvalidAgentRunInput for empty input, got %v", err)
	}

	// Validation: blank instruction only (all other fields present).
	op2 := core.Operation{ID: "op-input-blank-instr", RepoID: repo.ID, Type: core.OperationAgentRun, State: core.OperationQueued, CreatedBy: "test"}
	if _, err := store.Operations().CreateOperation(ctx, op2); err != nil {
		t.Fatalf("create op2: %v", err)
	}
	blankInstr := core.DurableAgentRunInput{
		OperationID:      op2.ID,
		RepoID:           repo.ID,
		Branch:           "main",
		WorkspacePath:    "/workspaces/repo-input-test/slot-1",
		Instruction:      "   ",
		PolicyContext:    nil,
		ExpectedRevision: "abc123",
	}
	if err := store.AgentRunInputs().CreateAgentRunInput(ctx, blankInstr); !errors.Is(err, storage.ErrInvalidAgentRunInput) {
		t.Fatalf("expected ErrInvalidAgentRunInput for blank instruction, got %v", err)
	}

	// Not found.
	if _, err := store.AgentRunInputs().GetAgentRunInput(ctx, "nonexistent-op"); !errors.Is(err, storage.ErrAgentRunInputNotFound) {
		t.Fatalf("expected ErrAgentRunInputNotFound, got %v", err)
	}
}

func TestPostgresWebhookDeliveryStoreRecordsRetryState(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	dStore := store.WebhookDeliveries()

	// 1. first enqueue creates pending
	del := core.WebhookDelivery{
		ID:             "delivery-1",
		EventID:        "event-1",
		SubscriptionID: "sub-1",
		Status:         core.WebhookDeliveryPending,
	}
	res1, err := dStore.EnqueueDelivery(ctx, del)
	if err != nil {
		t.Fatalf("first enqueue: %v", err)
	}
	if !res1.Created {
		t.Fatalf("expected Created = true for first enqueue")
	}
	if res1.Delivery.Status != core.WebhookDeliveryPending {
		t.Fatalf("expected status pending, got %s", res1.Delivery.Status)
	}

	// 2. duplicate enqueue returns existing/no duplicate
	res2, err := dStore.EnqueueDelivery(ctx, del)
	if err != nil {
		t.Fatalf("second enqueue: %v", err)
	}
	if res2.Created {
		t.Fatalf("expected Created = false for duplicate enqueue")
	}
	if res2.Delivery.ID != del.ID {
		t.Fatalf("expected return existing delivery ID %s, got %s", del.ID, res2.Delivery.ID)
	}

	// 3. pick moves to sending
	picked, err := dStore.PickPendingDeliveries(ctx, 10, time.Now().Add(time.Minute))
	if err != nil {
		t.Fatalf("pick pending: %v", err)
	}
	if len(picked) != 1 {
		t.Fatalf("expected 1 picked delivery, got %d", len(picked))
	}
	if picked[0].ID != del.ID {
		t.Fatalf("expected picked ID %s, got %s", del.ID, picked[0].ID)
	}
	if picked[0].Status != core.WebhookDeliverySending {
		t.Fatalf("expected status to transition to sending, got %s", picked[0].Status)
	}

	// Double check database state is updated to 'sending'
	dbDel, err := dStore.GetDelivery(ctx, del.ID)
	if err != nil {
		t.Fatalf("get delivery: %v", err)
	}
	if dbDel.Status != core.WebhookDeliverySending {
		t.Fatalf("expected DB status to be sending, got %s", dbDel.Status)
	}

	// 3b. StartDelivery id-scoped transition: enqueue a second delivery and verify
	// that StartDelivery(id) only claims the target delivery, not another.
	//
	// Not-yet-due check: delivery-B's NextAttemptAt is set by EnqueueDelivery to
	// approximately `now`. Passing an earlier `now` makes the delivery not-due
	// because next_attempt_at > now, so StartDelivery returns ok=false.
	now := time.Now().UTC()
	earlyNow := now.Add(-time.Hour)

	delB := core.WebhookDelivery{
		ID:             "delivery-B",
		EventID:        "event-B",
		SubscriptionID: "sub-B",
		Status:         core.WebhookDeliveryPending,
	}
	resB, err := dStore.EnqueueDelivery(ctx, delB)
	if err != nil {
		t.Fatalf("enqueue delivery-B: %v", err)
	}
	if !resB.Created {
		t.Fatalf("expected delivery-B to be created")
	}

	// delivery-B's NextAttemptAt ≈ now, but we pass earlyNow (now-1h).
	// Since next_attempt_at (≈now) > earlyNow, delivery-B is NOT due.
	startedB, okB, err := dStore.StartDelivery(ctx, delB.ID, earlyNow)
	if err != nil {
		t.Fatalf("start delivery-B (not due): %v", err)
	}
	if okB {
		t.Fatal("expected StartDelivery for not-yet-due delivery-B to return ok=false")
	}
	if startedB.ID != "" {
		t.Fatalf("expected zero delivery for not-yet-due StartDelivery, got %s", startedB.ID)
	}

	// 3c. Verify id-scoped transition: make delivery-B due and confirm only B
	// transitions to sending, while the already-sending delivery-A stays unchanged.
	_, err = dStore.RecordOutcome(ctx, delB.ID, core.WebhookDeliveryRetryable, 500, "simulated failure", now.Add(-time.Minute), now)
	if err != nil {
		t.Fatalf("make delivery-B due via RecordOutcome: %v", err)
	}

	// delivery-A is already in sending state; verify StartDelivery(delA.ID, now)
	// returns ok=false (already sending).
	_, okExisting, err := dStore.StartDelivery(ctx, del.ID, now)
	if err != nil {
		t.Fatalf("start already-sending delivery: %v", err)
	}
	if okExisting {
		t.Fatal("expected StartDelivery for already-sending delivery-A to return ok=false")
	}

	// delivery-B is now due (next_attempt_at = now - 1 minute).
	// StartDelivery(delB.ID, now) should claim only delivery-B.
	startedB, okB, err = dStore.StartDelivery(ctx, delB.ID, now)
	if err != nil {
		t.Fatalf("start delivery-B (now due): %v", err)
	}
	if !okB {
		t.Fatal("expected StartDelivery for due delivery-B to return ok=true")
	}
	if startedB.ID != "delivery-B" {
		t.Fatalf("expected claimed delivery ID delivery-B, got %s", startedB.ID)
	}
	if startedB.Status != core.WebhookDeliverySending {
		t.Fatalf("expected claimed delivery status sending, got %s", startedB.Status)
	}

	// Verify delivery-A is still in sending state (unchanged by StartDelivery on B).
	dbDelA, err := dStore.GetDelivery(ctx, del.ID)
	if err != nil {
		t.Fatalf("get delivery-A: %v", err)
	}
	if dbDelA.Status != core.WebhookDeliverySending {
		t.Fatalf("expected delivery-A status to remain sending, got %s", dbDelA.Status)
	}

	// 4. retryable record increments attempt count and sets next_attempt_at
	nextAttempt := time.Now().Add(time.Hour).Truncate(time.Microsecond)
	outcome, err := dStore.RecordOutcome(ctx, del.ID, core.WebhookDeliveryRetryable, 500, "internal error", nextAttempt, time.Now())
	if err != nil {
		t.Fatalf("record retryable outcome: %v", err)
	}
	if outcome.Status != core.WebhookDeliveryRetryable {
		t.Fatalf("expected status retryable, got %s", outcome.Status)
	}
	if outcome.AttemptCount != 1 {
		t.Fatalf("expected attempt count 1, got %d", outcome.AttemptCount)
	}
	if !outcome.NextAttemptAt.Equal(nextAttempt.UTC()) {
		t.Fatalf("expected next attempt at %v, got %v", nextAttempt.UTC(), outcome.NextAttemptAt)
	}
	if outcome.LastStatusCode != 500 {
		t.Fatalf("expected status code 500, got %d", outcome.LastStatusCode)
	}
	if outcome.LastError != "internal error" {
		t.Fatalf("expected last error 'internal error', got %s", outcome.LastError)
	}

	// 5. terminal failed and succeeded states are persisted
	outcome2, err := dStore.RecordOutcome(ctx, del.ID, core.WebhookDeliverySucceeded, 200, "", time.Time{}, time.Now())
	if err != nil {
		t.Fatalf("record succeeded outcome: %v", err)
	}
	if outcome2.Status != core.WebhookDeliverySucceeded {
		t.Fatalf("expected status succeeded, got %s", outcome2.Status)
	}
	if outcome2.AttemptCount != 2 {
		t.Fatalf("expected attempt count 2, got %d", outcome2.AttemptCount)
	}

	// test no rows behavior of GetDelivery
	_, err = dStore.GetDelivery(ctx, "non-existent")
	if err == nil {
		t.Fatalf("expected error for non-existent delivery")
	}
}

func TestPostgresProviderConfigStoreSavesGetsListsAndValidates(t *testing.T) {
	dsn := os.Getenv("GITO_TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("GITO_TEST_DATABASE_URL not set; skipping Postgres integration test")
	}

	ctx := context.Background()
	dsn = isolatedPostgresDSN(t, ctx, dsn)

	migrationData, err := os.ReadFile("../../migrations/00001_initial.sql")
	if err != nil {
		t.Fatalf("read migration: %v", err)
	}

	store, err := storage.NewPgStore(ctx, dsn, string(migrationData))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	defer store.Close()

	// Verify ProviderConfigs() returns a non-nil store
	if store.ProviderConfigs() == nil {
		t.Fatal("ProviderConfigs() must not return nil")
	}

	s := store.ProviderConfigs()

	// 1. First save: insert
	cfg1 := provider.Config{
		ID:           provider.ProviderGitHub,
		Endpoint:     "https://api.github.com",
		Capabilities: provider.CapabilityChangeRequest | provider.CapabilityWebhook,
	}
	saved, err := s.SaveProviderConfig(ctx, cfg1)
	if err != nil {
		t.Fatalf("SaveProviderConfig first save: %v", err)
	}
	if saved.ID != provider.ProviderGitHub {
		t.Errorf("saved config ID: got %q, want %q", saved.ID, provider.ProviderGitHub)
	}
	if saved.Endpoint != "https://api.github.com" {
		t.Errorf("saved config endpoint: got %q, want %q", saved.Endpoint, "https://api.github.com")
	}
	if saved.Capabilities != (provider.CapabilityChangeRequest | provider.CapabilityWebhook) {
		t.Errorf("saved config capabilities: got %d, want %d", saved.Capabilities, provider.CapabilityChangeRequest|provider.CapabilityWebhook)
	}
	// Verify created_at is set after first save
	if saved.CreatedAt.IsZero() {
		t.Error("created_at should be set after first save")
	}
	if saved.UpdatedAt.IsZero() {
		t.Error("updated_at should be set after first save")
	}

	// 2. Second save: update
	cfg2 := provider.Config{
		ID:            provider.ProviderGitHub,
		Endpoint:      "https://api.github.com/v2",
		CredentialRef: "secret:github-token",
		Capabilities:  provider.CapabilityChangeRequest | provider.CapabilityWebhook | provider.CapabilityChecks,
	}
	saved2, err := s.SaveProviderConfig(ctx, cfg2)
	if err != nil {
		t.Fatalf("SaveProviderConfig second save (update): %v", err)
	}
	if saved2.Endpoint != "https://api.github.com/v2" {
		t.Errorf("updated endpoint: got %q, want %q", saved2.Endpoint, "https://api.github.com/v2")
	}
	if saved2.CredentialRef != "secret:github-token" {
		t.Errorf("updated credential_ref: got %q, want %q", saved2.CredentialRef, "secret:github-token")
	}
	if saved2.Capabilities != (provider.CapabilityChangeRequest | provider.CapabilityWebhook | provider.CapabilityChecks) {
		t.Errorf("updated capabilities: got %d, want %d", saved2.Capabilities, provider.CapabilityChangeRequest|provider.CapabilityWebhook|provider.CapabilityChecks)
	}

	// 3. Get provider config
	got, ok, err := s.GetProviderConfig(ctx, provider.ProviderGitHub)
	if err != nil {
		t.Fatalf("GetProviderConfig: %v", err)
	}
	if !ok {
		t.Fatal("GetProviderConfig should return ok=true for existing config")
	}
	if got.ID != provider.ProviderGitHub {
		t.Errorf("got config ID: got %q, want %q", got.ID, provider.ProviderGitHub)
	}
	if got.Endpoint != "https://api.github.com/v2" {
		t.Errorf("got config endpoint: got %q, want %q", got.Endpoint, "https://api.github.com/v2")
	}
	if got.CredentialRef != "secret:github-token" {
		t.Errorf("got config credential_ref: got %q, want %q", got.CredentialRef, "secret:github-token")
	}
	if got.Capabilities != (provider.CapabilityChangeRequest | provider.CapabilityWebhook | provider.CapabilityChecks) {
		t.Errorf("got config capabilities: got %d, want %d", got.Capabilities, provider.CapabilityChangeRequest|provider.CapabilityWebhook|provider.CapabilityChecks)
	}

	// 4. Get not-found
	_, ok, err = s.GetProviderConfig(ctx, provider.ProviderID("nonexistent"))
	if err != nil {
		t.Fatalf("GetProviderConfig not-found: %v", err)
	}
	if ok {
		t.Error("GetProviderConfig should return ok=false for non-existing config")
	}

	// 5. List provider configs
	configs, err := s.ListProviderConfigs(ctx)
	if err != nil {
		t.Fatalf("ListProviderConfigs: %v", err)
	}
	if len(configs) != 1 {
		t.Errorf("list config count: got %d, want 1", len(configs))
	}

	// 6. Add second config and verify list ordering by id
	cfg3 := provider.Config{
		ID:           provider.ProviderGitLab,
		Endpoint:     "https://gitlab.com/api/v4",
		Capabilities: provider.CapabilityWebhook,
	}
	if _, err := s.SaveProviderConfig(ctx, cfg3); err != nil {
		t.Fatalf("SaveProviderConfig second config: %v", err)
	}

	configs, err = s.ListProviderConfigs(ctx)
	if err != nil {
		t.Fatalf("ListProviderConfigs after second: %v", err)
	}
	if len(configs) != 2 {
		t.Errorf("list config count: got %d, want 2", len(configs))
	}
	// Verify ordered by id (gitlab < github alphabetically)
	if configs[0].ID != provider.ProviderGitLab || configs[1].ID != provider.ProviderGitHub {
		t.Errorf("list order: got [%q, %q], want [%q, %q]", configs[0].ID, configs[1].ID, provider.ProviderGitLab, provider.ProviderGitHub)
	}

	// 7. Validation: empty id
	_, err = s.SaveProviderConfig(ctx, provider.Config{ID: ""})
	if err == nil || err.Error() == "" {
		t.Error("SaveProviderConfig with empty ID should return error")
	}

	// 8. Validation: empty endpoint
	_, err = s.SaveProviderConfig(ctx, provider.Config{ID: provider.ProviderPlane, Endpoint: ""})
	if err == nil {
		t.Error("SaveProviderConfig with empty endpoint should return error")
	}

	// 9. Validation: zero capabilities
	_, err = s.SaveProviderConfig(ctx, provider.Config{ID: provider.ProviderPlane, Endpoint: "https://example.com", Capabilities: 0})
	if err == nil {
		t.Error("SaveProviderConfig with zero capabilities should return error")
	}

	// 10. CredentialRef is stored as credential_ref column value (not the raw secret)
	// The CredentialRef field is stored as-is (it's a reference, not a value).
	// Verify the stored config contains the ref but no raw secret value.
	gotCred, ok, err := s.GetProviderConfig(ctx, provider.ProviderGitHub)
	if err != nil || !ok {
		t.Fatalf("GetProviderConfig for credential check: ok=%v, err=%v", ok, err)
	}
	if gotCred.CredentialRef != "secret:github-token" {
		t.Errorf("credential_ref roundtrip: got %q, want %q", gotCred.CredentialRef, "secret:github-token")
	}
}

// Test helpers.

func isolatedPostgresDSN(t *testing.T, ctx context.Context, dsn string) string {
	t.Helper()

	schema := fmt.Sprintf("gito_storage_test_%d", time.Now().UnixNano())
	adminPool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatalf("open admin pool: %v", err)
	}
	if err := adminPool.Ping(ctx); err != nil {
		adminPool.Close()
		t.Fatalf("ping admin pool: %v", err)
	}
	if _, err := adminPool.Exec(ctx, "CREATE SCHEMA "+schema); err != nil {
		adminPool.Close()
		t.Fatalf("create test schema: %v", err)
	}
	t.Cleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if _, err := adminPool.Exec(cleanupCtx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE"); err != nil {
			t.Logf("drop test schema %s: %v", schema, err)
		}
		adminPool.Close()
	})

	return postgresDSNWithSearchPath(t, dsn, schema)
}

func postgresDSNWithSearchPath(t *testing.T, dsn string, schema string) string {
	t.Helper()

	parsed, err := url.Parse(dsn)
	if err == nil && (parsed.Scheme == "postgres" || parsed.Scheme == "postgresql") {
		query := parsed.Query()
		query.Set("search_path", schema)
		parsed.RawQuery = query.Encode()
		return parsed.String()
	}

	return strings.TrimSpace(dsn) + " search_path=" + schema
}
