gito/services/core/internal/storage/postgres.go

package storage

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"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/provider"
)

// PgStore is a pgxpool-backed Store implementation.
type PgStore struct {
	pool              *pgxpool.Pool
	repos             *pgRepoStore
	leases            *pgWorkspaceLeaseStore
	ops               *pgOperationStore
	opEvents          *pgOperationEventStore
	watches           *pgBranchWatchStore
	cursors           *pgRevisionCursorStore
	deliveries        *pgProviderDeliveryStore
	agentRunInputs    *pgAgentRunInputStore
	webhookDeliveries *pgWebhookDeliveryStore
	providerConfigs   *pgProviderConfigStore
}

// NewPgStore opens a pgxpool connection and optionally applies a migration.
// migrationSQL is the raw SQL content (e.g. from //go:embed); pass an empty
// string to skip migration.
func NewPgStore(ctx context.Context, dsn string, migrationSQL string) (*PgStore, error) {
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		return nil, fmt.Errorf("pgstore: open pool: %w", err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("pgstore: ping: %w", err)
	}
	if migrationSQL != "" {
		if err := applyMigrationSQL(ctx, pool, migrationSQL); err != nil {
			pool.Close()
			return nil, fmt.Errorf("pgstore: migration: %w", err)
		}
	}
	s := &PgStore{pool: pool}
	s.repos = &pgRepoStore{pool: pool}
	s.leases = &pgWorkspaceLeaseStore{pool: pool}
	s.ops = &pgOperationStore{pool: pool}
	s.opEvents = &pgOperationEventStore{pool: pool}
	s.watches = &pgBranchWatchStore{pool: pool}
	s.cursors = &pgRevisionCursorStore{pool: pool}
	s.deliveries = &pgProviderDeliveryStore{pool: pool}
	s.agentRunInputs = &pgAgentRunInputStore{pool: pool}
	s.webhookDeliveries = &pgWebhookDeliveryStore{pool: pool}
	s.providerConfigs = newPgProviderConfigStore(pool)
	return s, nil
}

func (s *PgStore) Close() { s.pool.Close() }

func (s *PgStore) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }

func (s *PgStore) Repos() RepoStore                          { return s.repos }
func (s *PgStore) WorkspaceLeases() WorkspaceLeaseStore      { return s.leases }
func (s *PgStore) Operations() OperationStore                { return s.ops }
func (s *PgStore) OperationEvents() OperationEventStore      { return s.opEvents }
func (s *PgStore) BranchWatches() BranchWatchStore           { return s.watches }
func (s *PgStore) RevisionCursors() RevisionCursorStore      { return s.cursors }
func (s *PgStore) ProviderDeliveries() ProviderDeliveryStore { return s.deliveries }
func (s *PgStore) AgentRunInputs() AgentRunInputStore        { return s.agentRunInputs }
func (s *PgStore) WebhookDeliveries() WebhookDeliveryStore   { return s.webhookDeliveries }
func (s *PgStore) ProviderConfigs() ProviderConfigStore      { return s.providerConfigs }

// applyMigrationSQL executes the Up block extracted from goose-style SQL content.
func applyMigrationSQL(ctx context.Context, pool *pgxpool.Pool, content string) error {
	sql := extractUpBlock(content)
	if sql == "" {
		return nil
	}
	_, err := pool.Exec(ctx, sql)
	return err
}

func extractUpBlock(content string) string {
	const beginMarker = "-- +goose StatementBegin"
	const endMarker = "-- +goose StatementEnd"
	const upMarker = "-- +goose Up"

	upIdx := strings.Index(content, upMarker)
	if upIdx < 0 {
		return ""
	}
	after := content[upIdx:]
	beginIdx := strings.Index(after, beginMarker)
	endIdx := strings.Index(after, endMarker)
	if beginIdx < 0 || endIdx < 0 || endIdx <= beginIdx {
		return ""
	}
	return strings.TrimSpace(after[beginIdx+len(beginMarker) : endIdx])
}

// pgRepoStore

type pgRepoStore struct{ pool *pgxpool.Pool }

func (s *pgRepoStore) CreateRepo(ctx context.Context, repo core.Repo) error {
	repo, err := normalizeRepo(repo)
	if err != nil {
		return err
	}

	const q = `
		INSERT INTO repos (id, name, remote_url, default_branch, workspace_root, credential_ref)
		VALUES ($1, $2, $3, $4, $5, $6)
	`
	_, err = s.pool.Exec(ctx, q,
		repo.ID,
		repo.Name,
		repo.RemoteURL,
		repo.DefaultBranch,
		repo.WorkspaceRoot,
		nullableString(repo.CredentialRef),
	)
	if err != nil {
		if isUniqueViolation(err) {
			return fmt.Errorf("%w: %s", ErrRepoAlreadyExists, repo.ID)
		}
		return fmt.Errorf("create repo: %w", err)
	}
	return nil
}

func (s *pgRepoStore) GetRepo(ctx context.Context, id string) (core.Repo, error) {
	id = strings.TrimSpace(id)
	if id == "" {
		return core.Repo{}, fmt.Errorf("%w: id is required", ErrInvalidRepo)
	}

	const q = `
		SELECT id, name, remote_url, default_branch, workspace_root, credential_ref
		FROM repos
		WHERE id=$1
	`
	repo, err := scanRepo(s.pool.QueryRow(ctx, q, id))
	if err != nil {
		if isNoRows(err) {
			return core.Repo{}, fmt.Errorf("%w: %s", ErrRepoNotFound, id)
		}
		return core.Repo{}, fmt.Errorf("get repo: %w", err)
	}
	return repo, nil
}

func (s *pgRepoStore) ListRepos(ctx context.Context) ([]core.Repo, error) {
	const q = `
		SELECT id, name, remote_url, default_branch, workspace_root, credential_ref
		FROM repos
		ORDER BY id
	`
	rows, err := s.pool.Query(ctx, q)
	if err != nil {
		return nil, fmt.Errorf("list repos: %w", err)
	}
	defer rows.Close()

	var out []core.Repo
	for rows.Next() {
		repo, err := scanRepo(rows)
		if err != nil {
			return nil, fmt.Errorf("scan repo: %w", err)
		}
		out = append(out, repo)
	}
	return out, rows.Err()
}

