package persistence

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"regexp"
	"time"
)

const (
	runtimeRecordCommandTable          = "_System_RuntimeRecordCommand"
	runtimeRecordCommandOperationTable = "_System_RuntimeRecordCommandOperation"

	RuntimeRecordCommandAccepted  = "ACCEPTED"
	RuntimeRecordCommandCompleted = "COMPLETED"
	RuntimeRecordOperationPending = "PENDING"
	RuntimeRecordOperationDone    = "COMPLETED"
)

var (
	ErrRuntimeRecordCommandConflict              = errors.New("runtime record command identity is already bound to different content")
	ErrRuntimeRecordCommandServingBindingChanged = errors.New("runtime record command is bound to a different serving revision")
	ErrRuntimeRecordCommandMissing               = errors.New("runtime record command is missing")
	hexDigestPattern                             = regexp.MustCompile(`^[0-9a-f]{64}$`)
)

// RuntimeRecordCommandRepository is the database authority for record-command
// idempotency and terminal outcomes. Payloads are opaque canonical JSON owned
// by the records semantic adapter; persistence owns only transactionality,
// ordering, and compare-before-replay invariants.
type RuntimeRecordCommandRepository struct {
	db *sql.DB
}

func NewRuntimeRecordCommandRepository(db *sql.DB) *RuntimeRecordCommandRepository {
	return &RuntimeRecordCommandRepository{db: db}
}

type RuntimeRecordCommandAcceptance struct {
	RowID            string
	ScopeDigest      string
	CommandID        string
	RequestDigest    string
	BatchMode        string
	ReceiptsPayload  []byte
	OutcomeRetention time.Duration
	Operations       []RuntimeRecordCommandOperationAcceptance
}

type RuntimeRecordCommandOperationAcceptance struct {
	RowID         string
	Index         int
	Kind          string
	ObjectID      string
	BoundRecordID string
}

type RuntimeRecordCommandRow struct {
	RowID            string
	ScopeDigest      string
	CommandID        string
	RequestDigest    string
	BatchMode        string
	OperationCount   int
	State            string
	ReceiptsPayload  []byte
	ResultPayload    []byte
	OutcomeExpiresAt time.Time
	// OutcomeRetentionMicros is the immutable capability-owned duration used
	// for both the initial execution/reconciliation window and every terminal
	// proof window. Persisting the duration avoids reconstructing policy from
	// timestamps or duplicating it in worker configuration.
	OutcomeRetentionMicros int64
}

type RuntimeRecordCommandOperationRow struct {
	RowID         string
	CommandRowID  string
	Index         int
	Kind          string
	ObjectID      string
	BoundRecordID string
	State         string
	ResultPayload []byte
}

