package transaction

import (
	"bytes"
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/ncruces/go-sqlite3/driver"
)

const schemaVersion = 2

const schemaV1 = `
CREATE TABLE transactions (
    id              TEXT PRIMARY KEY,
    idempotency_key TEXT NOT NULL UNIQUE,
    source          TEXT NOT NULL,
    service         TEXT NOT NULL,
    request_json    TEXT NOT NULL,
    state           TEXT NOT NULL CHECK (state IN (
        'CREATED', 'VALIDATING', 'PREPARED', 'STARTING', 'SWITCHING',
        'VERIFYING', 'DRAINING', 'COMMITTED', 'ROLLING_BACK',
        'ROLLED_BACK', 'FAILED'
    )),
    version         INTEGER NOT NULL,
    created_at      TEXT NOT NULL,
    updated_at      TEXT NOT NULL
) STRICT;

CREATE UNIQUE INDEX one_unfinished_transaction
    ON transactions ((1))
    WHERE state NOT IN ('COMMITTED', 'ROLLED_BACK', 'FAILED');

CREATE TABLE transaction_steps (
    transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
    step_key       TEXT NOT NULL,
    name           TEXT NOT NULL,
    status         TEXT NOT NULL CHECK (status IN ('INTENT_RECORDED', 'SUCCEEDED', 'FAILED')),
    intent_json    TEXT NOT NULL,
    result_json    TEXT,
    error_message  TEXT NOT NULL DEFAULT '',
    created_at     TEXT NOT NULL,
    updated_at     TEXT NOT NULL,
    PRIMARY KEY (transaction_id, step_key)
) STRICT;

CREATE TABLE transaction_events (
    sequence       INTEGER PRIMARY KEY AUTOINCREMENT,
    transaction_id TEXT NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
    step_key       TEXT NOT NULL DEFAULT '',
    kind           TEXT NOT NULL,
    from_state     TEXT NOT NULL DEFAULT '',
    to_state       TEXT NOT NULL DEFAULT '',
    message        TEXT NOT NULL DEFAULT '',
    created_at     TEXT NOT NULL
) STRICT;

CREATE INDEX transaction_events_by_transaction
    ON transaction_events (transaction_id, sequence);

PRAGMA user_version = 1;
`

const schemaV2 = `
CREATE TABLE backend_container_deployment (
    singleton_id   INTEGER PRIMARY KEY CHECK (singleton_id = 1),
    active_port    INTEGER NOT NULL CHECK (active_port IN (8080, 8081)),
    container_name TEXT NOT NULL,
    image_digest   TEXT NOT NULL,
    container_id   TEXT NOT NULL,
    transaction_id TEXT NOT NULL REFERENCES transactions(id),
    updated_at     TEXT NOT NULL
) STRICT;

PRAGMA user_version = 2;
`

// Store 是服务端 SQLite 事务记录。一个进程只应创建一个 Store。
type Store struct {
	db  *sql.DB
	now func() time.Time
}

// OpenStore 打开本地 SQLite，并强制校验持久化参数和 schema 版本。
func OpenStore(ctx context.Context, path string) (*Store, error) {
	if path == "" {
		return nil, errors.New("sqlite path is required")
	}
	absPath, err := filepath.Abs(path)
	if err != nil {
		return nil, fmt.Errorf("resolve sqlite path: %w", err)
	}
	if err := os.MkdirAll(filepath.Dir(absPath), 0o750); err != nil {
		return nil, fmt.Errorf("create sqlite directory: %w", err)
	}

	dsn := (&url.URL{
		Scheme:   "file",
		Path:     absPath,
		RawQuery: url.Values{"_txlock": {"immediate"}}.Encode(),
	}).String()
	db, err := driver.Open(dsn)
	if err != nil {
		return nil, fmt.Errorf("open sqlite driver: %w", err)
	}
	// 单连接是服务端事务串行化的一部分，也保证连接级 PRAGMA 始终生效。
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)

	closeOnError := func(cause error) (*Store, error) {
		_ = db.Close()
		return nil, cause
	}
	if err := db.PingContext(ctx); err != nil {
		return closeOnError(fmt.Errorf("ping sqlite: %w", err))
	}
	if err := configureSQLite(ctx, db); err != nil {
		return closeOnError(err)
	}
	if err := migrate(ctx, db); err != nil {
		return closeOnError(err)
	}
	if err := os.Chmod(absPath, 0o600); err != nil {
		return closeOnError(fmt.Errorf("set sqlite permissions: %w", err))
	}
	return &Store{db: db, now: time.Now}, nil
}

