package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"
)

const RuntimeRecordCommandLifecycleWorkFamily = "runtime.record-command-lifecycle.v1"

var runtimeRecordCommandLifecycleRequiredColumns = map[string]string{
	"State": "varchar", "OutcomeExpiresAt": "datetime", "OutcomeRetentionMicros": "bigint",
}

type RuntimeRecordCommandLifecycleStore struct {
	db        *sql.DB
	ownership *RuntimeWorkOwnershipRepository
}

func NewRuntimeRecordCommandLifecycleStore(db *sql.DB, expected RuntimeWorkOwnership) (*RuntimeRecordCommandLifecycleStore, error) {
	ownership, err := NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	if expected.WorkFamily != RuntimeRecordCommandLifecycleWorkFamily {
		return nil, fmt.Errorf("runtime record command lifecycle: wrong work family %q", expected.WorkFamily)
	}
	return &RuntimeRecordCommandLifecycleStore{db: db, ownership: ownership}, nil
}

func (s *RuntimeRecordCommandLifecycleStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("runtime record command lifecycle: store is unavailable")
	}
	if err := s.ownership.CheckReady(ctx); err != nil {
		return err
	}
	return VerifyRuntimeRecordCommandSchema(ctx, s.db)
}

// VerifyRuntimeRecordCommandSchema proves the physical command/outcome
// authority required by both command admission and its independently fenced
// lifecycle worker. It deliberately does not inspect worker ownership: a
// producer uses it as schema readiness, while the worker's CheckReady adds its
// exact deployment fence. Keeping this one check shared prevents an API from
// accepting a durable command that its recovery owner cannot later inspect or
// retire safely.
func VerifyRuntimeRecordCommandSchema(ctx context.Context, db *sql.DB) error {
	if ctx == nil || db == nil {
		return errors.New("runtime record command lifecycle: context and database are required")
	}
	for _, table := range []string{runtimeRecordCommandTable, runtimeRecordCommandOperationTable} {
		var count int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, table).Scan(&count); err != nil {
			return fmt.Errorf("runtime record command lifecycle: inspect %s: %w", table, err)
		}
		if count != 1 {
			return fmt.Errorf("runtime record command lifecycle: required table %s is missing", table)
		}
	}
	if err := VerifyRequiredTableColumns(ctx, db, runtimeRecordCommandTable, runtimeRecordCommandLifecycleRequiredColumns); err != nil {
		return fmt.Errorf("runtime record command lifecycle: %w", err)
	}
	var deleteRule string
	if err := db.QueryRowContext(ctx, `SELECT DELETE_RULE
FROM INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS
WHERE CONSTRAINT_SCHEMA = DATABASE() AND TABLE_NAME = ?
  AND REFERENCED_TABLE_NAME = ?`, runtimeRecordCommandOperationTable, runtimeRecordCommandTable).Scan(&deleteRule); err != nil {
		return fmt.Errorf("runtime record command lifecycle: inspect operation cleanup ownership: %w", err)
	}
	if !strings.EqualFold(deleteRule, "CASCADE") {
		return errors.New("runtime record command lifecycle: operation rows require ON DELETE CASCADE")
	}
	if err := VerifyExactTableIndex(ctx, db, runtimeRecordCommandTable, "idx_runtime_record_command_lifecycle", false,
		[]string{"State", "OutcomeExpiresAt", "Id"}); err != nil {
		return fmt.Errorf("runtime record command lifecycle: %w", err)
	}
	return nil
}

type RuntimeRecordCommandLifecycleState struct {
	ExpiredCount int64
	OldestAge    time.Duration
}

// HasExpired is a bounded negative admission hint. ResolveExpired and
// DeleteExpired retain the fenced, locked authority for every durable change.
func (s *RuntimeRecordCommandLifecycleStore) HasExpired(ctx context.Context) (bool, error) {
	if s == nil || s.db == nil {
		return false, errors.New("runtime record command lifecycle: store is unavailable")
	}
	var exists bool
	if err := s.db.QueryRowContext(ctx, `SELECT EXISTS(
	SELECT 1 FROM _System_RuntimeRecordCommand
	WHERE State IN (?, ?) AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)
	LIMIT 1)`, RuntimeRecordCommandAccepted, RuntimeRecordCommandCompleted).Scan(&exists); err != nil {
		return false, fmt.Errorf("runtime record command lifecycle: hint: %w", err)
	}
	return exists, nil
}

