creator-hub/internal/phasea/store.go

package phasea

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

	"github.com/jackc/pgx/v5/pgconn"
	_ "github.com/jackc/pgx/v5/stdlib"
)

//go:embed migrations/001_phase_a.sql
var migration001 string

var (
	ErrConflict          = errors.New("resource conflicts with existing state")
	ErrInvalid           = errors.New("invalid phase A input")
	ErrNotFound          = errors.New("resource not found")
	idPattern            = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
	refPattern           = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
	platformKeyPattern   = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$`)
	credentialKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/[A-Za-z0-9][A-Za-z0-9._/-]{0,126}$`)
)

type Store struct{ db *sql.DB }

type Account struct {
	ID                  string              `json:"id"`
	Platform            string              `json:"platform"`
	PlatformAccountKey  string              `json:"platform_account_key"`
	AuthorizationKind   string              `json:"authorization_kind"`
	AuthorizationStatus string              `json:"authorization_status"`
	RuntimeStatus       string              `json:"runtime_status"`
	CredentialReference CredentialReference `json:"credential_reference"`
	CredentialKey       string              `json:"-"`
	Version             int64               `json:"version"`
}

type CredentialReference struct {
	ID       string `json:"id"`
	Provider string `json:"provider"`
}

type Draft struct {
	ID        string    `json:"id"`
	AccountID string    `json:"account_id"`
	Version   int64     `json:"version"`
	Content   string    `json:"content"`
	CreatedAt time.Time `json:"created_at"`
}

type Confirmation struct {
	ID             string    `json:"id"`
	AccountID      string    `json:"account_id"`
	AccountVersion int64     `json:"account_version"`
	DraftID        string    `json:"draft_id"`
	DraftVersion   int64     `json:"draft_version"`
	Version        int64     `json:"version"`
	ConfirmedAt    time.Time `json:"confirmed_at"`
}

type Task struct {
	ID                  string    `json:"id"`
	IdempotencyKey      string    `json:"idempotency_key"`
	AccountID           string    `json:"account_id"`
	AccountVersion      int64     `json:"account_version"`
	DraftID             string    `json:"draft_id"`
	DraftVersion        int64     `json:"draft_version"`
	ConfirmationID      string    `json:"confirmation_id"`
	ConfirmationVersion int64     `json:"confirmation_version"`
	State               string    `json:"state"`
	CreatedAt           time.Time `json:"created_at"`
}

type ConfirmationSnapshot struct {
	Confirmation
	BrowserEnvAlias   string `json:"browser_env_alias,omitempty"`
	NetworkExitID     string `json:"network_exit_id,omitempty"`
	RuntimeInstanceID string `json:"runtime_instance_id,omitempty"`
	BindingVersion    int64  `json:"binding_version,omitempty"`
}

type DraftDetail struct {
	Draft
	Account       Account                `json:"account"`
	Versions      []Draft                `json:"versions"`
	Confirmations []ConfirmationSnapshot `json:"confirmations"`
	Tasks         []Task                 `json:"tasks"`
}

type ReadinessError struct {
	Reason      string
	Unavailable bool
}

func (e *ReadinessError) Error() string { return e.Reason }

type Execution struct {
	TaskID              string `json:"task_id"`
	AttemptID           string `json:"attempt_id"`
	AccountID           string `json:"account_id"`
	ConfirmationID      string `json:"confirmation_id"`
	ConfirmationVersion int64  `json:"confirmation_version"`
	State               string `json:"state"`
	WasClaimed          bool   `json:"was_claimed"`
}

type AuditEvent struct {
	ID                  int64           `json:"id"`
	EventType           string          `json:"event_type"`
	AccountID           string          `json:"account_id,omitempty"`
	ConfirmationID      string          `json:"confirmation_id,omitempty"`
	ConfirmationVersion int64           `json:"confirmation_version,omitempty"`
	AttemptID           string          `json:"attempt_id,omitempty"`
	TaskID              string          `json:"task_id,omitempty"`
	BrowserEnvAlias     string          `json:"browser_env_alias,omitempty"`
	NetworkExitID       string          `json:"network_exit_id,omitempty"`
	RuntimeInstanceID   string          `json:"runtime_instance_id,omitempty"`
	BindingVersion      int64           `json:"binding_version,omitempty"`
	Actor               string          `json:"actor,omitempty"`
	ReasonCode          string          `json:"reason_code,omitempty"`
	OperationID         string          `json:"operation_id,omitempty"`
	Action              string          `json:"action,omitempty"`
	Outcome             string          `json:"outcome,omitempty"`
	OldImageVersion     string          `json:"old_image_version,omitempty"`
	NewImageVersion     string          `json:"new_image_version,omitempty"`
	Details             json.RawMessage `json:"details"`
	CreatedAt           time.Time       `json:"created_at"`
}

func Open(ctx context.Context, databaseURL string) (*Store, error) {
	db, err := sql.Open("pgx", databaseURL)
	if err != nil {
		return nil, errors.New("open phase A database")
	}
	db.SetMaxOpenConns(10)
	db.SetMaxIdleConns(2)
	db.SetConnMaxIdleTime(5 * time.Minute)
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		return nil, errors.New("connect to phase A database")
	}
	store := &Store{db: db}
	if err := store.migrate(ctx); err != nil {
		db.Close()
		return nil, err
	}
	return store, nil
}

func (s *Store) Close() error { return s.db.Close() }

func (s *Store) migrate(ctx context.Context) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin schema migration")
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(1542738011)`); err != nil {
		return errors.New("lock schema migration")
	}
	if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
		return errors.New("create schema migration table")
	}
	var applied bool
	if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = 1)`).Scan(&applied); err != nil {
		return errors.New("read schema migration state")
	}
	if !applied {
		if _, err := tx.ExecContext(ctx, migration001); err != nil {
			return fmt.Errorf("apply schema migration 1: %w", err)
		}
		if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES (1)`); err != nil {
			return errors.New("record schema migration 1")
		}
	}
	if err := tx.Commit(); err != nil {
		return errors.New("commit schema migration")
	}
	return nil
}

func (s *Store) CreateAccount(ctx context.Context, account Account) error {
	if !idPattern.MatchString(account.ID) || !idPattern.MatchString(account.Platform) ||
		!platformKeyPattern.MatchString(account.PlatformAccountKey) ||
		(account.AuthorizationKind != "owned" && account.AuthorizationKind != "authorized") ||
		!refPattern.MatchString(account.CredentialReference.ID) || !credentialKeyPattern.MatchString(account.CredentialKey) ||
		(account.CredentialReference.Provider != "os_keyring" && account.CredentialReference.Provider != "secret_manager") {
		return ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin account transaction")
	}
	defer tx.Rollback()
	if _, err := tx.ExecContext(ctx, `INSERT INTO credential_reference (id, provider, reference_key) VALUES ($1, $2, $3)`,
		account.CredentialReference.ID, account.CredentialReference.Provider, account.CredentialKey); err != nil {
		return publicDatabaseError(err)
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO social_account
			(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status, status)
		VALUES ($1, $2, $3, $4, $5, 'authorized', 'paused')`, account.ID, account.CredentialReference.ID,
		account.Platform, account.PlatformAccountKey, account.AuthorizationKind); err != nil {
		return publicDatabaseError(err)
	}
	if err := appendAudit(ctx, tx, "account_created", "account_created", account.ID, "", 0, "", "", map[string]string{"platform": account.Platform}); err != nil {
		return err
	}
	return commit(tx)
}