func configureSQLite(ctx context.Context, db *sql.DB) error {
	var journalMode string
	if err := db.QueryRowContext(ctx, "PRAGMA journal_mode = DELETE").Scan(&journalMode); err != nil {
		return fmt.Errorf("set sqlite journal mode: %w", err)
	}
	if journalMode != "delete" {
		return fmt.Errorf("sqlite journal mode mismatch: got %q, want %q", journalMode, "delete")
	}
	if _, err := db.ExecContext(ctx, "PRAGMA synchronous = EXTRA"); err != nil {
		return fmt.Errorf("set sqlite synchronous: %w", err)
	}
	var synchronous int
	if err := db.QueryRowContext(ctx, "PRAGMA synchronous").Scan(&synchronous); err != nil {
		return fmt.Errorf("read sqlite synchronous: %w", err)
	}
	if synchronous != 3 {
		return fmt.Errorf("sqlite synchronous mismatch: got %d, want 3 (EXTRA)", synchronous)
	}
	if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil {
		return fmt.Errorf("enable sqlite foreign keys: %w", err)
	}
	var foreignKeys int
	if err := db.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&foreignKeys); err != nil {
		return fmt.Errorf("read sqlite foreign_keys: %w", err)
	}
	if foreignKeys != 1 {
		return errors.New("sqlite foreign_keys is not enabled")
	}
	return nil
}

func migrate(ctx context.Context, db *sql.DB) error {
	var version int
	if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil {
		return fmt.Errorf("read sqlite schema version: %w", err)
	}
	if version > schemaVersion {
		return fmt.Errorf("sqlite schema version %d is newer than supported version %d", version, schemaVersion)
	}
	if version == schemaVersion {
		return nil
	}
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return fmt.Errorf("begin sqlite migration: %w", err)
	}
	defer tx.Rollback()
	for version < schemaVersion {
		nextVersion := version + 1
		var script string
		switch nextVersion {
		case 1:
			script = schemaV1
		case 2:
			script = schemaV2
		default:
			return fmt.Errorf("sqlite migration script is missing for version %d", nextVersion)
		}
		if _, err := tx.ExecContext(ctx, script); err != nil {
			return fmt.Errorf("apply sqlite schema version %d: %w", nextVersion, err)
		}
		version = nextVersion
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit sqlite migration: %w", err)
	}
	return nil
}

// Close 关闭服务端 SQLite。
func (s *Store) Close() error {
	return s.db.Close()
}