// Accept binds one scoped command identity to exactly one semantic digest and
// creates every position-stable operation slot. The caller must supply an
// active pessimistic transaction so acceptance and slots cannot split.
func (r *RuntimeRecordCommandRepository) Accept(ctx context.Context, input RuntimeRecordCommandAcceptance) (RuntimeRecordCommandRow, bool, error) {
	if r == nil || r.db == nil {
		return RuntimeRecordCommandRow{}, false, errors.New("runtime record command repository is unavailable")
	}
	if !hexDigestPattern.MatchString(input.ScopeDigest) || input.RowID == "" || input.CommandID == "" || input.RequestDigest == "" ||
		input.BatchMode == "" || len(input.ReceiptsPayload) == 0 || input.OutcomeRetention.Microseconds() <= 0 || len(input.Operations) == 0 {
		return RuntimeRecordCommandRow{}, false, errors.New("runtime record command acceptance is incomplete")
	}
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return RuntimeRecordCommandRow{}, false, err
	}
	// Only a terminal outcome may be recycled. An expired ACCEPTED row can own
	// committed independent positions, so deleting it would erase both outcome
	// and duplicate-prevention authority. The fenced command-lifecycle worker
	// closes such rows first and retains the terminal proof for a fresh window.
	if _, err := exec.ExecContext(ctx, `DELETE FROM _System_RuntimeRecordCommand
WHERE ScopeDigest = ? AND CommandId = ? AND State = 'COMPLETED'
  AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)`, input.ScopeDigest, input.CommandID); err != nil {
		return RuntimeRecordCommandRow{}, false, fmt.Errorf("recycle expired runtime record command: %w", err)
	}
	result, err := exec.ExecContext(ctx, `INSERT IGNORE INTO _System_RuntimeRecordCommand
(Id, ScopeDigest, CommandId, RequestDigest, BatchMode, OperationCount, State, Receipts, OutcomeExpiresAt, OutcomeRetentionMicros)
VALUES (?, ?, ?, ?, ?, ?, 'ACCEPTED', ?, TIMESTAMPADD(MICROSECOND, ?, UTC_TIMESTAMP(6)), ?)`, input.RowID, input.ScopeDigest, input.CommandID, input.RequestDigest,
		input.BatchMode, len(input.Operations), input.ReceiptsPayload, input.OutcomeRetention.Microseconds(), input.OutcomeRetention.Microseconds())
	if err != nil {
		return RuntimeRecordCommandRow{}, false, fmt.Errorf("accept runtime record command: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return RuntimeRecordCommandRow{}, false, fmt.Errorf("accept runtime record command rows affected: %w", err)
	}
	row, err := r.loadCommand(ctx, exec, input.ScopeDigest, input.CommandID, true)
	if err != nil {
		return RuntimeRecordCommandRow{}, false, err
	}
	if row.RequestDigest != input.RequestDigest || row.BatchMode != input.BatchMode || row.OperationCount != len(input.Operations) {
		return row, true, ErrRuntimeRecordCommandConflict
	}
	// A partially completed independent command must never resume against a
	// different Catalog/Security receipt and then claim one revision for mixed
	// execution. The immutable receipt is part of the durable execution
	// binding, while RequestDigest remains the caller-authored command identity.
	sameReceipts, err := equivalentJSONDocuments(row.ReceiptsPayload, input.ReceiptsPayload)
	if err != nil {
		return row, true, fmt.Errorf("compare runtime record command receipts: %w", err)
	}
	if !sameReceipts {
		return row, true, ErrRuntimeRecordCommandServingBindingChanged
	}
	created := affected == 1
	if !created {
		return row, true, nil
	}
	for _, operation := range input.Operations {
		if operation.RowID == "" || operation.Index < 0 || operation.Kind == "" {
			return RuntimeRecordCommandRow{}, false, errors.New("runtime record command operation acceptance is incomplete")
		}
		var boundRecordID any
		if operation.BoundRecordID != "" {
			boundRecordID = operation.BoundRecordID
		}
		var objectID any
		if operation.ObjectID != "" {
			objectID = operation.ObjectID
		}
		if _, err := exec.ExecContext(ctx, `INSERT INTO _System_RuntimeRecordCommandOperation
(Id, CommandRowId, OperationIndex, OperationKind, OperationObjectId, BoundRecordId, State)
VALUES (?, ?, ?, ?, ?, ?, 'PENDING')`, operation.RowID, row.RowID, operation.Index, operation.Kind, objectID, boundRecordID); err != nil {
			return RuntimeRecordCommandRow{}, false, fmt.Errorf("accept runtime record command operation %d: %w", operation.Index, err)
		}
	}
	return row, false, nil
}

// equivalentJSONDocuments compares the semantic value of two opaque JSON
// documents. Receipts are stored in a native JSON column, so MySQL/TiDB may
// legally reorder object keys and add insignificant whitespace before a read.
// Byte equality would therefore reject the value immediately after inserting
// it. UseNumber keeps large revision/generation integers exact rather than
// rounding them through float64.
func equivalentJSONDocuments(left, right []byte) (bool, error) {
	leftCanonical, err := canonicalJSONDocument(left)
	if err != nil {
		return false, fmt.Errorf("decode stored JSON: %w", err)
	}
	rightCanonical, err := canonicalJSONDocument(right)
	if err != nil {
		return false, fmt.Errorf("decode candidate JSON: %w", err)
	}
	return bytes.Equal(leftCanonical, rightCanonical), nil
}

func canonicalJSONDocument(payload []byte) ([]byte, error) {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, err
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		if err == nil {
			return nil, errors.New("multiple JSON values")
		}
		return nil, err
	}
	return json.Marshal(value)
}

func (r *RuntimeRecordCommandRepository) Load(ctx context.Context, scopeDigest, commandID string) (RuntimeRecordCommandRow, error) {
	if r == nil || r.db == nil {
		return RuntimeRecordCommandRow{}, errors.New("runtime record command repository is unavailable")
	}
	return r.loadCommand(ctx, r.executor(ctx), scopeDigest, commandID, false)
}

// LoadUnexpired is the reconciliation read path. Expiration is evaluated by
// the same database clock that owns atomic command recycling and cleanup.
func (r *RuntimeRecordCommandRepository) LoadUnexpired(ctx context.Context, scopeDigest, commandID string) (RuntimeRecordCommandRow, error) {
	if r == nil || r.db == nil {
		return RuntimeRecordCommandRow{}, errors.New("runtime record command repository is unavailable")
	}
	return r.loadCommandWithPredicate(ctx, r.executor(ctx), scopeDigest, commandID, false, " AND OutcomeExpiresAt > UTC_TIMESTAMP(6)")
}