func (s *pgRepoStore) UpdateRepo(ctx context.Context, repo core.Repo) error {
	repo, err := normalizeRepo(repo)
	if err != nil {
		return err
	}

	const q = `
		UPDATE repos
		SET name=$2,
			remote_url=$3,
			default_branch=$4,
			workspace_root=$5,
			credential_ref=$6,
			updated_at=now()
		WHERE id=$1
	`
	tag, err := s.pool.Exec(ctx, q,
		repo.ID,
		repo.Name,
		repo.RemoteURL,
		repo.DefaultBranch,
		repo.WorkspaceRoot,
		nullableString(repo.CredentialRef),
	)
	if err != nil {
		return fmt.Errorf("update repo: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return fmt.Errorf("%w: %s", ErrRepoNotFound, repo.ID)
	}
	return nil
}

func normalizeRepo(repo core.Repo) (core.Repo, error) {
	repo.ID = strings.TrimSpace(repo.ID)
	repo.Name = strings.TrimSpace(repo.Name)
	repo.RemoteURL = strings.TrimSpace(repo.RemoteURL)
	repo.DefaultBranch = strings.TrimSpace(repo.DefaultBranch)
	repo.WorkspaceRoot = strings.TrimSpace(repo.WorkspaceRoot)
	repo.CredentialRef = strings.TrimSpace(repo.CredentialRef)

	var missing []string
	if repo.ID == "" {
		missing = append(missing, "id")
	}
	if repo.Name == "" {
		missing = append(missing, "name")
	}
	if repo.RemoteURL == "" {
		missing = append(missing, "remote_url")
	}
	if repo.DefaultBranch == "" {
		missing = append(missing, "default_branch")
	}
	if repo.WorkspaceRoot == "" {
		missing = append(missing, "workspace_root")
	}
	if len(missing) > 0 {
		return core.Repo{}, fmt.Errorf("%w: required fields missing: %s", ErrInvalidRepo, strings.Join(missing, ", "))
	}
	return repo, nil
}

type repoScanner interface {
	Scan(dest ...any) error
}

func scanRepo(scanner repoScanner) (core.Repo, error) {
	var repo core.Repo
	var credentialRef sql.NullString
	if err := scanner.Scan(
		&repo.ID,
		&repo.Name,
		&repo.RemoteURL,
		&repo.DefaultBranch,
		&repo.WorkspaceRoot,
		&credentialRef,
	); err != nil {
		return core.Repo{}, err
	}
	if credentialRef.Valid {
		repo.CredentialRef = credentialRef.String
	}
	return repo, nil
}

func nullableString(value string) any {
	if value == "" {
		return nil
	}
	return value
}

// pgWorkspaceLeaseStore

type pgWorkspaceLeaseStore struct{ pool *pgxpool.Pool }

func (s *pgWorkspaceLeaseStore) SaveLease(ctx context.Context, lease core.WorkspaceLease) error {
	lease, err := normalizeLease(lease)
	if err != nil {
		return err
	}

	const q = `
		INSERT INTO workspace_leases (id, repo_id, slot, path, state, expires_at)
		VALUES ($1, $2, $3, $4, $5, $6)
	`
	_, err = s.pool.Exec(ctx, q,
		lease.ID,
		lease.RepoID,
		lease.Slot,
		lease.Path,
		lease.State,
		nullableTime(lease.ExpiresAt),
	)
	if err != nil {
		if isUniqueViolation(err) {
			return fmt.Errorf("%w: %s", ErrLeaseAlreadyExists, lease.ID)
		}
		return fmt.Errorf("save lease: %w", err)
	}
	return nil
}

func (s *pgWorkspaceLeaseStore) GetLease(ctx context.Context, id string) (core.WorkspaceLease, error) {
	id = strings.TrimSpace(id)
	if id == "" {
		return core.WorkspaceLease{}, fmt.Errorf("%w: id is required", ErrInvalidLease)
	}

	const q = `
		SELECT id, repo_id, slot, path, state, expires_at
		FROM workspace_leases
		WHERE id=$1
	`
	lease, err := scanLease(s.pool.QueryRow(ctx, q, id))
	if err != nil {
		if isNoRows(err) {
			return core.WorkspaceLease{}, fmt.Errorf("%w: %s", ErrLeaseNotFound, id)
		}
		return core.WorkspaceLease{}, fmt.Errorf("get lease: %w", err)
	}
	return lease, nil
}

func (s *pgWorkspaceLeaseStore) ListLeases(ctx context.Context, repoID string) ([]core.WorkspaceLease, error) {
	repoID = strings.TrimSpace(repoID)
	if repoID == "" {
		return nil, fmt.Errorf("%w: repo_id is required", ErrInvalidLease)
	}

	const q = `
		SELECT id, repo_id, slot, path, state, expires_at
		FROM workspace_leases
		WHERE repo_id=$1
		ORDER BY slot
	`
	rows, err := s.pool.Query(ctx, q, repoID)
	if err != nil {
		return nil, fmt.Errorf("list leases: %w", err)
	}
	defer rows.Close()

	var out []core.WorkspaceLease
	for rows.Next() {
		lease, err := scanLease(rows)
		if err != nil {
			return nil, fmt.Errorf("scan lease: %w", err)
		}
		out = append(out, lease)
	}
	return out, rows.Err()
}

func (s *pgWorkspaceLeaseStore) AcquireAvailableLease(ctx context.Context, repoID string, now time.Time, ttl time.Duration) (core.WorkspaceLease, bool, error) {
	repoID = strings.TrimSpace(repoID)
	if repoID == "" {
		return core.WorkspaceLease{}, false, fmt.Errorf("%w: repo_id is required", ErrInvalidLease)
	}
	if ttl <= 0 {
		return core.WorkspaceLease{}, false, fmt.Errorf("%w: ttl must be positive", ErrInvalidLease)
	}

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return core.WorkspaceLease{}, false, fmt.Errorf("begin lease acquire: %w", err)
	}
	defer tx.Rollback(ctx)

	expiresAt := now.UTC().Add(ttl)
	const q = `
		WITH candidate AS (
			SELECT id
			FROM workspace_leases
			WHERE repo_id=$1 AND state=$2
			ORDER BY slot
			FOR UPDATE SKIP LOCKED
			LIMIT 1
		)
		UPDATE workspace_leases AS lease
		SET state=$3,
			expires_at=$4,
			updated_at=now()
		FROM candidate
		WHERE lease.id=candidate.id
		RETURNING lease.id, lease.repo_id, lease.slot, lease.path, lease.state, lease.expires_at
	`
	lease, err := scanLease(tx.QueryRow(ctx, q, repoID, core.WorkspaceAvailable, core.WorkspaceLeased, expiresAt))
	if err != nil {
		if isNoRows(err) {
			return core.WorkspaceLease{}, false, nil
		}
		return core.WorkspaceLease{}, false, fmt.Errorf("acquire lease: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return core.WorkspaceLease{}, false, fmt.Errorf("commit lease acquire: %w", err)
	}
	return lease, true, nil
}

func (s *pgWorkspaceLeaseStore) ReleaseLease(ctx context.Context, id string, release LeaseRelease) (core.WorkspaceLease, error) {
	id = strings.TrimSpace(id)
	if id == "" {
		return core.WorkspaceLease{}, fmt.Errorf("%w: id is required", ErrInvalidLease)
	}
	nextState, err := leaseStateForRelease(release)
	if err != nil {
		return core.WorkspaceLease{}, err
	}

	const q = `
		UPDATE workspace_leases
		SET state=$2,
			expires_at=NULL,
			updated_at=now()
		WHERE id=$1 AND state=$3
		RETURNING id, repo_id, slot, path, state, expires_at
	`
	lease, err := scanLease(s.pool.QueryRow(ctx, q, id, nextState, core.WorkspaceLeased))
	if err != nil {
		if isNoRows(err) {
			existing, getErr := s.GetLease(ctx, id)
			if getErr != nil {
				return core.WorkspaceLease{}, getErr
			}
			return core.WorkspaceLease{}, fmt.Errorf("%w: lease %s is %s, not %s", ErrInvalidLease, existing.ID, existing.State, core.WorkspaceLeased)
		}
		return core.WorkspaceLease{}, fmt.Errorf("release lease: %w", err)
	}
	return lease, nil
}

func normalizeLease(lease core.WorkspaceLease) (core.WorkspaceLease, error) {
	lease.ID = strings.TrimSpace(lease.ID)
	lease.RepoID = strings.TrimSpace(lease.RepoID)
	lease.Slot = strings.TrimSpace(lease.Slot)
	lease.Path = strings.TrimSpace(lease.Path)

	var missing []string
	if lease.ID == "" {
		missing = append(missing, "id")
	}
	if lease.RepoID == "" {
		missing = append(missing, "repo_id")
	}
	if lease.Slot == "" {
		missing = append(missing, "slot")
	}
	if lease.Path == "" {
		missing = append(missing, "path")
	}
	if len(missing) > 0 {
		return core.WorkspaceLease{}, fmt.Errorf("%w: required fields missing: %s", ErrInvalidLease, strings.Join(missing, ", "))
	}
	if !isValidWorkspaceState(lease.State) {
		return core.WorkspaceLease{}, fmt.Errorf("%w: unsupported state %q", ErrInvalidLease, lease.State)
	}
	return lease, nil
}

func isValidWorkspaceState(state core.WorkspaceState) bool {
	switch state {
	case core.WorkspaceAvailable, core.WorkspaceLeased, core.WorkspaceDirty, core.WorkspaceError:
		return true
	default:
		return false
	}
}

func leaseStateForRelease(release LeaseRelease) (core.WorkspaceState, error) {
	if release.Dirty {
		return core.WorkspaceDirty, nil
	}
	switch release.OperationState {
	case core.OperationSucceeded, core.OperationCancelled:
		return core.WorkspaceAvailable, nil
	case core.OperationFailed:
		return core.WorkspaceError, nil
	default:
		return "", fmt.Errorf("%w: unsupported release operation state %q", ErrInvalidLease, release.OperationState)
	}
}

func scanLease(scanner repoScanner) (core.WorkspaceLease, error) {
	var lease core.WorkspaceLease
	var expiresAt sql.NullTime
	if err := scanner.Scan(
		&lease.ID,
		&lease.RepoID,
		&lease.Slot,
		&lease.Path,
		&lease.State,
		&expiresAt,
	); err != nil {
		return core.WorkspaceLease{}, err
	}
	if expiresAt.Valid {
		value := expiresAt.Time
		lease.ExpiresAt = &value
	}
	return lease, nil
}

func nullableTime(value *time.Time) any {
	if value == nil {
		return nil
	}
	return value.UTC()
}

// pgOperationStore

type pgOperationStore struct{ pool *pgxpool.Pool }

const opColumns = "id, repo_id, type, state, idempotency_key, created_by, created_at, updated_at"
const opReturningColumns = "op.id, op.repo_id, op.type, op.state, op.idempotency_key, op.created_by, op.created_at, op.updated_at"

// opIdempotencyConstraint is the partial unique index on operations.idempotency_key
// (see migrations/00001_initial.sql). A unique violation naming this constraint is
// an idempotent retry; any other unique violation is a duplicate create.
const opIdempotencyConstraint = "ux_operations_idempotency"

func (s *pgOperationStore) CreateOperation(ctx context.Context, op core.Operation) (OperationCreateResult, error) {
	op, err := normalizeOperation(op)
	if err != nil {
		return OperationCreateResult{}, err
	}

	const q = `
		INSERT INTO operations (id, repo_id, type, state, idempotency_key, created_by)
		VALUES ($1, $2, $3, $4, $5, $6)
		RETURNING ` + opColumns
	created, err := scanOperation(s.pool.QueryRow(ctx, q,
		op.ID,
		op.RepoID,
		op.Type,
		op.State,
		nullableString(op.IdempotencyKey),
		op.CreatedBy,
	))
	if err != nil {
		if constraint, ok := uniqueViolationConstraint(err); ok {
			// Only the idempotency index conflict represents an idempotent retry
			// and returns the existing operation. Any other unique violation
			// (id primary key, etc.) is a genuine duplicate create, regardless of
			// whether an idempotency key was supplied.
			if constraint == opIdempotencyConstraint && op.IdempotencyKey != "" {
				existing, getErr := s.getByIdempotencyKey(ctx, op.IdempotencyKey)
				if getErr != nil {
					return OperationCreateResult{}, getErr
				}
				return OperationCreateResult{Operation: existing, Created: false}, nil
			}
			return OperationCreateResult{}, fmt.Errorf("%w: %s", ErrOperationAlreadyExists, op.ID)
		}
		return OperationCreateResult{}, fmt.Errorf("create operation: %w", err)
	}
	return OperationCreateResult{Operation: created, Created: true}, nil
}

func (s *pgOperationStore) getByIdempotencyKey(ctx context.Context, key string) (core.Operation, error) {
	const q = `SELECT ` + opColumns + ` FROM operations WHERE idempotency_key=$1`
	op, err := scanOperation(s.pool.QueryRow(ctx, q, key))
	if err != nil {
		if isNoRows(err) {
			return core.Operation{}, fmt.Errorf("%w: idempotency_key %s", ErrOperationNotFound, key)
		}
		return core.Operation{}, fmt.Errorf("get operation by idempotency key: %w", err)
	}
	return op, nil
}

func (s *pgOperationStore) GetOperation(ctx context.Context, id string) (core.Operation, error) {
	id = strings.TrimSpace(id)
	if id == "" {
		return core.Operation{}, fmt.Errorf("%w: id is required", ErrInvalidOperation)
	}

	const q = `SELECT ` + opColumns + ` FROM operations WHERE id=$1`
	op, err := scanOperation(s.pool.QueryRow(ctx, q, id))
	if err != nil {
		if isNoRows(err) {
			return core.Operation{}, fmt.Errorf("%w: %s", ErrOperationNotFound, id)
		}
		return core.Operation{}, fmt.Errorf("get operation: %w", err)
	}
	return op, nil
}

func (s *pgOperationStore) ListOperations(ctx context.Context, repoID string) ([]core.Operation, error) {
	repoID = strings.TrimSpace(repoID)
	if repoID == "" {
		return nil, fmt.Errorf("%w: repo_id is required", ErrInvalidOperation)
	}

	const q = `SELECT ` + opColumns + ` FROM operations WHERE repo_id=$1 ORDER BY created_at, id`
	rows, err := s.pool.Query(ctx, q, repoID)
	if err != nil {
		return nil, fmt.Errorf("list operations: %w", err)
	}
	defer rows.Close()

	var out []core.Operation
	for rows.Next() {
		op, err := scanOperation(rows)
		if err != nil {
			return nil, fmt.Errorf("scan operation: %w", err)
		}
		out = append(out, op)
	}
	return out, rows.Err()
}

func (s *pgOperationStore) StartOperation(ctx context.Context, id string, now time.Time) (core.Operation, error) {
	return s.transition(ctx, id, core.OperationRunning, now)
}

func (s *pgOperationStore) SucceedOperation(ctx context.Context, id string, now time.Time) (core.Operation, error) {
	return s.transition(ctx, id, core.OperationSucceeded, now)
}

func (s *pgOperationStore) FailOperation(ctx context.Context, id string, now time.Time) (core.Operation, error) {
	return s.transition(ctx, id, core.OperationFailed, now)
}

func (s *pgOperationStore) CancelOperation(ctx context.Context, id string, now time.Time) (core.Operation, error) {
	return s.transition(ctx, id, core.OperationCancelled, now)
}

func (s *pgOperationStore) PickQueuedOperation(ctx context.Context, now time.Time) (core.Operation, bool, error) {
	const q = `
		WITH candidate AS (
			SELECT id
			FROM operations
			WHERE state=$1
			ORDER BY created_at, id
			FOR UPDATE SKIP LOCKED
			LIMIT 1
		)
		UPDATE operations AS op
		SET state=$2, updated_at=$3
		FROM candidate
		WHERE op.id=candidate.id
		RETURNING ` + opReturningColumns

	op, err := scanOperation(s.pool.QueryRow(ctx, q, core.OperationQueued, core.OperationRunning, now.UTC()))
	if err != nil {
		if isNoRows(err) {
			return core.Operation{}, false, nil
		}
		return core.Operation{}, false, fmt.Errorf("pick queued operation: %w", err)
	}
	return op, true, nil
}

func (s *pgOperationStore) PickQueuedAgentRunOperation(ctx context.Context, now time.Time) (core.Operation, bool, error) {
	const q = `
		WITH candidate AS (
			SELECT id
			FROM operations
			WHERE state=$1 AND type=$2
			ORDER BY created_at, id
			FOR UPDATE SKIP LOCKED
			LIMIT 1
		)
		UPDATE operations AS op
		SET state=$3, updated_at=$4
		FROM candidate
		WHERE op.id=candidate.id
		RETURNING ` + opReturningColumns

	op, err := scanOperation(s.pool.QueryRow(ctx, q, core.OperationQueued, core.OperationAgentRun, core.OperationRunning, now.UTC()))
	if err != nil {
		if isNoRows(err) {
			return core.Operation{}, false, nil
		}
		return core.Operation{}, false, fmt.Errorf("pick queued agent_run operation: %w", err)
	}
	return op, true, nil
}

// transition reads the current operation, validates the state change against
// the core transition rules, and persists the next state with an optimistic
// guard on the observed current state.
func (s *pgOperationStore) transition(ctx context.Context, id string, next core.OperationState, now time.Time) (core.Operation, error) {
	id = strings.TrimSpace(id)
	if id == "" {
		return core.Operation{}, fmt.Errorf("%w: id is required", ErrInvalidOperation)
	}

	current, err := s.GetOperation(ctx, id)
	if err != nil {
		return core.Operation{}, err
	}
	if !current.State.CanTransitionTo(next) {
		return core.Operation{}, fmt.Errorf("%w: cannot transition %s from %s to %s", ErrInvalidOperation, id, current.State, next)
	}

	const q = `
		UPDATE operations
		SET state=$3,
			updated_at=$4
		WHERE id=$1 AND state=$2
		RETURNING ` + opColumns
	op, err := scanOperation(s.pool.QueryRow(ctx, q, id, current.State, next, now.UTC()))
	if err != nil {
		if isNoRows(err) {
			// State changed between read and update; surface as an invalid transition.
			return core.Operation{}, fmt.Errorf("%w: %s no longer in state %s", ErrInvalidOperation, id, current.State)
		}
		return core.Operation{}, fmt.Errorf("transition operation: %w", err)
	}
	return op, nil
}

func normalizeOperation(op core.Operation) (core.Operation, error) {
	op.ID = strings.TrimSpace(op.ID)
	op.RepoID = strings.TrimSpace(op.RepoID)
	op.IdempotencyKey = strings.TrimSpace(op.IdempotencyKey)
	op.CreatedBy = strings.TrimSpace(op.CreatedBy)

	var missing []string
	if op.ID == "" {
		missing = append(missing, "id")
	}
	if op.RepoID == "" {
		missing = append(missing, "repo_id")
	}
	if op.CreatedBy == "" {
		missing = append(missing, "created_by")
	}
	if len(missing) > 0 {
		return core.Operation{}, fmt.Errorf("%w: required fields missing: %s", ErrInvalidOperation, strings.Join(missing, ", "))
	}
	if !op.Type.Valid() {
		return core.Operation{}, fmt.Errorf("%w: unsupported type %q", ErrInvalidOperation, op.Type)
	}
	if op.State == "" {
		op.State = core.OperationQueued
	}
	if op.State != core.OperationQueued {
		return core.Operation{}, fmt.Errorf("%w: create state must be %q, got %q", ErrInvalidOperation, core.OperationQueued, op.State)
	}
	return op, nil
}

func scanOperation(scanner repoScanner) (core.Operation, error) {
	var op core.Operation
	var idempotencyKey sql.NullString
	if err := scanner.Scan(
		&op.ID,
		&op.RepoID,
		&op.Type,
		&op.State,
		&idempotencyKey,
		&op.CreatedBy,
		&op.CreatedAt,
		&op.UpdatedAt,
	); err != nil {
		return core.Operation{}, err
	}
	if idempotencyKey.Valid {
		op.IdempotencyKey = idempotencyKey.String
	}
	return op, nil
}

// pgBranchWatchStore

type pgBranchWatchStore struct{ pool *pgxpool.Pool }

func (s *pgBranchWatchStore) UpsertBranchWatch(ctx context.Context, watch core.BranchWatch) (core.BranchWatch, error) {
	const q = `
		INSERT INTO branch_watches (id, provider, repo_id, branch, created_at)
		VALUES ($1, $2, $3, $4, $5)
		ON CONFLICT (provider, repo_id, branch) DO UPDATE
			SET id = branch_watches.id, created_at = branch_watches.created_at
		RETURNING id, provider, repo_id, branch, created_at
	`
	var out core.BranchWatch
	row := s.pool.QueryRow(ctx, q, watch.ID, watch.Provider, watch.RepoID, watch.Branch, watch.CreatedAt)
	if err := row.Scan(&out.ID, &out.Provider, &out.RepoID, &out.Branch, &out.CreatedAt); err != nil {
		return core.BranchWatch{}, fmt.Errorf("upsert branch watch: %w", err)
	}
	return out, nil
}

func (s *pgBranchWatchStore) FindBranchWatch(ctx context.Context, provider, repoID, branch string) (core.BranchWatch, bool, error) {
	const q = `SELECT id, provider, repo_id, branch, created_at FROM branch_watches WHERE provider=$1 AND repo_id=$2 AND branch=$3`
	var out core.BranchWatch
	row := s.pool.QueryRow(ctx, q, provider, repoID, branch)
	if err := row.Scan(&out.ID, &out.Provider, &out.RepoID, &out.Branch, &out.CreatedAt); err != nil {
		if isNoRows(err) {
			return core.BranchWatch{}, false, nil
		}
		return core.BranchWatch{}, false, fmt.Errorf("find branch watch: %w", err)
	}
	return out, true, nil
}

func (s *pgBranchWatchStore) ListBranchWatches(ctx context.Context) ([]core.BranchWatch, error) {
	const q = `SELECT id, provider, repo_id, branch, created_at FROM branch_watches ORDER BY id`
	rows, err := s.pool.Query(ctx, q)
	if err != nil {
		return nil, fmt.Errorf("list branch watches: %w", err)
	}
	defer rows.Close()
	var out []core.BranchWatch
	for rows.Next() {
		var w core.BranchWatch
		if err := rows.Scan(&w.ID, &w.Provider, &w.RepoID, &w.Branch, &w.CreatedAt); err != nil {
			return nil, fmt.Errorf("scan branch watch: %w", err)
		}
		out = append(out, w)
	}
	return out, rows.Err()
}

// pgRevisionCursorStore

type pgRevisionCursorStore struct{ pool *pgxpool.Pool }

func (s *pgRevisionCursorStore) UpsertRevisionCursor(ctx context.Context, cursor core.RevisionCursor) error {
	const q = `
		INSERT INTO revision_cursors (repo_id, branch, revision, observed_at)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT (repo_id, branch) DO UPDATE
			SET revision = EXCLUDED.revision, observed_at = EXCLUDED.observed_at
	`
	_, err := s.pool.Exec(ctx, q, cursor.RepoID, cursor.Branch, cursor.Revision, cursor.ObservedAt)
	if err != nil {
		return fmt.Errorf("upsert revision cursor: %w", err)
	}
	return nil
}

func (s *pgRevisionCursorStore) GetRevisionCursor(ctx context.Context, repoID, branch string) (core.RevisionCursor, bool, error) {
	const q = `SELECT repo_id, branch, revision, observed_at FROM revision_cursors WHERE repo_id=$1 AND branch=$2`
	var out core.RevisionCursor
	row := s.pool.QueryRow(ctx, q, repoID, branch)
	if err := row.Scan(&out.RepoID, &out.Branch, &out.Revision, &out.ObservedAt); err != nil {
		if isNoRows(err) {
			return core.RevisionCursor{}, false, nil
		}
		return core.RevisionCursor{}, false, fmt.Errorf("get revision cursor: %w", err)
	}
	return out, true, nil
}

// pgProviderDeliveryStore

type pgProviderDeliveryStore struct{ pool *pgxpool.Pool }

func (s *pgProviderDeliveryStore) RecordOnce(ctx context.Context, delivery core.ProviderDelivery) (DeliveryResult, error) {
	const insert = `
		INSERT INTO provider_deliveries
			(id, provider, delivery_id, dedupe_key, event_id, repo_id, branch, revision, created_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
		ON CONFLICT (provider, dedupe_key) DO NOTHING
	`
	tag, err := s.pool.Exec(ctx, insert,
		delivery.ID, delivery.Provider, delivery.DeliveryID, delivery.DedupeKey,
		delivery.EventID, delivery.RepoID, delivery.Branch, delivery.Revision,
		delivery.CreatedAt,
	)
	if err != nil {
		return DeliveryResult{}, fmt.Errorf("record delivery: %w", err)
	}
	if tag.RowsAffected() == 1 {
		return DeliveryResult{First: true}, nil
	}
	// Duplicate: fetch existing event_id and created_at for idempotent response.
	const sel = `SELECT event_id, created_at FROM provider_deliveries WHERE provider=$1 AND dedupe_key=$2`
	var existingEventID string
	var existingCreatedAt time.Time
	row := s.pool.QueryRow(ctx, sel, delivery.Provider, delivery.DedupeKey)
	if err := row.Scan(&existingEventID, &existingCreatedAt); err != nil {
		return DeliveryResult{}, fmt.Errorf("fetch existing delivery: %w", err)
	}
	return DeliveryResult{
		First:             false,
		ExistingEventID:   existingEventID,
		ExistingCreatedAt: existingCreatedAt,
	}, nil
}

func (s *pgProviderDeliveryStore) DeleteDelivery(ctx context.Context, provider, dedupeKey string) error {
	const query = `DELETE FROM provider_deliveries WHERE provider=$1 AND dedupe_key=$2`
	_, err := s.pool.Exec(ctx, query, provider, dedupeKey)
	if err != nil {
		return fmt.Errorf("delete delivery: %w", err)
	}
	return nil
}

func isNoRows(err error) bool {
	return err != nil && (errors.Is(err, pgx.ErrNoRows) || strings.Contains(err.Error(), "no rows"))
}

func isUniqueViolation(err error) bool {
	_, ok := uniqueViolationConstraint(err)
	return ok
}

// uniqueViolationConstraint reports whether err is a Postgres unique violation
// (SQLSTATE 23505) and returns the violated constraint/index name when so. The
// name lets callers distinguish, for example, an idempotency-key index conflict
// from a primary-key collision.
func uniqueViolationConstraint(err error) (string, bool) {
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
		return pgErr.ConstraintName, true
	}
	return "", false
}