// CreateTransaction 原子处理幂等重试和单活动事务约束。created=false 表示返回已有幂等事务。
func (s *Store) CreateTransaction(ctx context.Context, request CreateRequest) (record Transaction, created bool, err error) {
	if err := validateCreateRequest(&request); err != nil {
		return Transaction{}, false, err
	}
	if request.ID == "" {
		request.ID = rand.Text()
	}
	now := s.now().UTC()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Transaction{}, false, fmt.Errorf("begin create transaction: %w", err)
	}
	defer tx.Rollback()

	existing, err := getTransactionByIdempotencyKey(ctx, tx, request.IdempotencyKey)
	if err == nil {
		if existing.State != StateFailed && existing.State != StateRolledBack {
			return existing, false, nil
		}
		archivedKey := existing.IdempotencyKey + ":terminal:" + existing.ID
		result, updateErr := tx.ExecContext(ctx, `
            UPDATE transactions
               SET idempotency_key = ?, version = version + 1, updated_at = ?
             WHERE id = ? AND version = ? AND idempotency_key = ?
               AND state IN (?, ?)`,
			archivedKey,
			formatTime(now),
			existing.ID,
			existing.Version,
			existing.IdempotencyKey,
			StateFailed,
			StateRolledBack,
		)
		if updateErr != nil {
			return Transaction{}, false, fmt.Errorf("archive terminal transaction idempotency key: %w", updateErr)
		}
		rows, rowsErr := result.RowsAffected()
		if rowsErr != nil {
			return Transaction{}, false, fmt.Errorf("read archived transaction rows: %w", rowsErr)
		}
		if rows != 1 {
			return Transaction{}, false, errors.New("terminal transaction changed concurrently")
		}
		if eventErr := insertEvent(ctx, tx, existing.ID, "", "TRANSACTION_RETRY_RELEASED", existing.State, existing.State, "terminal transaction idempotency key archived for manual retry", now); eventErr != nil {
			return Transaction{}, false, eventErr
		}
		err = ErrNotFound
	}
	if !errors.Is(err, ErrNotFound) {
		return Transaction{}, false, err
	}
	active, err := getActiveTransaction(ctx, tx)
	if err == nil {
		return Transaction{}, false, &ActiveTransactionError{TransactionID: active.ID}
	}
	if !errors.Is(err, ErrNotFound) {
		return Transaction{}, false, err
	}

	requestJSON := string(request.Request)
	_, err = tx.ExecContext(ctx, `
        INSERT INTO transactions (
            id, idempotency_key, source, service, request_json,
            state, version, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)`,
		request.ID,
		request.IdempotencyKey,
		request.Source,
		request.Service,
		requestJSON,
		StateCreated,
		formatTime(now),
		formatTime(now),
	)
	if err != nil {
		return Transaction{}, false, fmt.Errorf("insert transaction: %w", err)
	}
	if err := insertEvent(ctx, tx, request.ID, "", "TRANSACTION_CREATED", "", StateCreated, "", now); err != nil {
		return Transaction{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return Transaction{}, false, fmt.Errorf("commit create transaction: %w", err)
	}
	return Transaction{
		ID:             request.ID,
		IdempotencyKey: request.IdempotencyKey,
		Source:         request.Source,
		Service:        request.Service,
		Request:        cloneJSON(request.Request),
		State:          StateCreated,
		Version:        1,
		CreatedAt:      now,
		UpdatedAt:      now,
	}, true, nil
}

func validateCreateRequest(request *CreateRequest) error {
	if strings.TrimSpace(request.IdempotencyKey) == "" {
		return errors.New("idempotency key is required")
	}
	if strings.TrimSpace(request.Source) == "" {
		return errors.New("transaction source is required")
	}
	if strings.TrimSpace(request.Service) == "" {
		return errors.New("transaction service is required")
	}
	if len(request.Request) == 0 {
		request.Request = json.RawMessage(`{}`)
	}
	if !json.Valid(request.Request) {
		return errors.New("transaction request is not valid JSON")
	}
	return nil
}

// Transaction 返回指定事务的最新持久化快照。
func (s *Store) Transaction(ctx context.Context, id string) (Transaction, error) {
	return getTransactionByID(ctx, s.db, id)
}

// ActiveTransaction 返回当前唯一未结束事务。
func (s *Store) ActiveTransaction(ctx context.Context) (Transaction, error) {
	return getActiveTransaction(ctx, s.db)
}