func (r *RuntimeRecordCommandRepository) LockCommand(ctx context.Context, commandRowID string) (RuntimeRecordCommandRow, error) {
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return RuntimeRecordCommandRow{}, err
	}
	var row RuntimeRecordCommandRow
	err = exec.QueryRowContext(ctx, `SELECT Id, ScopeDigest, CommandId, RequestDigest, BatchMode, OperationCount, State,
	       Receipts, COALESCE(ResultPayload, ''), OutcomeExpiresAt, OutcomeRetentionMicros
FROM _System_RuntimeRecordCommand WHERE Id = ? FOR UPDATE`, commandRowID).Scan(
		&row.RowID, &row.ScopeDigest, &row.CommandID, &row.RequestDigest, &row.BatchMode,
		&row.OperationCount, &row.State, &row.ReceiptsPayload, &row.ResultPayload, &row.OutcomeExpiresAt, &row.OutcomeRetentionMicros,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeRecordCommandRow{}, ErrRuntimeRecordCommandMissing
	}
	if err != nil {
		return RuntimeRecordCommandRow{}, fmt.Errorf("lock runtime record command: %w", err)
	}
	return row, nil
}

func (r *RuntimeRecordCommandRepository) LockOperation(ctx context.Context, commandRowID string, index int) (RuntimeRecordCommandOperationRow, error) {
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return RuntimeRecordCommandOperationRow{}, err
	}
	return scanRuntimeRecordCommandOperation(exec.QueryRowContext(ctx, `SELECT Id, CommandRowId, OperationIndex, OperationKind,
       COALESCE(OperationObjectId, ''), COALESCE(BoundRecordId, ''), State, COALESCE(ResultPayload, '')
FROM _System_RuntimeRecordCommandOperation
WHERE CommandRowId = ? AND OperationIndex = ? FOR UPDATE`, commandRowID, index))
}

func (r *RuntimeRecordCommandRepository) CompleteOperation(ctx context.Context, commandRowID string, index int, payload []byte) error {
	if len(payload) == 0 {
		return errors.New("runtime record command operation result is empty")
	}
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return err
	}
	result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeRecordCommandOperation
SET State = 'COMPLETED', ResultPayload = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE CommandRowId = ? AND OperationIndex = ? AND State = 'PENDING'`, payload, commandRowID, index)
	if err != nil {
		return fmt.Errorf("complete runtime record command operation: %w", err)
	}
	return requireOneAffected(result, "complete runtime record command operation")
}

// LockOperations returns every slot in stable position order while retaining
// row locks until the caller's transaction commits or rolls back.
func (r *RuntimeRecordCommandRepository) LockOperations(ctx context.Context, commandRowID string) ([]RuntimeRecordCommandOperationRow, error) {
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return nil, err
	}
	return loadRuntimeRecordCommandOperations(ctx, exec, commandRowID, true)
}

// LoadOperations reads the durable position/kind binding used to validate a
// reconciliation response. Unlike LockOperations it is safe outside a
// transaction and does not claim mutation authority.
func (r *RuntimeRecordCommandRepository) LoadOperations(ctx context.Context, commandRowID string) ([]RuntimeRecordCommandOperationRow, error) {
	if r == nil || r.db == nil || commandRowID == "" {
		return nil, ErrRuntimeRecordCommandMissing
	}
	return loadRuntimeRecordCommandOperations(ctx, r.executor(ctx), commandRowID, false)
}

func loadRuntimeRecordCommandOperations(ctx context.Context, exec Executor, commandRowID string, forUpdate bool) ([]RuntimeRecordCommandOperationRow, error) {
	query := `SELECT Id, CommandRowId, OperationIndex, OperationKind,
       COALESCE(OperationObjectId, ''), COALESCE(BoundRecordId, ''), State, COALESCE(ResultPayload, '')
FROM _System_RuntimeRecordCommandOperation
WHERE CommandRowId = ? ORDER BY OperationIndex`
	if forUpdate {
		query += " FOR UPDATE"
	}
	rows, err := exec.QueryContext(ctx, query, commandRowID)
	if err != nil {
		return nil, fmt.Errorf("lock runtime record command operations: %w", err)
	}
	defer rows.Close()
	result := make([]RuntimeRecordCommandOperationRow, 0)
	for rows.Next() {
		row, scanErr := scanRuntimeRecordCommandOperation(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		result = append(result, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate runtime record command operations: %w", err)
	}
	return result, nil
}

func (r *RuntimeRecordCommandRepository) CompleteCommand(ctx context.Context, commandRowID string, resultPayload []byte, outcomeExpiresAt time.Time) error {
	if len(resultPayload) == 0 || outcomeExpiresAt.IsZero() {
		return errors.New("runtime record command result is empty")
	}
	exec, err := r.transactionExecutor(ctx)
	if err != nil {
		return err
	}
	result, err := exec.ExecContext(ctx, `UPDATE _System_RuntimeRecordCommand