func (s *Store) ListAccounts(ctx context.Context) ([]Account, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
			account.authorization_status, account.status, reference.id, reference.provider, account.version
		FROM social_account account
		JOIN credential_reference reference ON reference.id = account.credential_reference_id
		ORDER BY account.created_at, account.id`)
	if err != nil {
		return nil, errors.New("read accounts")
	}
	defer rows.Close()
	accounts := []Account{}
	for rows.Next() {
		account, err := scanAccount(rows)
		if err != nil {
			return nil, err
		}
		accounts = append(accounts, account)
	}
	return accounts, rows.Err()
}

func (s *Store) GetAccount(ctx context.Context, id string) (Account, error) {
	if !idPattern.MatchString(id) {
		return Account{}, ErrInvalid
	}
	return scanAccount(s.db.QueryRowContext(ctx, `
		SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
			account.authorization_status, account.status, reference.id, reference.provider, account.version
		FROM social_account account
		JOIN credential_reference reference ON reference.id = account.credential_reference_id
		WHERE account.id = $1`, id))
}

type accountScanner interface{ Scan(...any) error }

func scanAccount(row accountScanner) (Account, error) {
	var account Account
	if err := row.Scan(&account.ID, &account.Platform, &account.PlatformAccountKey, &account.AuthorizationKind,
		&account.AuthorizationStatus, &account.RuntimeStatus, &account.CredentialReference.ID,
		&account.CredentialReference.Provider, &account.Version); err != nil {
		return Account{}, rowError(err)
	}
	return account, nil
}

func (s *Store) CreateDraft(ctx context.Context, draft Draft) error {
	if !refPattern.MatchString(draft.ID) || !idPattern.MatchString(draft.AccountID) || draft.Version < 1 || strings.TrimSpace(draft.Content) == "" {
		return ErrInvalid
	}
	_, err := s.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ($1, $2, $3, $4)`,
		draft.ID, draft.AccountID, draft.Version, draft.Content)
	return publicDatabaseError(err)
}

func (s *Store) CreateDraftVersion(ctx context.Context, accountID, content string) (Draft, error) {
	if !idPattern.MatchString(accountID) || strings.TrimSpace(content) == "" {
		return Draft{}, ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Draft{}, errors.New("begin draft transaction")
	}
	defer tx.Rollback()
	if err := tx.QueryRowContext(ctx, `SELECT id FROM social_account WHERE id = $1 FOR UPDATE`, accountID).Scan(&accountID); err != nil {
		return Draft{}, rowError(err)
	}
	draft := Draft{ID: "draft-" + newID()[:24], AccountID: accountID, Content: content}
	if err := tx.QueryRowContext(ctx, `
		INSERT INTO content_draft (id, account_id, version, content)
		SELECT $1, $2, COALESCE(max(version), 0) + 1, $3 FROM content_draft WHERE account_id = $2
		RETURNING version, created_at`, draft.ID, accountID, content).Scan(&draft.Version, &draft.CreatedAt); err != nil {
		return Draft{}, publicDatabaseError(err)
	}
	if err := appendAudit(ctx, tx, "draft_created", "draft_created", accountID, "", 0, "", "", map[string]any{
		"draft_id": draft.ID, "draft_version": draft.Version,
	}); err != nil {
		return Draft{}, err
	}
	if err := commit(tx); err != nil {
		return Draft{}, err
	}
	return draft, nil
}

func (s *Store) ListDrafts(ctx context.Context, accountID string) ([]Draft, error) {
	if accountID != "" && !idPattern.MatchString(accountID) {
		return nil, ErrInvalid
	}
	rows, err := s.db.QueryContext(ctx, `
		SELECT id, account_id, version, content, created_at FROM content_draft
		WHERE $1 = '' OR account_id = $1 ORDER BY account_id, version DESC, created_at DESC`, accountID)
	if err != nil {
		return nil, errors.New("read drafts")
	}
	defer rows.Close()
	drafts := []Draft{}
	for rows.Next() {
		draft, err := scanDraft(rows)
		if err != nil {
			return nil, err
		}
		drafts = append(drafts, draft)
	}
	return drafts, rows.Err()
}

func (s *Store) GetDraft(ctx context.Context, id string) (Draft, error) {
	if !refPattern.MatchString(id) {
		return Draft{}, ErrInvalid
	}
	return scanDraft(s.db.QueryRowContext(ctx, `
		SELECT id, account_id, version, content, created_at FROM content_draft WHERE id = $1`, id))
}

type draftScanner interface{ Scan(...any) error }

func scanDraft(row draftScanner) (Draft, error) {
	var draft Draft
	if err := row.Scan(&draft.ID, &draft.AccountID, &draft.Version, &draft.Content, &draft.CreatedAt); err != nil {
		return Draft{}, rowError(err)
	}
	return draft, nil
}

func (s *Store) GetDraftDetail(ctx context.Context, id string) (DraftDetail, error) {
	draft, err := s.GetDraft(ctx, id)
	if err != nil {
		return DraftDetail{}, err
	}
	account, err := s.GetAccount(ctx, draft.AccountID)
	if err != nil {
		return DraftDetail{}, err
	}
	versions, err := s.ListDrafts(ctx, draft.AccountID)
	if err != nil {
		return DraftDetail{}, err
	}
	confirmations, err := s.ListConfirmations(ctx, draft.ID)
	if err != nil {
		return DraftDetail{}, err
	}
	tasks, err := s.ListTasks(ctx, draft.AccountID, draft.ID)
	if err != nil {
		return DraftDetail{}, err
	}
	return DraftDetail{Draft: draft, Account: account, Versions: versions, Confirmations: confirmations, Tasks: tasks}, nil
}