// ListRecent returns committed, rolled-back, failed, and in-progress transactions
// in reverse creation order. Filters are exact values and never inferred.
func (s *Store) ListRecent(ctx context.Context, filter ListFilter) ([]Transaction, error) {
	if filter.Limit <= 0 || filter.Limit > 1000 {
		return nil, errors.New("transaction history limit must be between 1 and 1000")
	}
	query := `SELECT id, idempotency_key, source, service, request_json, state, version, created_at, updated_at FROM transactions WHERE 1=1`
	args := make([]any, 0, 3)
	if filter.Service != "" {
		query += " AND service = ?"
		args = append(args, filter.Service)
	}
	if filter.State != "" {
		query += " AND state = ?"
		args = append(args, string(filter.State))
	}
	query += " ORDER BY created_at DESC, id DESC LIMIT ?"
	args = append(args, filter.Limit)
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list transaction history: %w", err)
	}
	defer rows.Close()
	result := make([]Transaction, 0, filter.Limit)
	for rows.Next() {
		var item Transaction
		var state string
		var requestJSON string
		var createdAt, updatedAt string
		if err := rows.Scan(&item.ID, &item.IdempotencyKey, &item.Source, &item.Service, &requestJSON, &state, &item.Version, &createdAt, &updatedAt); err != nil {
			return nil, fmt.Errorf("scan transaction history: %w", err)
		}
		item.Request = json.RawMessage(requestJSON)
		item.State = State(state)
		item.CreatedAt, err = time.Parse(time.RFC3339Nano, createdAt)
		if err != nil {
			return nil, fmt.Errorf("parse transaction history creation time: %w", err)
		}
		item.UpdatedAt, err = time.Parse(time.RFC3339Nano, updatedAt)
		if err != nil {
			return nil, fmt.Errorf("parse transaction history update time: %w", err)
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate transaction history: %w", err)
	}
	return result, nil
}

// Transition 校验并原子提交状态变化及其恢复事件。
func (s *Store) Transition(ctx context.Context, id string, next State, message string) (Transaction, error) {
	if !next.Valid() {
		return Transaction{}, fmt.Errorf("unknown transaction state: %q", next)
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Transaction{}, fmt.Errorf("begin state transition: %w", err)
	}
	defer tx.Rollback()
	record, err := getTransactionByID(ctx, tx, id)
	if err != nil {
		return Transaction{}, err
	}
	if record.State == next {
		return record, nil
	}
	if !record.State.CanTransitionTo(next) {
		return Transaction{}, &TransitionError{From: record.State, To: next}
	}
	now := s.now().UTC()
	result, err := tx.ExecContext(ctx, `
        UPDATE transactions
           SET state = ?, version = version + 1, updated_at = ?
         WHERE id = ? AND version = ?`,
		next, formatTime(now), id, record.Version,
	)
	if err != nil {
		return Transaction{}, fmt.Errorf("update transaction state: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return Transaction{}, fmt.Errorf("read updated transaction rows: %w", err)
	}
	if rows != 1 {
		return Transaction{}, errors.New("transaction changed concurrently")
	}
	if err := insertEvent(ctx, tx, id, "", "TRANSACTION_STATE_CHANGED", record.State, next, message, now); err != nil {
		return Transaction{}, err
	}
	if err := tx.Commit(); err != nil {
		return Transaction{}, fmt.Errorf("commit state transition: %w", err)
	}
	record.State = next
	record.Version++
	record.UpdatedAt = now
	return record, nil
}

// RecordStepIntent 先于外部副作用持久化步骤意图。created=false 表示相同意图已经存在。
func (s *Store) RecordStepIntent(ctx context.Context, transactionID string, intent StepIntent) (record Step, created bool, err error) {
	if strings.TrimSpace(intent.Key) == "" {
		return Step{}, false, errors.New("step key is required")
	}
	if strings.TrimSpace(intent.Name) == "" {
		return Step{}, false, errors.New("step name is required")
	}
	if len(intent.Intent) == 0 {
		intent.Intent = json.RawMessage(`{}`)
	}
	if !json.Valid(intent.Intent) {
		return Step{}, false, errors.New("step intent is not valid JSON")
	}

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, false, fmt.Errorf("begin record step intent: %w", err)
	}
	defer tx.Rollback()
	transactionRecord, err := getTransactionByID(ctx, tx, transactionID)
	if err != nil {
		return Step{}, false, err
	}
	if transactionRecord.State.Terminal() {
		return Step{}, false, fmt.Errorf("cannot record a step for terminal transaction %s", transactionID)
	}
	existing, err := getStep(ctx, tx, transactionID, intent.Key)
	if err == nil {
		if existing.Name != intent.Name || !bytes.Equal(existing.Intent, intent.Intent) {
			return Step{}, false, ErrStepConflict
		}
		return existing, false, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return Step{}, false, err
	}

	now := s.now().UTC()
	_, err = tx.ExecContext(ctx, `
        INSERT INTO transaction_steps (
            transaction_id, step_key, name, status, intent_json,
            result_json, error_message, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, NULL, '', ?, ?)`,
		transactionID,
		intent.Key,
		intent.Name,
		StepIntentRecorded,
		string(intent.Intent),
		formatTime(now),
		formatTime(now),
	)
	if err != nil {
		return Step{}, false, fmt.Errorf("insert step intent: %w", err)
	}
	if err := insertEvent(ctx, tx, transactionID, intent.Key, "STEP_INTENT_RECORDED", "", "", intent.Name, now); err != nil {
		return Step{}, false, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, false, fmt.Errorf("commit step intent: %w", err)
	}
	return Step{
		TransactionID: transactionID,
		Key:           intent.Key,
		Name:          intent.Name,
		Status:        StepIntentRecorded,
		Intent:        cloneJSON(intent.Intent),
		CreatedAt:     now,
		UpdatedAt:     now,
	}, true, nil
}