// pgOperationEventStore

type pgOperationEventStore struct {
	pool *pgxpool.Pool
}

func (s *pgOperationEventStore) AppendEvent(ctx context.Context, ev OperationEvent) error {
	opID := strings.TrimSpace(ev.OperationID)
	eventID := strings.TrimSpace(ev.Event.ID)
	eventType := strings.TrimSpace(ev.Event.Type)
	subject := strings.TrimSpace(ev.Event.Subject)

	if eventID == "" || eventType == "" || subject == "" {
		return fmt.Errorf("%w: event_id, type, and subject are required", ErrInvalidOperation)
	}

	if ev.PublishedAt != nil {
		return fmt.Errorf("%w: published_at must be nil on append", ErrInvalidOperation)
	}

	payload, err := validateAndNormalizePayload(ev.Event.Payload)
	if err != nil {
		return err
	}

	createdAt := ev.Event.CreatedAt
	if createdAt.IsZero() {
		createdAt = time.Now().UTC()
	} else {
		createdAt = createdAt.UTC()
	}

	const q = `
		INSERT INTO operation_events (id, operation_id, type, subject, payload, created_at)
		VALUES ($1, $2, $3, $4, $5, $6)
	`
	var opIDVal *string
	if opID != "" {
		opIDVal = &opID
	}
	_, err = s.pool.Exec(ctx, q, eventID, opIDVal, eventType, subject, payload, createdAt)
	if err != nil {
		return fmt.Errorf("append operation event: %w", err)
	}
	return nil
}

