package persistence

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"hash/fnv"
	"regexp"
	"strconv"
	"strings"
)

const (
	RuntimeWorkOwnershipTable = "_System_RuntimeWorkOwnership"
	RuntimeWorkFenceTable     = "_System_RuntimeWorkFence"
	// RuntimeWorkFenceSlots is a protocol constant, not deployment tuning. A
	// transfer drains this complete fixed set before changing the active tuple,
	// while independent durable keys lock only their stable slot.
	RuntimeWorkFenceSlots = 64
)

var (
	runtimeWorkFamilyPattern = regexp.MustCompile(`^[a-z][a-z0-9.-]{0,63}$`)
	runtimeDeploymentPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,254}$`)
	runtimeDigestPattern     = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
	runtimeSHA256Pattern     = regexp.MustCompile(`^[0-9a-f]{64}$`)

	// ErrRuntimeWorkOwnershipLost means the configured Runtime Worker
	// deployment is no longer the database authority for its work family.
	// This is process-fatal, not a row-level lease race.
	ErrRuntimeWorkOwnershipLost = errors.New("runtime work ownership lost")
)

// RuntimeWorkOwnership is the platform-owned deployment fence shared by all
// Runtime Worker families. A family package owns its payload and effects; this
// tuple alone owns which deployment generation may claim or commit them.
type RuntimeWorkOwnership struct {
	WorkFamily          string
	DeploymentID        string
	WorkerReleaseDigest string
	OwnershipEpoch      uint64
}

// RuntimePoisonRepair is the shared operator-authored compare-and-set proof
// for durable Runtime work. A family store still owns its row schema and replay
// transition; this value prevents every queue from inventing another repair
// identity grammar.
type RuntimePoisonRepair struct {
	RowID                   string
	ExpectedRetryCount      int
	ExpectedLastErrorSHA256 string
	RepairID                string
	Reason                  string
}

func (r RuntimePoisonRepair) Validate() error {
	if r.RowID == "" || len(r.RowID) > 255 || r.ExpectedRetryCount <= 0 ||
		!runtimeSHA256Pattern.MatchString(r.ExpectedLastErrorSHA256) ||
		!ValidRuntimeOperationID(r.RepairID) || strings.TrimSpace(r.Reason) == "" || len(r.Reason) > 1024 {
		return errors.New("runtime poison repair: exact row, poison observation, repair ID, and reason are required")
	}
	return nil
}

func (r RuntimePoisonRepair) Matches(retryCount int, lastError string) bool {
	digest := sha256.Sum256([]byte(lastError))
	return retryCount == r.ExpectedRetryCount && hex.EncodeToString(digest[:]) == r.ExpectedLastErrorSHA256
}

func (o RuntimeWorkOwnership) Validate() error {
	if !runtimeWorkFamilyPattern.MatchString(o.WorkFamily) {
		return errors.New("runtime work ownership: work family must be canonical and at most 64 bytes")
	}
	if !runtimeDeploymentPattern.MatchString(o.DeploymentID) {
		return errors.New("runtime work ownership: deployment ID must be canonical and at most 255 bytes")
	}
	if !runtimeDigestPattern.MatchString(o.WorkerReleaseDigest) {
		return errors.New("runtime work ownership: worker release digest must be canonical sha256")
	}
	if o.OwnershipEpoch == 0 {
		return errors.New("runtime work ownership: ownership epoch must be positive")
	}
	return nil
}

func (o RuntimeWorkOwnership) EpochDecimal() string {
	return fmt.Sprintf("%d", o.OwnershipEpoch)
}

// ValidRuntimeOperationID validates operator-authored activation and repair
// identities. Deployment identity remains part of RuntimeWorkOwnership.Validate.
func ValidRuntimeOperationID(value string) bool { return runtimeDeploymentPattern.MatchString(value) }

func RuntimeWorkOwnershipLost(cause error) error {
	if cause == nil {
		return ErrRuntimeWorkOwnershipLost
	}
	return fmt.Errorf("%w: %v", ErrRuntimeWorkOwnershipLost, cause)
}

// RuntimeWorkOwnershipRepository is the sole database read/lock path for the
// generic worker deployment authority. Family stores must call LockActive in
// the same transaction as every durable effect or disposition.
type RuntimeWorkOwnershipRepository struct {
	db       *sql.DB
	expected RuntimeWorkOwnership
}

func NewRuntimeWorkOwnershipRepository(db *sql.DB, expected RuntimeWorkOwnership) (*RuntimeWorkOwnershipRepository, error) {
	if err := expected.Validate(); err != nil {
		return nil, err
	}
	if db == nil {
		return nil, errors.New("runtime work ownership: database is required")
	}
	return &RuntimeWorkOwnershipRepository{db: db, expected: expected}, nil
}

var runtimeWorkOwnershipRequiredColumns = map[string]string{
	"WorkFamily":          "varchar(64)",
	"DeploymentId":        "varchar(255)",
	"WorkerReleaseDigest": "char(71)",
	"OwnershipEpoch":      "decimal(20,0)",
	"IsActive":            "tinyint(1)",
	"OperationId":         "varchar(255)",
}

var runtimeWorkFenceRequiredColumns = map[string]string{
	"Id":                  "varchar(255)",
	"WorkFamily":          "varchar(64)",
	"FenceSlot":           "int",
	"DeploymentId":        "varchar(255)",
	"WorkerReleaseDigest": "char(71)",
	"OwnershipEpoch":      "decimal(20,0)",
}

func (r *RuntimeWorkOwnershipRepository) CheckReady(ctx context.Context) error {
	if r == nil || r.db == nil {
		return errors.New("runtime work ownership: repository is unavailable")
	}
	if err := verifyRuntimeWorkOwnershipAuthoritySchema(ctx, r.db); err != nil {
		return err
	}
	return r.VerifyActive(ctx)
}

// CheckFencesReady proves the additive fixed-slot fence is complete and bound
// to the same exact tuple as the singleton admission authority. Only work
// families that execute independent keyed effects need this stronger surface.
func (r *RuntimeWorkOwnershipRepository) CheckFencesReady(ctx context.Context) error {
	if r == nil || r.db == nil {
		return errors.New("runtime work fence: repository is unavailable")
	}
	if err := VerifyRequiredTableColumns(ctx, r.db, RuntimeWorkFenceTable, runtimeWorkFenceRequiredColumns); err != nil {
		return fmt.Errorf("runtime work fence: %w", err)
	}
	if err := VerifyExactTableIndex(ctx, r.db, RuntimeWorkFenceTable, "ux_runtime_work_fence_slot", true, []string{"WorkFamily", "FenceSlot"}); err != nil {
		return fmt.Errorf("runtime work fence: %w", err)
	}
	var total, exact int
	if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(
DeploymentId = ? AND WorkerReleaseDigest = ? AND OwnershipEpoch = ?
AND FenceSlot >= 0 AND FenceSlot < ?), 0)
FROM _System_RuntimeWorkFence WHERE WorkFamily = ?`,
		r.expected.DeploymentID, r.expected.WorkerReleaseDigest, r.expected.EpochDecimal(),
		RuntimeWorkFenceSlots, r.expected.WorkFamily,
	).Scan(&total, &exact); err != nil {
		return fmt.Errorf("runtime work fence: verify exact slots: %w", err)
	}
	if total != RuntimeWorkFenceSlots || exact != RuntimeWorkFenceSlots {
		return RuntimeWorkOwnershipLost(fmt.Errorf("fixed slots do not match configured deployment: total=%d exact=%d", total, exact))
	}
	return nil
}