func (s *Store) ConfirmDraft(ctx context.Context, draftID string, accountVersion, draftVersion int64) (Confirmation, bool, error) {
	if !refPattern.MatchString(draftID) || accountVersion < 1 || draftVersion < 1 {
		return Confirmation{}, false, ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Confirmation{}, false, errors.New("begin confirmation transaction")
	}
	defer tx.Rollback()
	var accountID string
	var currentAccountVersion, currentDraftVersion, latestDraftVersion int64
	if err := tx.QueryRowContext(ctx, `
		SELECT draft.account_id, account.version, draft.version,
			(SELECT max(version) FROM content_draft WHERE account_id = draft.account_id)
		FROM content_draft draft JOIN social_account account ON account.id = draft.account_id
		WHERE draft.id = $1 FOR UPDATE OF draft, account`, draftID).
		Scan(&accountID, &currentAccountVersion, &currentDraftVersion, &latestDraftVersion); err != nil {
		return Confirmation{}, false, rowError(err)
	}
	if accountVersion != currentAccountVersion {
		return Confirmation{}, false, &ReadinessError{Reason: "account_version_changed"}
	}
	if draftVersion != currentDraftVersion || draftVersion != latestDraftVersion {
		return Confirmation{}, false, &ReadinessError{Reason: "draft_version_changed"}
	}
	var existing Confirmation
	err = tx.QueryRowContext(ctx, `
		SELECT id, account_id, account_version, draft_id, draft_version, version, confirmed_at
		FROM confirmation WHERE account_id = $1 AND account_version = $2 AND draft_id = $3 AND draft_version = $4
		ORDER BY version DESC LIMIT 1`, accountID, accountVersion, draftID, draftVersion).
		Scan(&existing.ID, &existing.AccountID, &existing.AccountVersion, &existing.DraftID, &existing.DraftVersion, &existing.Version, &existing.ConfirmedAt)
	if err == nil {
		if err := commit(tx); err != nil {
			return Confirmation{}, false, err
		}
		return existing, false, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return Confirmation{}, false, publicDatabaseError(err)
	}
	confirmation := Confirmation{
		ID: "confirmation-" + newID()[:20], AccountID: accountID, AccountVersion: accountVersion,
		DraftID: draftID, DraftVersion: draftVersion,
	}
	if err := tx.QueryRowContext(ctx, `
		INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
		SELECT $1, $2, $3, $4, $5, COALESCE(max(version), 0) + 1 FROM confirmation WHERE draft_id = $4
		RETURNING version, confirmed_at`, confirmation.ID, accountID, accountVersion, draftID, draftVersion).
		Scan(&confirmation.Version, &confirmation.ConfirmedAt); err != nil {
		return Confirmation{}, false, publicDatabaseError(err)
	}
	if err := appendAudit(ctx, tx, "draft_confirmed", "draft_confirmed", accountID, confirmation.ID, confirmation.Version, "", "", map[string]any{
		"account_version": accountVersion, "draft_id": draftID, "draft_version": draftVersion,
	}); err != nil {
		return Confirmation{}, false, err
	}
	if err := commit(tx); err != nil {
		return Confirmation{}, false, err
	}
	return confirmation, true, nil
}

func (s *Store) Confirm(ctx context.Context, confirmation Confirmation) error {
	if !refPattern.MatchString(confirmation.ID) || !idPattern.MatchString(confirmation.AccountID) ||
		!refPattern.MatchString(confirmation.DraftID) || confirmation.AccountVersion < 1 || confirmation.DraftVersion < 1 || confirmation.Version < 1 {
		return ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin confirmation transaction")
	}
	defer tx.Rollback()
	var valid bool
	if err := tx.QueryRowContext(ctx, `
		SELECT EXISTS (
			SELECT 1 FROM social_account a JOIN content_draft d ON d.account_id = a.id
			WHERE a.id = $1 AND a.version = $2 AND d.id = $3 AND d.version = $4
		)`, confirmation.AccountID, confirmation.AccountVersion, confirmation.DraftID, confirmation.DraftVersion).Scan(&valid); err != nil {
		return errors.New("validate confirmation versions")
	}
	if !valid {
		return ErrConflict
	}
	if _, err := tx.ExecContext(ctx, `
		INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
		VALUES ($1, $2, $3, $4, $5, $6)`, confirmation.ID, confirmation.AccountID, confirmation.AccountVersion,
		confirmation.DraftID, confirmation.DraftVersion, confirmation.Version); err != nil {
		return publicDatabaseError(err)
	}
	if err := appendAudit(ctx, tx, "draft_confirmed", "draft_confirmed", confirmation.AccountID, confirmation.ID, confirmation.Version, "", "", map[string]any{
		"account_version": confirmation.AccountVersion, "draft_id": confirmation.DraftID, "draft_version": confirmation.DraftVersion,
	}); err != nil {
		return err
	}
	return commit(tx)
}

func (s *Store) ListConfirmations(ctx context.Context, draftID string) ([]ConfirmationSnapshot, error) {
	if draftID != "" && !refPattern.MatchString(draftID) {
		return nil, ErrInvalid
	}
	rows, err := s.db.QueryContext(ctx, `
		SELECT confirmation.id, confirmation.account_id, confirmation.account_version, confirmation.draft_id,
			confirmation.draft_version, confirmation.version, confirmation.confirmed_at,
			audit.browser_env_alias, audit.network_exit_id, audit.runtime_instance_id, audit.binding_version
		FROM confirmation
		LEFT JOIN LATERAL (
			SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
			FROM audit_event WHERE confirmation_id = confirmation.id AND event_type = 'draft_confirmed'
			ORDER BY id DESC LIMIT 1
		) audit ON true
		WHERE $1 = '' OR confirmation.draft_id = $1
		ORDER BY confirmation.confirmed_at DESC, confirmation.version DESC`, draftID)
	if err != nil {
		return nil, errors.New("read confirmations")
	}
	defer rows.Close()
	confirmations := []ConfirmationSnapshot{}
	for rows.Next() {
		confirmation, err := scanConfirmation(rows)
		if err != nil {
			return nil, err
		}
		confirmations = append(confirmations, confirmation)
	}
	return confirmations, rows.Err()
}