func validateAndNormalizePayload(payload []byte) ([]byte, error) {
	if len(payload) == 0 {
		return []byte("{}"), nil
	}
	var m map[string]any
	if err := json.Unmarshal(payload, &m); err != nil {
		return nil, fmt.Errorf("%w: payload must be a valid JSON object", ErrInvalidOperation)
	}
	if m == nil {
		return nil, fmt.Errorf("%w: payload cannot be null", ErrInvalidOperation)
	}
	return payload, nil
}

func (s *pgOperationEventStore) ListEvents(ctx context.Context, operationID string) ([]OperationEvent, error) {
	operationID = strings.TrimSpace(operationID)
	if operationID == "" {
		return nil, fmt.Errorf("%w: operation_id is required", ErrInvalidOperation)
	}

	const q = `
		SELECT operation_id, id, type, subject, payload, created_at, published_at
		FROM operation_events
		WHERE operation_id=$1
		ORDER BY created_at, id
	`
	rows, err := s.pool.Query(ctx, q, operationID)
	if err != nil {
		return nil, fmt.Errorf("list operation events: %w", err)
	}
	defer rows.Close()

	var out []OperationEvent
	for rows.Next() {
		ev, err := scanOperationEvent(rows)
		if err != nil {
			return nil, fmt.Errorf("scan operation event: %w", err)
		}
		out = append(out, ev)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list operation events rows error: %w", err)
	}
	return out, nil
}