SET State = 'COMPLETED', ResultPayload = ?, OutcomeExpiresAt = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND State = 'ACCEPTED'`, resultPayload, outcomeExpiresAt.UTC(), commandRowID)
	if err != nil {
		return fmt.Errorf("complete runtime record command: %w", err)
	}
	return requireOneAffected(result, "complete runtime record command")
}

// ListExpiredAccepted returns bounded candidate identities for lifecycle
// closure. The lifecycle owner locks and rechecks each command in its fenced
// completion transaction; this read is discovery, never transition authority.
func (r *RuntimeRecordCommandRepository) ListExpiredAccepted(ctx context.Context, limit int) ([]string, error) {
	if r == nil || r.db == nil || limit <= 0 || limit > 10_000 {
		return nil, errors.New("runtime record command lifecycle discovery is invalid")
	}
	rows, err := r.db.QueryContext(ctx, `SELECT Id FROM _System_RuntimeRecordCommand
WHERE State = 'ACCEPTED' AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)
ORDER BY OutcomeExpiresAt, Id LIMIT ?`, limit)
	if err != nil {
		return nil, fmt.Errorf("list expired accepted runtime record commands: %w", err)
	}
	defer rows.Close()
	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("scan expired accepted runtime record command: %w", err)
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate expired accepted runtime record commands: %w", err)
	}
	return ids, nil
}

func (r *RuntimeRecordCommandRepository) executor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

func (r *RuntimeRecordCommandRepository) transactionExecutor(ctx context.Context) (Executor, error) {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx, nil
	}
	return nil, errors.New("runtime record command operation requires an active transaction")
}

func (r *RuntimeRecordCommandRepository) loadCommand(ctx context.Context, exec Executor, scopeDigest, commandID string, forUpdate bool) (RuntimeRecordCommandRow, error) {
	return r.loadCommandWithPredicate(ctx, exec, scopeDigest, commandID, forUpdate, "")
}

func (r *RuntimeRecordCommandRepository) loadCommandWithPredicate(ctx context.Context, exec Executor, scopeDigest, commandID string, forUpdate bool, predicate string) (RuntimeRecordCommandRow, error) {
	if !hexDigestPattern.MatchString(scopeDigest) || commandID == "" {
		return RuntimeRecordCommandRow{}, ErrRuntimeRecordCommandMissing
	}
	query := `SELECT Id, ScopeDigest, CommandId, RequestDigest, BatchMode, OperationCount, State,
	       Receipts, COALESCE(ResultPayload, ''), OutcomeExpiresAt, OutcomeRetentionMicros
FROM _System_RuntimeRecordCommand WHERE ScopeDigest = ? AND CommandId = ?` + predicate
	if forUpdate {
		query += " FOR UPDATE"
	}
	var row RuntimeRecordCommandRow
	err := exec.QueryRowContext(ctx, query, scopeDigest, commandID).Scan(
		&row.RowID, &row.ScopeDigest, &row.CommandID, &row.RequestDigest, &row.BatchMode,
		&row.OperationCount, &row.State, &row.ReceiptsPayload, &row.ResultPayload, &row.OutcomeExpiresAt, &row.OutcomeRetentionMicros,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeRecordCommandRow{}, ErrRuntimeRecordCommandMissing
	}
	if err != nil {
		return RuntimeRecordCommandRow{}, fmt.Errorf("load runtime record command: %w", err)
	}
	return row, nil
}

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

func scanRuntimeRecordCommandOperation(scanner runtimeRecordCommandRowScanner) (RuntimeRecordCommandOperationRow, error) {
	var row RuntimeRecordCommandOperationRow
	if err := scanner.Scan(&row.RowID, &row.CommandRowID, &row.Index, &row.Kind, &row.ObjectID, &row.BoundRecordID, &row.State, &row.ResultPayload); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return RuntimeRecordCommandOperationRow{}, ErrRuntimeRecordCommandMissing
		}
		return RuntimeRecordCommandOperationRow{}, fmt.Errorf("scan runtime record command operation: %w", err)
	}
	return row, nil
}

func requireOneAffected(result sql.Result, operation string) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("%s rows affected: %w", operation, err)
	}
	if affected != 1 {
		return fmt.Errorf("%s lost authority", operation)
	}
	return nil
}