// CompleteStep 原子记录外部状态核对后的最终结果。
func (s *Store) CompleteStep(ctx context.Context, transactionID, stepKey string, status StepStatus, result json.RawMessage, errorMessage string) (Step, error) {
	if status != StepSucceeded && status != StepFailed {
		return Step{}, fmt.Errorf("invalid final step status: %q", status)
	}
	if len(result) > 0 && !json.Valid(result) {
		return Step{}, errors.New("step result is not valid JSON")
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, fmt.Errorf("begin complete step: %w", err)
	}
	defer tx.Rollback()
	record, err := getStep(ctx, tx, transactionID, stepKey)
	if err != nil {
		return Step{}, err
	}
	if record.Status == status {
		if !bytes.Equal(record.Result, result) || record.Error != errorMessage {
			return Step{}, ErrStepConflict
		}
		return record, nil
	}
	if record.Status != StepIntentRecorded {
		return Step{}, ErrStepNotPending
	}
	now := s.now().UTC()
	var resultValue any
	if len(result) > 0 {
		resultValue = string(result)
	}
	updateResult, err := tx.ExecContext(ctx, `
        UPDATE transaction_steps
           SET status = ?, result_json = ?, error_message = ?, updated_at = ?
         WHERE transaction_id = ? AND step_key = ? AND status = ?`,
		status,
		resultValue,
		errorMessage,
		formatTime(now),
		transactionID,
		stepKey,
		StepIntentRecorded,
	)
	if err != nil {
		return Step{}, fmt.Errorf("update step result: %w", err)
	}
	updatedRows, err := updateResult.RowsAffected()
	if err != nil {
		return Step{}, fmt.Errorf("read updated step rows: %w", err)
	}
	if updatedRows != 1 {
		return Step{}, errors.New("step changed concurrently")
	}
	eventKind := "STEP_SUCCEEDED"
	if status == StepFailed {
		eventKind = "STEP_FAILED"
	}
	if err := insertEvent(ctx, tx, transactionID, stepKey, eventKind, "", "", errorMessage, now); err != nil {
		return Step{}, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, fmt.Errorf("commit step result: %w", err)
	}
	record.Status = status
	record.Result = cloneJSON(result)
	record.Error = errorMessage
	record.UpdatedAt = now
	return record, nil
}