func (s *pgOperationEventStore) ListPendingEvents(ctx context.Context, limit int) ([]OperationEvent, error) {
	if limit <= 0 {
		return nil, fmt.Errorf("%w: limit must be positive", ErrInvalidOperation)
	}

	const q = `
		SELECT operation_id, id, type, subject, payload, created_at, published_at
		FROM operation_events
		WHERE published_at IS NULL
		ORDER BY created_at, id
		LIMIT $1
	`
	rows, err := s.pool.Query(ctx, q, limit)
	if err != nil {
		return nil, fmt.Errorf("list pending operation events: %w", err)
	}
	defer rows.Close()

	var out []OperationEvent
	for rows.Next() {
		ev, err := scanOperationEvent(rows)
		if err != nil {
			return nil, fmt.Errorf("scan operation event: %w", err)
		}
		out = append(out, ev)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list pending operation events rows error: %w", err)
	}
	return out, nil
}

func (s *pgOperationEventStore) MarkPublished(ctx context.Context, eventID string, now time.Time) (OperationEvent, error) {
	eventID = strings.TrimSpace(eventID)
	if eventID == "" {
		return OperationEvent{}, fmt.Errorf("%w: event_id is required", ErrInvalidOperation)
	}

	const q = `
		UPDATE operation_events
		SET published_at=$2
		WHERE id=$1
		RETURNING operation_id, id, type, subject, payload, created_at, published_at
	`
	ev, err := scanOperationEvent(s.pool.QueryRow(ctx, q, eventID, now.UTC()))
	if err != nil {
		if isNoRows(err) {
			return OperationEvent{}, fmt.Errorf("%w: event %s not found", ErrOperationNotFound, eventID)
		}
		return OperationEvent{}, fmt.Errorf("scan operation event: %w", err)
	}
	return ev, nil
}