func (s *RuntimeRecordCommandLifecycleStore) Observe(ctx context.Context) (RuntimeRecordCommandLifecycleState, error) {
	if s == nil || s.db == nil {
		return RuntimeRecordCommandLifecycleState{}, errors.New("runtime record command lifecycle: store is unavailable")
	}
	var count, oldestMicros int64
	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*),
	       COALESCE(TIMESTAMPDIFF(MICROSECOND, MIN(OutcomeExpiresAt), UTC_TIMESTAMP(6)), 0)
	FROM _System_RuntimeRecordCommand
	WHERE State IN (?, ?) AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)`,
		RuntimeRecordCommandAccepted, RuntimeRecordCommandCompleted,
	).Scan(&count, &oldestMicros); err != nil {
		return RuntimeRecordCommandLifecycleState{}, fmt.Errorf("runtime record command lifecycle: observe: %w", err)
	}
	if count < 0 || oldestMicros < 0 {
		return RuntimeRecordCommandLifecycleState{}, errors.New("runtime record command lifecycle: invalid aggregate")
	}
	return RuntimeRecordCommandLifecycleState{ExpiredCount: count, OldestAge: time.Duration(oldestMicros) * time.Microsecond}, nil
}

// DeleteExpired performs one bounded, fenced cleanup transaction. The parent
// delete owns operation-payload cleanup through the schema's checked CASCADE.
// An in-flight mutation holds the same command row lock, so cleanup cannot
// remove command authority while an operation is committing.
func (s *RuntimeRecordCommandLifecycleStore) DeleteExpired(ctx context.Context, batchSize int) (int64, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return 0, errors.New("runtime record command lifecycle: store is unavailable")
	}
	if batchSize <= 0 || batchSize > 10_000 {
		return 0, errors.New("runtime record command lifecycle: invalid cleanup bounds")
	}
	tx, cleanup, err := BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return 0, fmt.Errorf("runtime record command lifecycle: begin: %w", err)
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (int64, error) {
		if rollbackErr := tx.Rollback(); rollbackErr != nil {
			return 0, errors.Join(cause, fmt.Errorf("runtime record command lifecycle: rollback: %w", rollbackErr))
		}
		return 0, cause
	}
	if err := s.ownership.LockActive(ctx, tx); err != nil {
		return rollback(err)
	}
	rows, err := tx.QueryContext(ctx, `SELECT Id FROM _System_RuntimeRecordCommand
WHERE State = 'COMPLETED' AND OutcomeExpiresAt <= UTC_TIMESTAMP(6)
ORDER BY OutcomeExpiresAt, Id LIMIT ? FOR UPDATE`, batchSize)
	if err != nil {
		return rollback(fmt.Errorf("runtime record command lifecycle: select expired: %w", err))
	}
	ids := make([]string, 0, batchSize)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			_ = rows.Close()
			return rollback(fmt.Errorf("runtime record command lifecycle: scan expired: %w", err))
		}
		ids = append(ids, id)
	}
	if err := rows.Close(); err != nil {
		return rollback(fmt.Errorf("runtime record command lifecycle: close expired rows: %w", err))
	}
	if err := rows.Err(); err != nil {
		return rollback(fmt.Errorf("runtime record command lifecycle: iterate expired: %w", err))
	}
	if len(ids) > 0 {
		placeholders := strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")
		args := make([]any, len(ids))
		for i := range ids {
			args[i] = ids[i]
		}
		result, err := tx.ExecContext(ctx, `DELETE FROM _System_RuntimeRecordCommand WHERE Id IN (`+placeholders+`)`, args...)
		if err != nil {
			return rollback(fmt.Errorf("runtime record command lifecycle: delete expired: %w", err))
		}
		deleted, err := result.RowsAffected()
		if err != nil {
			return rollback(fmt.Errorf("runtime record command lifecycle: deleted rows: %w", err))
		}
		if deleted != int64(len(ids)) {
			return rollback(fmt.Errorf("runtime record command lifecycle: deleted %d of %d locked commands", deleted, len(ids)))
		}
	}
	if err := tx.Commit(); err != nil {
		// Delete is idempotent. The next fenced pass safely resolves an unknown
		// commit by observing whether the rows remain.
		return 0, fmt.Errorf("runtime record command lifecycle: commit outcome unknown: %w", err)
	}
	return int64(len(ids)), nil
}

// LockActiveContext lets the record-domain lifecycle close nonterminal
// commands under the same deployment fence used by terminal cleanup.
func (s *RuntimeRecordCommandLifecycleStore) LockActiveContext(ctx context.Context) error {
	if s == nil || s.ownership == nil {
		return errors.New("runtime record command lifecycle: ownership is unavailable")
	}
	return s.ownership.LockActiveContext(ctx)
}