// ReopenFailedStep makes one manually resumed external step pending again.
// The coordinator calls this only after Inspect has established an exact
// APPLIED or NOT_APPLIED state; UNKNOWN never reopens a failed step.
func (s *Store) ReopenFailedStep(ctx context.Context, transactionID, stepKey, message string) (Step, error) {
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return Step{}, fmt.Errorf("begin reopen failed step: %w", err)
	}
	defer tx.Rollback()
	record, err := getStep(ctx, tx, transactionID, stepKey)
	if err != nil {
		return Step{}, err
	}
	if record.Status == StepIntentRecorded {
		return record, nil
	}
	if record.Status != StepFailed {
		return Step{}, ErrStepNotPending
	}
	now := s.now().UTC()
	result, err := tx.ExecContext(ctx, `
        UPDATE transaction_steps
           SET status = ?, result_json = NULL, error_message = '', updated_at = ?
         WHERE transaction_id = ? AND step_key = ? AND status = ?`,
		StepIntentRecorded,
		formatTime(now),
		transactionID,
		stepKey,
		StepFailed,
	)
	if err != nil {
		return Step{}, fmt.Errorf("reopen failed step: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return Step{}, fmt.Errorf("read reopened step rows: %w", err)
	}
	if rows != 1 {
		return Step{}, errors.New("failed step changed concurrently")
	}
	if err := insertEvent(ctx, tx, transactionID, stepKey, "STEP_REOPENED", "", "", message, now); err != nil {
		return Step{}, err
	}
	if err := tx.Commit(); err != nil {
		return Step{}, fmt.Errorf("commit reopened step: %w", err)
	}
	record.Status = StepIntentRecorded
	record.Result = nil
	record.Error = ""
	record.UpdatedAt = now
	return record, nil
}

// PendingSteps 返回重启后必须先核对实际外部状态的步骤。
func (s *Store) PendingSteps(ctx context.Context, transactionID string) ([]Step, error) {
	rows, err := s.db.QueryContext(ctx, `
        SELECT transaction_id, step_key, name, status, intent_json,
               result_json, error_message, created_at, updated_at
          FROM transaction_steps
         WHERE transaction_id = ? AND status = ?
         ORDER BY created_at, step_key`, transactionID, StepIntentRecorded)
	if err != nil {
		return nil, fmt.Errorf("query pending steps: %w", err)
	}
	defer rows.Close()
	var records []Step
	for rows.Next() {
		record, err := scanStep(rows)
		if err != nil {
			return nil, err
		}
		records = append(records, record)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate pending steps: %w", err)
	}
	return records, nil
}

// EventsAfter 返回指定顺序位置之后的事务事件。
func (s *Store) EventsAfter(ctx context.Context, transactionID string, afterSequence int64, limit int) ([]Event, error) {
	if afterSequence < 0 {
		return nil, errors.New("event sequence must not be negative")
	}
	if limit < 1 || limit > 1000 {
		return nil, errors.New("event limit must be between 1 and 1000")
	}
	rows, err := s.db.QueryContext(ctx, `
        SELECT sequence, transaction_id, step_key, kind,
               from_state, to_state, message, created_at
          FROM transaction_events
         WHERE transaction_id = ? AND sequence > ?
         ORDER BY sequence
         LIMIT ?`, transactionID, afterSequence, limit)
	if err != nil {
		return nil, fmt.Errorf("query transaction events: %w", err)
	}
	defer rows.Close()
	var events []Event
	for rows.Next() {
		var event Event
		var fromState, toState, createdAt string
		if err := rows.Scan(
			&event.Sequence,
			&event.TransactionID,
			&event.StepKey,
			&event.Kind,
			&fromState,
			&toState,
			&event.Message,
			&createdAt,
		); err != nil {
			return nil, fmt.Errorf("scan transaction event: %w", err)
		}
		event.FromState = State(fromState)
		event.ToState = State(toState)
		event.CreatedAt, err = parseTime(createdAt)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate transaction events: %w", err)
	}
	return events, nil
}