func scanOperationEvent(scanner repoScanner) (OperationEvent, error) {
	var ev OperationEvent
	var opID sql.NullString
	var publishedAt sql.NullTime
	if err := scanner.Scan(
		&opID,
		&ev.Event.ID,
		&ev.Event.Type,
		&ev.Event.Subject,
		&ev.Event.Payload,
		&ev.Event.CreatedAt,
		&publishedAt,
	); err != nil {
		return OperationEvent{}, err
	}
	if opID.Valid {
		ev.OperationID = opID.String
	}
	ev.Event.CreatedAt = ev.Event.CreatedAt.UTC()
	if publishedAt.Valid {
		t := publishedAt.Time.UTC()
		ev.PublishedAt = &t
	}
	return ev, nil
}

// pgAgentRunInputStore

type pgAgentRunInputStore struct{ pool *pgxpool.Pool }

func (s *pgAgentRunInputStore) CreateAgentRunInput(ctx context.Context, input core.DurableAgentRunInput) error {
	input.OperationID = strings.TrimSpace(input.OperationID)
	input.RepoID = strings.TrimSpace(input.RepoID)
	input.Branch = strings.TrimSpace(input.Branch)
	input.WorkspacePath = strings.TrimSpace(input.WorkspacePath)
	input.Instruction = strings.TrimSpace(input.Instruction)
	input.ExpectedRevision = strings.TrimSpace(input.ExpectedRevision)

	var missing []string
	if input.OperationID == "" {
		missing = append(missing, "operation_id")
	}
	if input.RepoID == "" {
		missing = append(missing, "repo_id")
	}
	if input.Branch == "" {
		missing = append(missing, "branch")
	}
	if input.WorkspacePath == "" {
		missing = append(missing, "workspace_path")
	}
	if input.Instruction == "" {
		missing = append(missing, "instruction")
	}
	if input.ExpectedRevision == "" {
		missing = append(missing, "expected_revision")
	}
	if len(missing) > 0 {
		return fmt.Errorf("%w: required fields missing: %s", ErrInvalidAgentRunInput, strings.Join(missing, ", "))
	}

	policyJSON, err := json.Marshal(input.PolicyContext)
	if err != nil {
		return fmt.Errorf("%w: marshal policy_context: %v", ErrInvalidAgentRunInput, err)
	}

	createdAt := input.CreatedAt
	if createdAt.IsZero() {
		createdAt = time.Now().UTC()
	}

	const q = `
		INSERT INTO agent_run_inputs
			(operation_id, repo_id, branch, workspace_path, instruction, policy_context, expected_revision, created_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
	`
	_, err = s.pool.Exec(ctx, q,
		input.OperationID, input.RepoID, input.Branch, input.WorkspacePath,
		input.Instruction, policyJSON, input.ExpectedRevision, createdAt,
	)
	if err != nil {
		return fmt.Errorf("create agent run input: %w", err)
	}
	return nil
}