func (s *Store) GetConfirmation(ctx context.Context, id string) (ConfirmationSnapshot, error) {
	if !refPattern.MatchString(id) {
		return ConfirmationSnapshot{}, ErrInvalid
	}
	return scanConfirmation(s.db.QueryRowContext(ctx, `
		SELECT confirmation.id, confirmation.account_id, confirmation.account_version, confirmation.draft_id,
			confirmation.draft_version, confirmation.version, confirmation.confirmed_at,
			audit.browser_env_alias, audit.network_exit_id, audit.runtime_instance_id, audit.binding_version
		FROM confirmation
		LEFT JOIN LATERAL (
			SELECT browser_env_alias, network_exit_id, runtime_instance_id, binding_version
			FROM audit_event WHERE confirmation_id = confirmation.id AND event_type = 'draft_confirmed'
			ORDER BY id DESC LIMIT 1
		) audit ON true WHERE confirmation.id = $1`, id))
}

type confirmationScanner interface{ Scan(...any) error }

func scanConfirmation(row confirmationScanner) (ConfirmationSnapshot, error) {
	var confirmation ConfirmationSnapshot
	var browser, network, runtime sql.NullString
	var bindingVersion sql.NullInt64
	if err := row.Scan(&confirmation.ID, &confirmation.AccountID, &confirmation.AccountVersion, &confirmation.DraftID,
		&confirmation.DraftVersion, &confirmation.Version, &confirmation.ConfirmedAt,
		&browser, &network, &runtime, &bindingVersion); err != nil {
		return ConfirmationSnapshot{}, rowError(err)
	}
	confirmation.BrowserEnvAlias, confirmation.NetworkExitID = browser.String, network.String
	confirmation.RuntimeInstanceID, confirmation.BindingVersion = runtime.String, bindingVersion.Int64
	return confirmation, nil
}

func (s *Store) ListTasks(ctx context.Context, accountID, draftID string) ([]Task, error) {
	if (accountID != "" && !idPattern.MatchString(accountID)) || (draftID != "" && !refPattern.MatchString(draftID)) {
		return nil, ErrInvalid
	}
	rows, err := s.db.QueryContext(ctx, `
		SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
			confirmation_id, confirmation_version, state, created_at
		FROM operation_task WHERE ($1 = '' OR account_id = $1) AND ($2 = '' OR draft_id = $2)
		ORDER BY created_at DESC, id`, accountID, draftID)
	if err != nil {
		return nil, errors.New("read tasks")
	}
	defer rows.Close()
	tasks := []Task{}
	for rows.Next() {
		task, err := scanTask(rows)
		if err != nil {
			return nil, err
		}
		tasks = append(tasks, task)
	}
	return tasks, rows.Err()
}

func (s *Store) GetTask(ctx context.Context, id string) (Task, error) {
	if !refPattern.MatchString(id) {
		return Task{}, ErrInvalid
	}
	return scanTask(s.db.QueryRowContext(ctx, `
		SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
			confirmation_id, confirmation_version, state, created_at
		FROM operation_task WHERE id = $1`, id))
}

type taskScanner interface{ Scan(...any) error }

func scanTask(row taskScanner) (Task, error) {
	var task Task
	var confirmationID sql.NullString
	var confirmationVersion sql.NullInt64
	if err := row.Scan(&task.ID, &task.IdempotencyKey, &task.AccountID, &task.AccountVersion, &task.DraftID,
		&task.DraftVersion, &confirmationID, &confirmationVersion, &task.State, &task.CreatedAt); err != nil {
		return Task{}, rowError(err)
	}
	task.ConfirmationID, task.ConfirmationVersion = confirmationID.String, confirmationVersion.Int64
	return task, nil
}