func verifyRuntimeWorkOwnershipAuthoritySchema(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return errors.New("runtime work ownership: database is required")
	}
	if err := VerifyRequiredTableColumns(ctx, db, RuntimeWorkOwnershipTable, runtimeWorkOwnershipRequiredColumns); err != nil {
		return fmt.Errorf("runtime work ownership: %w", err)
	}
	// WorkFamily is the single durable authority identity. Without the exact
	// unique primary key two rows could let different deployments each satisfy
	// active-family readiness or a transactional LockActive under competing
	// ownership epochs.
	if err := VerifyExactTableIndex(ctx, db, RuntimeWorkOwnershipTable, "PRIMARY", true, []string{"WorkFamily"}); err != nil {
		return fmt.Errorf("runtime work ownership: %w", err)
	}
	return nil
}

// VerifyActiveRuntimeWorkFamily proves that the canonical ownership table has
// one valid active owner for a required downstream work family. Producers use
// this only as readiness evidence: it neither selects a worker nor creates or
// transfers authority. Consumers still require an exact expected tuple and
// transactionally call RuntimeWorkOwnershipRepository.LockActive.
func VerifyActiveRuntimeWorkFamily(ctx context.Context, db *sql.DB, workFamily string) error {
	if ctx == nil {
		return errors.New("runtime work ownership: context is required")
	}
	if !runtimeWorkFamilyPattern.MatchString(workFamily) {
		return errors.New("runtime work ownership: required work family is invalid")
	}
	if err := verifyRuntimeWorkOwnershipAuthoritySchema(ctx, db); err != nil {
		return err
	}
	var ownership RuntimeWorkOwnership
	var epoch string
	err := db.QueryRowContext(ctx, readActiveRuntimeWorkOwnershipSQL, workFamily).
		Scan(&ownership.DeploymentID, &ownership.WorkerReleaseDigest, &epoch)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeWorkOwnershipLost(errors.New("required work family has no active owner"))
	}
	if err != nil {
		return fmt.Errorf("runtime work ownership: read required active family: %w", err)
	}
	ownership.WorkFamily = workFamily
	ownership.OwnershipEpoch, err = strconv.ParseUint(epoch, 10, 64)
	if err != nil {
		return errors.New("runtime work ownership: required active family has invalid epoch")
	}
	if err := ownership.Validate(); err != nil {
		return fmt.Errorf("runtime work ownership: required active family is invalid: %w", err)
	}
	return nil
}

const readActiveRuntimeWorkOwnershipSQL = `SELECT DeploymentId, WorkerReleaseDigest, CAST(OwnershipEpoch AS CHAR)
FROM _System_RuntimeWorkOwnership
WHERE WorkFamily = ? AND IsActive = 1`