func (s *pgAgentRunInputStore) GetAgentRunInput(ctx context.Context, operationID string) (core.DurableAgentRunInput, error) {
	operationID = strings.TrimSpace(operationID)
	if operationID == "" {
		return core.DurableAgentRunInput{}, fmt.Errorf("%w: operation_id is required", ErrAgentRunInputNotFound)
	}

	const q = `
		SELECT operation_id, repo_id, branch, workspace_path, instruction, policy_context, expected_revision, created_at
		FROM agent_run_inputs
		WHERE operation_id=$1
	`
	var out core.DurableAgentRunInput
	var policyJSON []byte
	row := s.pool.QueryRow(ctx, q, operationID)
	if err := row.Scan(
		&out.OperationID, &out.RepoID, &out.Branch, &out.WorkspacePath,
		&out.Instruction, &policyJSON, &out.ExpectedRevision, &out.CreatedAt,
	); err != nil {
		if isNoRows(err) {
			return core.DurableAgentRunInput{}, fmt.Errorf("%w: %s", ErrAgentRunInputNotFound, operationID)
		}
		return core.DurableAgentRunInput{}, fmt.Errorf("get agent run input: %w", err)
	}
	out.CreatedAt = out.CreatedAt.UTC()
	if len(policyJSON) > 0 {
		if err := json.Unmarshal(policyJSON, &out.PolicyContext); err != nil {
			return core.DurableAgentRunInput{}, fmt.Errorf("unmarshal policy_context: %w", err)
		}
	}
	return out, nil
}

// pgWebhookDeliveryStore

type pgWebhookDeliveryStore struct{ pool *pgxpool.Pool }

func (s *pgWebhookDeliveryStore) EnqueueDelivery(ctx context.Context, delivery core.WebhookDelivery) (WebhookDeliveryCreateResult, error) {
	if delivery.ID == "" {
		delivery.ID = newID()
	}
	if delivery.Status == "" {
		delivery.Status = core.WebhookDeliveryPending
	}

	now := time.Now().UTC()
	createdAt := delivery.CreatedAt
	if createdAt.IsZero() {
		createdAt = now
	}
	updatedAt := delivery.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = now
	}
	nextAttemptAt := delivery.NextAttemptAt
	if nextAttemptAt.IsZero() {
		nextAttemptAt = now
	}

	const insert = `
		INSERT INTO webhook_deliveries
			(id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
		ON CONFLICT (event_id, subscription_id) DO NOTHING
	`
	tag, err := s.pool.Exec(ctx, insert,
		delivery.ID, delivery.EventID, delivery.SubscriptionID, string(delivery.Status),
		delivery.AttemptCount, nextAttemptAt, delivery.LastStatusCode, delivery.LastError,
		createdAt, updatedAt,
	)
	if err != nil {
		return WebhookDeliveryCreateResult{}, fmt.Errorf("enqueue webhook delivery: %w", err)
	}
	if tag.RowsAffected() == 1 {
		delivery.CreatedAt = createdAt
		delivery.UpdatedAt = updatedAt
		delivery.NextAttemptAt = nextAttemptAt
		return WebhookDeliveryCreateResult{Delivery: delivery, Created: true}, nil
	}

	const sel = `
		SELECT id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at
		FROM webhook_deliveries
		WHERE event_id=$1 AND subscription_id=$2
	`
	var existing core.WebhookDelivery
	var statusStr string
	row := s.pool.QueryRow(ctx, sel, delivery.EventID, delivery.SubscriptionID)
	if err := row.Scan(
		&existing.ID, &existing.EventID, &existing.SubscriptionID, &statusStr,
		&existing.AttemptCount, &existing.NextAttemptAt, &existing.LastStatusCode, &existing.LastError,
		&existing.CreatedAt, &existing.UpdatedAt,
	); err != nil {
		return WebhookDeliveryCreateResult{}, fmt.Errorf("fetch existing webhook delivery: %w", err)
	}
	existing.Status = core.WebhookDeliveryStatus(statusStr)
	existing.NextAttemptAt = existing.NextAttemptAt.UTC()
	existing.CreatedAt = existing.CreatedAt.UTC()
	existing.UpdatedAt = existing.UpdatedAt.UTC()
	return WebhookDeliveryCreateResult{Delivery: existing, Created: false}, nil
}

func (s *pgWebhookDeliveryStore) GetDelivery(ctx context.Context, id string) (core.WebhookDelivery, error) {
	const sel = `
		SELECT id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at
		FROM webhook_deliveries
		WHERE id=$1
	`
	var out core.WebhookDelivery
	var statusStr string
	row := s.pool.QueryRow(ctx, sel, id)
	if err := row.Scan(
		&out.ID, &out.EventID, &out.SubscriptionID, &statusStr,
		&out.AttemptCount, &out.NextAttemptAt, &out.LastStatusCode, &out.LastError,
		&out.CreatedAt, &out.UpdatedAt,
	); err != nil {
		if isNoRows(err) {
			return core.WebhookDelivery{}, fmt.Errorf("webhook delivery not found: %s", id)
		}
		return core.WebhookDelivery{}, fmt.Errorf("get webhook delivery: %w", err)
	}
	out.Status = core.WebhookDeliveryStatus(statusStr)
	out.NextAttemptAt = out.NextAttemptAt.UTC()
	out.CreatedAt = out.CreatedAt.UTC()
	out.UpdatedAt = out.UpdatedAt.UTC()
	return out, nil
}

func (s *pgWebhookDeliveryStore) PickPendingDeliveries(ctx context.Context, limit int, now time.Time) ([]core.WebhookDelivery, error) {
	const q = `
		WITH selected AS (
			SELECT id
			FROM webhook_deliveries
			WHERE status IN ('pending', 'retryable') AND next_attempt_at <= $1
			ORDER BY next_attempt_at ASC, id ASC
			LIMIT $2
			FOR UPDATE SKIP LOCKED
		)
		UPDATE webhook_deliveries
		SET status = 'sending', updated_at = $3
		FROM selected
		WHERE webhook_deliveries.id = selected.id
		RETURNING webhook_deliveries.id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at
	`
	rows, err := s.pool.Query(ctx, q, now, limit, now)
	if err != nil {
		return nil, fmt.Errorf("pick pending deliveries: %w", err)
	}
	defer rows.Close()

	var list []core.WebhookDelivery
	for rows.Next() {
		var d core.WebhookDelivery
		var statusStr string
		if err := rows.Scan(
			&d.ID, &d.EventID, &d.SubscriptionID, &statusStr,
			&d.AttemptCount, &d.NextAttemptAt, &d.LastStatusCode, &d.LastError,
			&d.CreatedAt, &d.UpdatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan webhook delivery: %w", err)
		}
		d.Status = core.WebhookDeliveryStatus(statusStr)
		d.NextAttemptAt = d.NextAttemptAt.UTC()
		d.CreatedAt = d.CreatedAt.UTC()
		d.UpdatedAt = d.UpdatedAt.UTC()
		list = append(list, d)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("rows error: %w", err)
	}
	return list, nil
}