func (s *Store) EnqueueConfirmation(ctx context.Context, confirmationID string) (Task, bool, error) {
	if !refPattern.MatchString(confirmationID) {
		return Task{}, false, ErrInvalid
	}
	sum := sha256.Sum256([]byte(confirmationID))
	idempotencyKey := "enqueue-" + hex.EncodeToString(sum[:])
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Task{}, false, errors.New("begin confirmed enqueue transaction")
	}
	defer tx.Rollback()
	if existing, err := scanTask(tx.QueryRowContext(ctx, `
		SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
			confirmation_id, confirmation_version, state, created_at
		FROM operation_task WHERE idempotency_key = $1`, idempotencyKey)); err == nil {
		if err := commit(tx); err != nil {
			return Task{}, false, err
		}
		return existing, false, nil
	} else if !errors.Is(err, ErrNotFound) {
		return Task{}, false, err
	}

	var task Task
	var currentAccountVersion, currentDraftVersion, latestDraftVersion int64
	var authorizationStatus, accountStatus string
	if err := tx.QueryRowContext(ctx, `
		SELECT confirmation.account_id, confirmation.account_version, confirmation.draft_id,
			confirmation.draft_version, confirmation.version, account.version, account.authorization_status,
			account.status, draft.version, (SELECT max(version) FROM content_draft WHERE account_id = confirmation.account_id)
		FROM confirmation
		JOIN social_account account ON account.id = confirmation.account_id
		JOIN content_draft draft ON draft.id = confirmation.draft_id
		WHERE confirmation.id = $1 FOR UPDATE OF confirmation, account, draft`, confirmationID).
		Scan(&task.AccountID, &task.AccountVersion, &task.DraftID, &task.DraftVersion, &task.ConfirmationVersion,
			&currentAccountVersion, &authorizationStatus, &accountStatus, &currentDraftVersion, &latestDraftVersion); err != nil {
		return Task{}, false, rowError(err)
	}
	task.ConfirmationID, task.IdempotencyKey, task.ID = confirmationID, idempotencyKey, "task-"+newID()[:24]
	if task.AccountVersion != currentAccountVersion {
		return Task{}, false, &ReadinessError{Reason: "account_version_changed"}
	}
	if task.DraftVersion != currentDraftVersion || task.DraftVersion != latestDraftVersion {
		return Task{}, false, &ReadinessError{Reason: "draft_version_changed"}
	}
	if authorizationStatus != "authorized" {
		return Task{}, false, &ReadinessError{Reason: "account_revoked"}
	}
	if accountStatus != "active" {
		return Task{}, false, &ReadinessError{Reason: "account_paused"}
	}
	var bindingID string
	var networkExitID sql.NullString
	var bindingVersion int64
	var cleanupPending bool
	if err := tx.QueryRowContext(ctx, `
		SELECT id, network_exit_id, version, runtime_cleanup_pending
		FROM environment_binding WHERE account_id = $1 FOR SHARE`, task.AccountID).
		Scan(&bindingID, &networkExitID, &bindingVersion, &cleanupPending); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Task{}, false, &ReadinessError{Reason: "binding_missing", Unavailable: true}
		}
		return Task{}, false, publicDatabaseError(err)
	}
	if !networkExitID.Valid {
		return Task{}, false, &ReadinessError{Reason: "network_exit_missing", Unavailable: true}
	}
	var exitStatus string
	if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR SHARE`, networkExitID.String).Scan(&exitStatus); err != nil {
		return Task{}, false, publicDatabaseError(err)
	}
	if exitStatus != "healthy" {
		return Task{}, false, &ReadinessError{Reason: "network_exit_unhealthy", Unavailable: true}
	}
	if cleanupPending {
		return Task{}, false, &ReadinessError{Reason: "runtime_stop_pending", Unavailable: true}
	}
	var runtimeID string
	var runtimeBindingVersion int64
	var leaseActive bool
	if err := tx.QueryRowContext(ctx, `
		SELECT id, binding_version, lease_until > now() FROM runtime_instance
		WHERE binding_id = $1 AND released_at IS NULL ORDER BY acquired_at DESC LIMIT 1 FOR SHARE`, bindingID).
		Scan(&runtimeID, &runtimeBindingVersion, &leaseActive); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Task{}, false, &ReadinessError{Reason: "runtime_missing", Unavailable: true}
		}
		return Task{}, false, publicDatabaseError(err)
	}
	if !leaseActive {
		return Task{}, false, &ReadinessError{Reason: "runtime_lease_expired", Unavailable: true}
	}
	if runtimeBindingVersion != bindingVersion {
		return Task{}, false, &ReadinessError{Reason: "binding_version_changed"}
	}
	created, inserted, err := enqueueTask(ctx, tx, task)
	if err != nil {
		return Task{}, false, err
	}
	if err := commit(tx); err != nil {
		return Task{}, false, err
	}
	return created, inserted, nil
}

func (s *Store) Enqueue(ctx context.Context, task Task) (Task, bool, error) {
	if !refPattern.MatchString(task.ID) || !refPattern.MatchString(task.IdempotencyKey) || !idPattern.MatchString(task.AccountID) ||
		!refPattern.MatchString(task.DraftID) || task.AccountVersion < 1 || task.DraftVersion < 1 ||
		((task.ConfirmationID == "") != (task.ConfirmationVersion == 0)) ||
		(task.ConfirmationID != "" && (!refPattern.MatchString(task.ConfirmationID) || task.ConfirmationVersion < 1)) {
		return Task{}, false, ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Task{}, false, errors.New("begin task transaction")
	}
	defer tx.Rollback()
	created, inserted, err := enqueueTask(ctx, tx, task)
	if err != nil {
		return Task{}, false, err
	}
	if err := commit(tx); err != nil {
		return Task{}, false, err
	}
	return created, inserted, nil
}

func enqueueTask(ctx context.Context, tx *sql.Tx, task Task) (Task, bool, error) {
	var insertedID string
	err := tx.QueryRowContext(ctx, `
		INSERT INTO operation_task (id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
		VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, 0))
		ON CONFLICT (idempotency_key) DO NOTHING RETURNING id, created_at`, task.ID, task.IdempotencyKey, task.AccountID, task.AccountVersion,
		task.DraftID, task.DraftVersion, task.ConfirmationID, task.ConfirmationVersion).Scan(&insertedID, &task.CreatedAt)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return Task{}, false, publicDatabaseError(err)
	}
	if insertedID != "" {
		task.State = "queued"
		if err := appendAudit(ctx, tx, "task_queued", "task_queued", task.AccountID, task.ConfirmationID, task.ConfirmationVersion, "", task.ID, nil); err != nil {
			return Task{}, false, err
		}
		return task, true, nil
	}
	existing, err := scanTask(tx.QueryRowContext(ctx, `
		SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version,
			confirmation_id, confirmation_version, state, created_at
		FROM operation_task WHERE idempotency_key = $1`, task.IdempotencyKey))
	if err != nil {
		return Task{}, false, err
	}
	if existing.AccountID != task.AccountID || existing.AccountVersion != task.AccountVersion || existing.DraftID != task.DraftID ||
		existing.DraftVersion != task.DraftVersion || existing.ConfirmationID != task.ConfirmationID || existing.ConfirmationVersion != task.ConfirmationVersion {
		return Task{}, false, ErrConflict
	}
	return existing, false, nil
}

func (s *Store) PauseAccount(ctx context.Context, accountID string) error {
	return s.disableAccount(ctx, accountID, false)
}

func (s *Store) RevokeAccount(ctx context.Context, accountID string) error {
	return s.disableAccount(ctx, accountID, true)
}

func (s *Store) disableAccount(ctx context.Context, accountID string, revoke bool) error {
	if !idPattern.MatchString(accountID) {
		return ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin account state transaction")
	}
	defer tx.Rollback()
	var version int64
	var authorizationStatus, runtimeStatus string
	if err := tx.QueryRowContext(ctx, `SELECT version, authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
		Scan(&version, &authorizationStatus, &runtimeStatus); err != nil {
		return rowError(err)
	}
	unchanged := (revoke && authorizationStatus == "revoked") || (!revoke && runtimeStatus == "paused")
	if !unchanged {
		if err := tx.QueryRowContext(ctx, `
			UPDATE social_account
			SET authorization_status = CASE WHEN $2 THEN 'revoked' ELSE authorization_status END,
				status = 'paused', paused_at = now(), revoked_at = CASE WHEN $2 THEN now() ELSE revoked_at END,
				version = version + 1, updated_at = now()
			WHERE id = $1 RETURNING version`, accountID, revoke).Scan(&version); err != nil {
			return errors.New("change account state")
		}
	}
	held, err := holdQueuedTasks(ctx, tx, accountID)
	if err != nil {
		return err
	}
	reason := "account_paused"
	if revoke {
		reason = "account_revoked"
	}
	interrupted, err := interruptExecutingTasks(ctx, tx, accountID, reason)
	if err != nil {
		return err
	}
	if unchanged {
		return commit(tx)
	}
	eventType := "account_paused"
	if revoke {
		eventType = "account_revoked"
	}
	if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{
		"account_version": version, "tasks_held": held, "attempts_interrupted": interrupted,
	}); err != nil {
		return err
	}
	return commit(tx)
}