const lockActiveRuntimeWorkOwnershipSQL = `SELECT DeploymentId, WorkerReleaseDigest, CAST(OwnershipEpoch AS CHAR)
FROM _System_RuntimeWorkOwnership
WHERE WorkFamily = ? AND DeploymentId = ? AND WorkerReleaseDigest = ?
  AND OwnershipEpoch = ? AND IsActive = 1
FOR UPDATE`

const lockActiveRuntimeWorkFenceSQL = `SELECT DeploymentId, WorkerReleaseDigest, CAST(OwnershipEpoch AS CHAR)
FROM _System_RuntimeWorkFence
WHERE WorkFamily = ? AND FenceSlot = ? AND DeploymentId = ? AND WorkerReleaseDigest = ?
  AND OwnershipEpoch = ?
FOR UPDATE`

func (r *RuntimeWorkOwnershipRepository) VerifyActive(ctx context.Context) error {
	if r == nil || r.db == nil {
		return errors.New("runtime work ownership: repository is unavailable")
	}
	var deploymentID, releaseDigest, epoch string
	err := r.db.QueryRowContext(ctx, readActiveRuntimeWorkOwnershipSQL, r.expected.WorkFamily).
		Scan(&deploymentID, &releaseDigest, &epoch)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeWorkOwnershipLost(errors.New("active authority row is missing"))
	}
	if err != nil {
		return fmt.Errorf("runtime work ownership: read active authority: %w", err)
	}
	if deploymentID != r.expected.DeploymentID || releaseDigest != r.expected.WorkerReleaseDigest || epoch != r.expected.EpochDecimal() {
		return RuntimeWorkOwnershipLost(errors.New("active authority does not match configured deployment"))
	}
	return nil
}

func (r *RuntimeWorkOwnershipRepository) LockActive(ctx context.Context, tx *sql.Tx) error {
	if r == nil || tx == nil {
		return errors.New("runtime work ownership: active transaction is required")
	}
	var deploymentID, releaseDigest, epoch string
	err := tx.QueryRowContext(ctx, lockActiveRuntimeWorkOwnershipSQL,
		r.expected.WorkFamily, r.expected.DeploymentID, r.expected.WorkerReleaseDigest, r.expected.EpochDecimal()).
		Scan(&deploymentID, &releaseDigest, &epoch)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeWorkOwnershipLost(errors.New("configured authority is no longer active"))
	}
	if err != nil {
		return fmt.Errorf("runtime work ownership: lock active authority: %w", err)
	}
	return nil
}

// LockActiveContext is the generic engine seam used by work-family processors
// whose transaction is carried in context. It keeps knowledge of the
// persistence transaction key inside its owning module.
func (r *RuntimeWorkOwnershipRepository) LockActiveContext(ctx context.Context) error {
	if r == nil {
		return errors.New("runtime work ownership: repository is unavailable")
	}
	tx, ok := ctx.Value(txContextKey{}).(*sql.Tx)
	if !ok || tx == nil {
		return errors.New("runtime work ownership: active transaction is required")
	}
	return r.LockActive(ctx, tx)
}

// LockActiveFenceContext locks one deterministic effect slot inside the
// caller's transaction. It preserves the ownership-transfer commit barrier
// without making unrelated durable keys wait behind one global row.
func (r *RuntimeWorkOwnershipRepository) LockActiveFenceContext(ctx context.Context, durableKey string) error {
	if r == nil {
		return errors.New("runtime work fence: repository is unavailable")
	}
	if durableKey == "" {
		return errors.New("runtime work fence: durable key is required")
	}
	tx, ok := ctx.Value(txContextKey{}).(*sql.Tx)
	if !ok || tx == nil {
		return errors.New("runtime work fence: active transaction is required")
	}
	return r.LockActiveFence(ctx, tx, durableKey)
}

func RuntimeWorkFenceSlot(durableKey string) int {
	h := fnv.New32a()
	_, _ = h.Write([]byte(durableKey))
	return int(h.Sum32() % RuntimeWorkFenceSlots)
}

func (r *RuntimeWorkOwnershipRepository) LockActiveFence(ctx context.Context, tx *sql.Tx, durableKey string) error {
	if r == nil || tx == nil {
		return errors.New("runtime work fence: active transaction is required")
	}
	if durableKey == "" {
		return errors.New("runtime work fence: durable key is required")
	}
	slot := RuntimeWorkFenceSlot(durableKey)
	var deploymentID, releaseDigest, epoch string
	err := tx.QueryRowContext(ctx, lockActiveRuntimeWorkFenceSQL,
		r.expected.WorkFamily, slot, r.expected.DeploymentID,
		r.expected.WorkerReleaseDigest, r.expected.EpochDecimal(),
	).Scan(&deploymentID, &releaseDigest, &epoch)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeWorkOwnershipLost(errors.New("configured effect fence is no longer active"))
	}
	if err != nil {
		return fmt.Errorf("runtime work fence: lock active slot: %w", err)
	}
	return nil
}