// StartDelivery claims a single delivery by id when it is pending/retryable and
// due. It returns the claimed delivery, true when a transition happened, and
// false when the delivery is not yet due or is in a terminal state.
func (s *pgWebhookDeliveryStore) StartDelivery(ctx context.Context, id string, now time.Time) (core.WebhookDelivery, bool, error) {
	const q = `
		UPDATE webhook_deliveries
		SET status = 'sending', updated_at = $2
		WHERE id = $1
		  AND status IN ('pending', 'retryable')
		  AND next_attempt_at <= $2
		RETURNING id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at
	`
	var d core.WebhookDelivery
	var statusStr string
	row := s.pool.QueryRow(ctx, q, id, now)
	if err := row.Scan(
		&d.ID, &d.EventID, &d.SubscriptionID, &statusStr,
		&d.AttemptCount, &d.NextAttemptAt, &d.LastStatusCode, &d.LastError,
		&d.CreatedAt, &d.UpdatedAt,
	); err != nil {
		if isNoRows(err) {
			return core.WebhookDelivery{}, false, nil
		}
		return core.WebhookDelivery{}, false, fmt.Errorf("start delivery: %w", err)
	}
	d.Status = core.WebhookDeliveryStatus(statusStr)
	d.NextAttemptAt = d.NextAttemptAt.UTC()
	d.CreatedAt = d.CreatedAt.UTC()
	d.UpdatedAt = d.UpdatedAt.UTC()
	return d, true, nil
}

func (s *pgWebhookDeliveryStore) RecordOutcome(ctx context.Context, id string, status core.WebhookDeliveryStatus, statusCode int, lastErr string, nextAttemptAt time.Time, now time.Time) (core.WebhookDelivery, error) {
	const q = `
		UPDATE webhook_deliveries
		SET status = $2,
		    attempt_count = attempt_count + 1,
		    next_attempt_at = $3,
		    last_status_code = $4,
		    last_error = $5,
		    updated_at = $6
		WHERE id = $1
		RETURNING id, event_id, subscription_id, status, attempt_count, next_attempt_at, last_status_code, last_error, created_at, updated_at
	`
	var d core.WebhookDelivery
	var statusStr string
	row := s.pool.QueryRow(ctx, q, id, string(status), nextAttemptAt, statusCode, lastErr, now)
	if err := row.Scan(
		&d.ID, &d.EventID, &d.SubscriptionID, &statusStr,
		&d.AttemptCount, &d.NextAttemptAt, &d.LastStatusCode, &d.LastError,
		&d.CreatedAt, &d.UpdatedAt,
	); err != nil {
		if isNoRows(err) {
			return core.WebhookDelivery{}, fmt.Errorf("webhook delivery not found for outcome: %s", id)
		}
		return core.WebhookDelivery{}, fmt.Errorf("record outcome: %w", err)
	}
	d.Status = core.WebhookDeliveryStatus(statusStr)
	d.NextAttemptAt = d.NextAttemptAt.UTC()
	d.CreatedAt = d.CreatedAt.UTC()
	d.UpdatedAt = d.UpdatedAt.UTC()
	return d, nil
}

func newID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

// pgProviderConfigStore

type pgProviderConfigStore struct{ pool *pgxpool.Pool }

func newPgProviderConfigStore(pool *pgxpool.Pool) *pgProviderConfigStore {
	return &pgProviderConfigStore{pool: pool}
}

func normalizeProviderConfig(cfg provider.Config) provider.Config {
	cfg.ID = provider.ProviderID(strings.TrimSpace(strings.ToLower(string(cfg.ID))))
	cfg.Endpoint = strings.TrimSpace(cfg.Endpoint)
	cfg.CredentialRef = strings.TrimSpace(cfg.CredentialRef)
	return cfg
}

func validateProviderConfig(cfg provider.Config) error {
	if cfg.ID == "" {
		return fmt.Errorf("%w: id is required", ErrInvalidProviderConfig)
	}
	if cfg.Endpoint == "" {
		return fmt.Errorf("%w: endpoint is required", ErrInvalidProviderConfig)
	}
	if cfg.Capabilities == 0 {
		return fmt.Errorf("%w: capabilities must be non-zero", ErrInvalidProviderConfig)
	}
	return nil
}

func scanProviderConfig(scanner repoScanner) (provider.Config, error) {
	var cfg provider.Config
	var credentialRef sql.NullString
	var capabilities int64
	if err := scanner.Scan(
		&cfg.ID,
		&cfg.Endpoint,
		&credentialRef,
		&capabilities,
		&cfg.CreatedAt,
		&cfg.UpdatedAt,
	); err != nil {
		return provider.Config{}, err
	}
	cfg.Capabilities = provider.Capability(capabilities)
	if credentialRef.Valid {
		cfg.CredentialRef = credentialRef.String
	}
	return cfg, nil
}

func (s *pgProviderConfigStore) SaveProviderConfig(ctx context.Context, cfg provider.Config) (provider.Config, error) {
	cfg = normalizeProviderConfig(cfg)
	if err := validateProviderConfig(cfg); err != nil {
		return provider.Config{}, err
	}

	const q = `
		INSERT INTO provider_configs (id, endpoint, credential_ref, capabilities, created_at, updated_at)
		VALUES ($1, $2, $3, $4, now(), now())
		ON CONFLICT (id) DO UPDATE
			SET endpoint = EXCLUDED.endpoint,
			    credential_ref = EXCLUDED.credential_ref,
			    capabilities = EXCLUDED.capabilities,
			    updated_at = now()
		RETURNING id, endpoint, credential_ref, capabilities, created_at, updated_at
	`
	out, err := scanProviderConfig(s.pool.QueryRow(ctx, q,
		cfg.ID,
		cfg.Endpoint,
		nullableString(cfg.CredentialRef),
		int64(cfg.Capabilities),
	))
	if err != nil {
		return provider.Config{}, fmt.Errorf("save provider config: %w", err)
	}
	return out, nil
}

func (s *pgProviderConfigStore) GetProviderConfig(ctx context.Context, id provider.ProviderID) (provider.Config, bool, error) {
	id = provider.ProviderID(strings.TrimSpace(strings.ToLower(string(id))))
	if id == "" {
		return provider.Config{}, false, fmt.Errorf("%w: id is required", ErrInvalidProviderConfig)
	}

	const q = `SELECT id, endpoint, credential_ref, capabilities, created_at, updated_at FROM provider_configs WHERE id=$1`
	cfg, err := scanProviderConfig(s.pool.QueryRow(ctx, q, id))
	if err != nil {
		if isNoRows(err) {
			return provider.Config{}, false, nil
		}
		return provider.Config{}, false, fmt.Errorf("get provider config: %w", err)
	}
	return cfg, true, nil
}

func (s *pgProviderConfigStore) ListProviderConfigs(ctx context.Context) ([]provider.Config, error) {
	const q = `SELECT id, endpoint, credential_ref, capabilities, created_at, updated_at FROM provider_configs ORDER BY id`
	rows, err := s.pool.Query(ctx, q)
	if err != nil {
		return nil, fmt.Errorf("list provider configs: %w", err)
	}
	defer rows.Close()

	var out []provider.Config
	for rows.Next() {
		cfg, err := scanProviderConfig(rows)
		if err != nil {
			return nil, fmt.Errorf("scan provider config: %w", err)
		}
		out = append(out, cfg)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("list provider configs rows error: %w", err)
	}
	return out, nil
}