func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
	if !idPattern.MatchString(accountID) {
		return ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin resume transaction")
	}
	defer tx.Rollback()
	var authorizationStatus, runtimeStatus string
	if err := tx.QueryRowContext(ctx, `SELECT authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
		Scan(&authorizationStatus, &runtimeStatus); err != nil {
		return rowError(err)
	}
	if authorizationStatus == "revoked" {
		return ErrConflict
	}
	var ready bool
	if err := tx.QueryRowContext(ctx, `
		SELECT EXISTS (
			SELECT 1 FROM environment_binding binding
			JOIN network_exit network ON network.id = binding.network_exit_id
			WHERE binding.account_id = $1 AND network.health_status = 'healthy'
				AND NOT binding.runtime_cleanup_pending
				AND NOT EXISTS (
					SELECT 1 FROM runtime_instance runtime
					WHERE runtime.binding_id = binding.id AND runtime.released_at IS NULL
				)
		)`, accountID).Scan(&ready); err != nil {
		return errors.New("validate account binding")
	}
	if !ready {
		return ErrConflict
	}
	if runtimeStatus == "active" {
		return commit(tx)
	}
	var version int64
	if err := tx.QueryRowContext(ctx, `
		UPDATE social_account SET status = 'active', paused_at = NULL, version = version + 1, updated_at = now()
		WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil {
		return errors.New("resume account")
	}
	if err := appendAudit(ctx, tx, "account_resumed", "account_resumed", accountID, "", 0, "", "", map[string]any{"account_version": version}); err != nil {
		return err
	}
	return commit(tx)
}

func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID string) (int64, error) {
	result, err := tx.ExecContext(ctx, `
		UPDATE operation_task SET state = 'policy_hold', updated_at = now()
		WHERE account_id = $1 AND state = 'queued'`, accountID)
	if err != nil {
		return 0, errors.New("hold queued account tasks")
	}
	held, _ := result.RowsAffected()
	return held, nil
}

func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) (int64, error) {
	rows, err := tx.QueryContext(ctx, `
		UPDATE operation_task SET state = 'needs_confirmation', lease_owner = NULL, lease_until = NULL, updated_at = now()
		WHERE account_id = $1 AND state = 'executing'
		RETURNING id, current_attempt_id, confirmation_id, confirmation_version`, accountID)
	if err != nil {
		return 0, errors.New("interrupt executing account tasks")
	}
	type interruptedTask struct {
		taskID, attemptID, confirmationID string
		confirmationVersion               int64
	}
	var tasks []interruptedTask
	for rows.Next() {
		var task interruptedTask
		var attemptID, confirmationID sql.NullString
		var confirmationVersion sql.NullInt64
		if err := rows.Scan(&task.taskID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
			rows.Close()
			return 0, errors.New("read interrupted account task")
		}
		task.attemptID, task.confirmationID, task.confirmationVersion = attemptID.String, confirmationID.String, confirmationVersion.Int64
		tasks = append(tasks, task)
	}
	if err := rows.Close(); err != nil {
		return 0, errors.New("close interrupted account task rows")
	}
	for _, task := range tasks {
		if task.attemptID != "" {
			if _, err := tx.ExecContext(ctx, `
				UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain'
				WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil {
				return 0, errors.New("mark interrupted attempt uncertain")
			}
		}
		if err := appendAudit(ctx, tx, "task_interrupted", reason, accountID, task.confirmationID,
			task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
			return 0, err
		}
	}
	return int64(len(tasks)), nil
}

func (s *Store) CancelTask(ctx context.Context, taskID string) error {
	if !refPattern.MatchString(taskID) {
		return ErrInvalid
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.New("begin cancel transaction")
	}
	defer tx.Rollback()
	var accountID, state string
	var attemptID sql.NullString
	var confirmationID sql.NullString
	var confirmationVersion sql.NullInt64
	err = tx.QueryRowContext(ctx, `
		UPDATE operation_task SET state = CASE WHEN state = 'executing' THEN 'needs_confirmation' ELSE 'cancelled' END,
			lease_owner = NULL, lease_until = NULL, updated_at = now()
		WHERE id = $1 AND state IN ('queued', 'executing', 'needs_confirmation', 'policy_hold')
		RETURNING account_id, state, current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan(
		&accountID, &state, &attemptID, &confirmationID, &confirmationVersion)
	if err != nil {
		return rowError(err)
	}
	if attemptID.Valid {
		if _, err := tx.ExecContext(ctx, `
			UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain'
			WHERE id = $1 AND finished_at IS NULL`, attemptID.String); err != nil {
			return errors.New("mark cancelled attempt uncertain")
		}
	}
	if err := appendAudit(ctx, tx, "task_cancel_requested", "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil {
		return err
	}
	return commit(tx)
}

func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) {
	if !refPattern.MatchString(workerID) || (outcome != "succeeded" && outcome != "failed" && outcome != "uncertain" && outcome != "policy_hold") {
		return Execution{}, ErrInvalid
	}
	execution, err := s.claim(ctx, workerID)
	if err != nil || !execution.WasClaimed {
		return execution, err
	}
	return s.complete(ctx, workerID, execution, outcome)
}

func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Execution{}, errors.New("begin claim transaction")
	}
	defer tx.Rollback()
	if err := quarantineExpired(ctx, tx); err != nil {
		return Execution{}, err
	}
	if err := quarantineInvalid(ctx, tx); err != nil {
		return Execution{}, err
	}
	var execution Execution
	err = tx.QueryRowContext(ctx, `
		WITH candidate AS (
			SELECT t.id
			FROM operation_task t
			JOIN social_account a ON a.id = t.account_id
			JOIN content_draft d ON d.id = t.draft_id
			JOIN confirmation c ON c.id = t.confirmation_id
			JOIN environment_binding binding ON binding.account_id = a.id
			JOIN browser_env environment ON environment.alias = binding.browser_env_alias
			JOIN network_exit network ON network.id = binding.network_exit_id
			JOIN runtime_instance runtime ON runtime.binding_id = binding.id
				AND runtime.released_at IS NULL AND runtime.lease_until > now() AND runtime.binding_version = binding.version
			WHERE t.state = 'queued' AND a.status = 'active' AND a.authorization_status = 'authorized'
			  AND a.version = t.account_version
			  AND d.account_id = t.account_id AND d.version = t.draft_version
			  AND c.account_id = t.account_id AND c.account_version = t.account_version
			  AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
			  AND c.version = t.confirmation_version
			  AND network.health_status = 'healthy' AND NOT binding.runtime_cleanup_pending
			ORDER BY t.created_at, t.id
			FOR UPDATE OF t, a, binding, network, runtime SKIP LOCKED LIMIT 1
		)
		UPDATE operation_task t SET state = 'executing', lease_owner = $1,
			lease_until = now() + interval '1 minute', updated_at = now()
		FROM candidate WHERE t.id = candidate.id
		RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version`, workerID).Scan(
		&execution.TaskID, &execution.AccountID, &execution.ConfirmationID, &execution.ConfirmationVersion)
	if errors.Is(err, sql.ErrNoRows) {
		if err := commit(tx); err != nil {
			return Execution{}, err
		}
		return Execution{}, nil
	}
	if err != nil {
		return Execution{}, errors.New("claim task")
	}
	execution.AttemptID = newID()
	execution.State = "executing"
	execution.WasClaimed = true
	if _, err := tx.ExecContext(ctx, `INSERT INTO execution_attempt (id, task_id) VALUES ($1, $2)`, execution.AttemptID, execution.TaskID); err != nil {
		return Execution{}, publicDatabaseError(err)
	}
	if _, err := tx.ExecContext(ctx, `UPDATE operation_task SET current_attempt_id = $1 WHERE id = $2`, execution.AttemptID, execution.TaskID); err != nil {
		return Execution{}, errors.New("attach execution attempt")
	}
	if err := appendAudit(ctx, tx, "task_claimed", "task_claimed", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"worker_id": workerID}); err != nil {
		return Execution{}, err
	}
	if err := commit(tx); err != nil {
		return Execution{}, err
	}
	return execution, nil
}