type queryRower interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

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

func getTransactionByID(ctx context.Context, query queryRower, id string) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE id = ?`, id))
}

func getTransactionByIdempotencyKey(ctx context.Context, query queryRower, key string) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE idempotency_key = ?`, key))
}

func getActiveTransaction(ctx context.Context, query queryRower) (Transaction, error) {
	return scanTransaction(query.QueryRowContext(ctx, `
        SELECT id, idempotency_key, source, service, request_json,
               state, version, created_at, updated_at
          FROM transactions
         WHERE state NOT IN (?, ?, ?)
         LIMIT 1`, StateCommitted, StateRolledBack, StateFailed))
}

func scanTransaction(row rowScanner) (Transaction, error) {
	var record Transaction
	var requestJSON, state, createdAt, updatedAt string
	if err := row.Scan(
		&record.ID,
		&record.IdempotencyKey,
		&record.Source,
		&record.Service,
		&requestJSON,
		&state,
		&record.Version,
		&createdAt,
		&updatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Transaction{}, ErrNotFound
		}
		return Transaction{}, fmt.Errorf("scan transaction: %w", err)
	}
	record.Request = json.RawMessage(requestJSON)
	record.State = State(state)
	var err error
	record.CreatedAt, err = parseTime(createdAt)
	if err != nil {
		return Transaction{}, err
	}
	record.UpdatedAt, err = parseTime(updatedAt)
	if err != nil {
		return Transaction{}, err
	}
	return record, nil
}

func getStep(ctx context.Context, query queryRower, transactionID, stepKey string) (Step, error) {
	return scanStep(query.QueryRowContext(ctx, `
        SELECT transaction_id, step_key, name, status, intent_json,
               result_json, error_message, created_at, updated_at
          FROM transaction_steps
         WHERE transaction_id = ? AND step_key = ?`, transactionID, stepKey))
}

func scanStep(row rowScanner) (Step, error) {
	var record Step
	var status, intentJSON, createdAt, updatedAt string
	var resultJSON sql.NullString
	if err := row.Scan(
		&record.TransactionID,
		&record.Key,
		&record.Name,
		&status,
		&intentJSON,
		&resultJSON,
		&record.Error,
		&createdAt,
		&updatedAt,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Step{}, ErrNotFound
		}
		return Step{}, fmt.Errorf("scan transaction step: %w", err)
	}
	record.Status = StepStatus(status)
	record.Intent = json.RawMessage(intentJSON)
	if resultJSON.Valid {
		record.Result = json.RawMessage(resultJSON.String)
	}
	var err error
	record.CreatedAt, err = parseTime(createdAt)
	if err != nil {
		return Step{}, err
	}
	record.UpdatedAt, err = parseTime(updatedAt)
	if err != nil {
		return Step{}, err
	}
	return record, nil
}

func insertEvent(
	ctx context.Context,
	tx *sql.Tx,
	transactionID string,
	stepKey string,
	kind string,
	fromState State,
	toState State,
	message string,
	createdAt time.Time,
) error {
	_, err := tx.ExecContext(ctx, `
        INSERT INTO transaction_events (
            transaction_id, step_key, kind, from_state, to_state, message, created_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		transactionID,
		stepKey,
		kind,
		fromState,
		toState,
		message,
		formatTime(createdAt),
	)
	if err != nil {
		return fmt.Errorf("insert transaction event: %w", err)
	}
	return nil
}

func formatTime(value time.Time) string {
	return value.UTC().Format(time.RFC3339Nano)
}

func parseTime(value string) (time.Time, error) {
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil {
		return time.Time{}, fmt.Errorf("parse persisted time %q: %w", value, err)
	}
	return parsed, nil
}

func cloneJSON(value json.RawMessage) json.RawMessage {
	if len(value) == 0 {
		return nil
	}
	return bytes.Clone(value)
}