func (s *Store) complete(ctx context.Context, workerID string, execution Execution, outcome string) (Execution, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return Execution{}, errors.New("begin completion transaction")
	}
	defer tx.Rollback()
	var leaseUntil time.Time
	err = tx.QueryRowContext(ctx, `
		SELECT lease_until FROM operation_task
		WHERE id = $1 AND state = 'executing' AND lease_owner = $2 AND current_attempt_id = $3
		FOR UPDATE`, execution.TaskID, workerID, execution.AttemptID).Scan(&leaseUntil)
	if err != nil {
		return Execution{}, rowError(err)
	}
	state := map[string]string{"succeeded": "succeeded", "failed": "failed", "uncertain": "needs_confirmation", "policy_hold": "policy_hold"}[outcome]
	if leaseUntil.Before(time.Now()) {
		outcome, state = "uncertain", "needs_confirmation"
	}
	result, _ := json.Marshal(map[string]string{"mock_outcome": outcome})
	if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = $1, result = $2 WHERE id = $3`, outcome, result, execution.AttemptID); err != nil {
		return Execution{}, errors.New("finish execution attempt")
	}
	if _, err := tx.ExecContext(ctx, `
		UPDATE operation_task SET state = $1, lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $2`, state, execution.TaskID); err != nil {
		return Execution{}, errors.New("finish task")
	}
	reason := map[string]string{
		"succeeded": "task_succeeded", "failed": "task_failed",
		"needs_confirmation": "task_result_uncertain", "policy_hold": "task_policy_hold",
	}[state]
	if err := appendAudit(ctx, tx, "task_finished", reason, execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"state": state}); err != nil {
		return Execution{}, err
	}
	if err := commit(tx); err != nil {
		return Execution{}, err
	}
	execution.State = state
	return execution, nil
}

func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
	rows, err := tx.QueryContext(ctx, `
		UPDATE operation_task SET state = 'needs_confirmation', lease_owner = NULL, lease_until = NULL, updated_at = now()
		WHERE state = 'executing' AND lease_until < now()
		RETURNING id, account_id, current_attempt_id, confirmation_id, confirmation_version`)
	if err != nil {
		return errors.New("quarantine expired tasks")
	}
	type expiredTask struct {
		taskID, accountID, attemptID, confirmationID string
		confirmationVersion                          int64
	}
	var expired []expiredTask
	for rows.Next() {
		var taskID, accountID string
		var attemptID sql.NullString
		var confirmationID sql.NullString
		var confirmationVersion sql.NullInt64
		if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
			rows.Close()
			return errors.New("read expired task")
		}
		expired = append(expired, expiredTask{taskID, accountID, attemptID.String, confirmationID.String, confirmationVersion.Int64})
	}
	if err := rows.Close(); err != nil {
		return errors.New("close expired task rows")
	}
	for _, task := range expired {
		if task.attemptID != "" {
			if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil {
				return errors.New("mark expired attempt uncertain")
			}
		}
		if err := appendAudit(ctx, tx, "task_lease_expired", "execution_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
			return err
		}
	}
	return nil
}

func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
	rows, err := tx.QueryContext(ctx, `
		WITH invalid AS (
			SELECT t.id,
				CASE
					WHEN account.id IS NULL OR account.status <> 'active' OR account.authorization_status <> 'authorized' THEN 'policy_hold'
					WHEN account.version <> t.account_version THEN 'needs_confirmation'
					WHEN draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version THEN 'needs_confirmation'
					WHEN confirmation.id IS NULL OR confirmation.account_id <> t.account_id
						OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
						OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version THEN 'needs_confirmation'
					WHEN runtime.id IS NOT NULL AND runtime.binding_version IS DISTINCT FROM binding.version THEN 'needs_confirmation'
					ELSE 'policy_hold'
				END AS state,
				CASE
					WHEN account.id IS NULL THEN 'account_missing'
					WHEN account.authorization_status = 'revoked' THEN 'account_revoked'
					WHEN account.status <> 'active' THEN 'account_paused'
					WHEN account.version <> t.account_version THEN 'account_version_changed'
					WHEN draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version THEN 'draft_version_changed'
					WHEN confirmation.id IS NULL THEN 'confirmation_missing'
					WHEN confirmation.account_id <> t.account_id OR confirmation.account_version <> t.account_version
						OR confirmation.draft_id <> t.draft_id OR confirmation.draft_version <> t.draft_version
						OR confirmation.version <> t.confirmation_version THEN 'confirmation_version_changed'
					WHEN binding.id IS NULL THEN 'binding_missing'
					WHEN environment.alias IS NULL THEN 'environment_missing'
					WHEN network.id IS NULL THEN 'exit_missing'
					WHEN network.health_status <> 'healthy' THEN 'exit_unhealthy'
					WHEN binding.runtime_cleanup_pending THEN 'runtime_stop_pending'
					WHEN runtime.id IS NULL THEN 'runtime_missing'
					WHEN runtime.binding_version IS DISTINCT FROM binding.version THEN 'binding_version_changed'
					WHEN runtime.lease_until <= now() THEN 'runtime_lease_expired'
				END AS reason_code
			FROM operation_task t
			LEFT JOIN social_account account ON account.id = t.account_id
			LEFT JOIN content_draft draft ON draft.id = t.draft_id
			LEFT JOIN confirmation confirmation ON confirmation.id = t.confirmation_id
			LEFT JOIN environment_binding binding ON binding.account_id = t.account_id
			LEFT JOIN browser_env environment ON environment.alias = binding.browser_env_alias
			LEFT JOIN network_exit network ON network.id = binding.network_exit_id
			LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL
			WHERE t.state = 'queued' AND (
				account.id IS NULL OR account.status <> 'active' OR account.authorization_status <> 'authorized'
				OR account.version <> t.account_version
				OR draft.id IS NULL OR draft.account_id <> t.account_id OR draft.version <> t.draft_version
				OR confirmation.id IS NULL OR confirmation.account_id <> t.account_id
				OR confirmation.account_version <> t.account_version OR confirmation.draft_id <> t.draft_id
				OR confirmation.draft_version <> t.draft_version OR confirmation.version <> t.confirmation_version
				OR binding.id IS NULL OR environment.alias IS NULL OR network.id IS NULL OR network.health_status <> 'healthy'
				OR binding.runtime_cleanup_pending OR runtime.id IS NULL
				OR runtime.binding_version IS DISTINCT FROM binding.version OR runtime.lease_until <= now()
			)
			FOR UPDATE OF t SKIP LOCKED
		)
		UPDATE operation_task task SET state = invalid.state, updated_at = now()
		FROM invalid WHERE task.id = invalid.id
		RETURNING task.id, task.account_id, task.confirmation_id, task.confirmation_version, task.state, invalid.reason_code`)
	if err != nil {
		return errors.New("quarantine invalid tasks")
	}
	type invalidTask struct {
		taskID, accountID, confirmationID, state, reasonCode string
		confirmationVersion                                  int64
	}
	var invalid []invalidTask
	for rows.Next() {
		var taskID, accountID, state, reasonCode string
		var confirmationID sql.NullString
		var confirmationVersion sql.NullInt64
		if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state, &reasonCode); err != nil {
			rows.Close()
			return errors.New("read quarantined task")
		}
		invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, reasonCode, confirmationVersion.Int64})
	}
	if err := rows.Close(); err != nil {
		return errors.New("close quarantined task rows")
	}
	for _, task := range invalid {
		if err := appendAudit(ctx, tx, "task_quarantined", task.reasonCode, task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id,
			browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code,
			operation_id, action, outcome, old_image_version, new_image_version,
			details, created_at
		FROM audit_event ORDER BY id`)
	if err != nil {
		return nil, errors.New("read audit events")
	}
	defer rows.Close()
	var events []AuditEvent
	for rows.Next() {
		var event AuditEvent
		var accountID, confirmationID, attemptID, taskID, browserEnvAlias, networkExitID sql.NullString
		var runtimeInstanceID, actor, reasonCode, operationID, action, outcome, oldImage, newImage sql.NullString
		var confirmationVersion, bindingVersion sql.NullInt64
		if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID,
			&browserEnvAlias, &networkExitID, &runtimeInstanceID, &bindingVersion, &actor, &reasonCode,
			&operationID, &action, &outcome, &oldImage, &newImage,
			&event.Details, &event.CreatedAt); err != nil {
			return nil, errors.New("decode audit event")
		}
		event.AccountID, event.ConfirmationID, event.ConfirmationVersion = accountID.String, confirmationID.String, confirmationVersion.Int64
		event.AttemptID, event.TaskID = attemptID.String, taskID.String
		event.BrowserEnvAlias, event.NetworkExitID = browserEnvAlias.String, networkExitID.String
		event.RuntimeInstanceID, event.BindingVersion = runtimeInstanceID.String, bindingVersion.Int64
		event.Actor, event.ReasonCode = actor.String, reasonCode.String
		event.OperationID, event.Action, event.Outcome = operationID.String, action.String, outcome.String
		event.OldImageVersion, event.NewImageVersion = oldImage.String, newImage.String
		events = append(events, event)
	}
	return events, rows.Err()
}

func appendAudit(ctx context.Context, tx *sql.Tx, eventType, reasonCode, accountID, confirmationID string, confirmationVersion int64, attemptID, taskID string, details any) error {
	if details == nil {
		details = map[string]any{}
	}
	encoded, err := json.Marshal(details)
	if err != nil {
		return errors.New("encode audit details")
	}
	_, err = tx.ExecContext(ctx, `
		INSERT INTO audit_event
			(event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id,
			 browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code, details)
		SELECT $1, NULLIF($3, ''), NULLIF($4, ''), NULLIF($5, 0), NULLIF($6, ''), NULLIF($7, ''),
			binding.browser_env_alias, binding.network_exit_id, runtime.id, binding.version, 'local-user', $2, $8
		FROM (VALUES (1)) AS singleton(value)
		LEFT JOIN environment_binding binding ON binding.account_id = NULLIF($3, '')
		LEFT JOIN runtime_instance runtime ON runtime.binding_id = binding.id AND runtime.released_at IS NULL`,
		eventType, reasonCode, accountID, confirmationID, confirmationVersion, attemptID, taskID, encoded)
	if err != nil {
		return errors.New("append audit event")
	}
	return nil
}

func newID() string {
	var value [16]byte
	_, _ = rand.Read(value[:])
	return hex.EncodeToString(value[:])
}

func NewAccountID() string { return "account-" + newID()[:24] }

func commit(tx *sql.Tx) error {
	if err := tx.Commit(); err != nil {
		return errors.New("commit transaction")
	}
	return nil
}

func rowError(err error) error {
	if errors.Is(err, sql.ErrNoRows) {
		return ErrNotFound
	}
	return publicDatabaseError(err)
}

func publicDatabaseError(err error) error {
	if err == nil {
		return nil
	}
	var postgresError *pgconn.PgError
	if errors.As(err, &postgresError) && (postgresError.Code == "23505" || postgresError.Code == "23503" || postgresError.Code == "23514") {
		return ErrConflict
	}
	return errors.New("phase A persistence operation failed")
}