package persistence

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

var ErrRuntimeWorkOwnershipConflict = errors.New("runtime work ownership activation conflict")

// RuntimeWorkOwnershipActivation is the control-plane-authored compare-and-set
// operation that transfers one durable work family to an exact worker release.
// OwnershipEpoch is derived from ExpectedOwnershipEpoch; callers cannot author
// a competing epoch fact.
type RuntimeWorkOwnershipActivation struct {
	OperationID            string
	WorkFamily             string
	DeploymentID           string
	WorkerReleaseDigest    string
	ExpectedOwnershipEpoch uint64
}

// RuntimeWorkOwnershipControl is the sole mutation path for the generic worker
// deployment fence. Runtime workers only read and lock the resulting authority;
// they can never activate themselves.
type RuntimeWorkOwnershipControl struct {
	db *sql.DB
}

func NewRuntimeWorkOwnershipControl(db *sql.DB) (*RuntimeWorkOwnershipControl, error) {
	if db == nil {
		return nil, errors.New("runtime work ownership control: database is required")
	}
	return &RuntimeWorkOwnershipControl{db: db}, nil
}

func (c *RuntimeWorkOwnershipControl) Activate(ctx context.Context, request RuntimeWorkOwnershipActivation) (RuntimeWorkOwnership, error) {
	if c == nil || c.db == nil {
		return RuntimeWorkOwnership{}, errors.New("runtime work ownership control: authority is unavailable")
	}
	if ctx == nil {
		return RuntimeWorkOwnership{}, errors.New("runtime work ownership control: context is required")
	}
	if !ValidRuntimeOperationID(request.OperationID) {
		return RuntimeWorkOwnership{}, errors.New("runtime work ownership control: operation ID must be canonical and at most 255 bytes")
	}
	if request.ExpectedOwnershipEpoch == ^uint64(0) {
		return RuntimeWorkOwnership{}, errors.New("runtime work ownership control: ownership epoch is exhausted")
	}
	target := RuntimeWorkOwnership{
		WorkFamily: request.WorkFamily, DeploymentID: request.DeploymentID,
		WorkerReleaseDigest: request.WorkerReleaseDigest, OwnershipEpoch: request.ExpectedOwnershipEpoch + 1,
	}
	if err := target.Validate(); err != nil {
		return RuntimeWorkOwnership{}, err
	}

	tx, cleanup, err := BeginPessimisticTx(ctx, c.db)
	if err != nil {
		return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: begin activation: %w", err)
	}
	defer func() { _ = cleanup() }()
	defer func() { _ = tx.Rollback() }()

	current, active, operationID, found, err := loadRuntimeWorkOwnershipForUpdate(ctx, tx, request.WorkFamily)
	if err != nil {
		return RuntimeWorkOwnership{}, err
	}
	if found && operationID == request.OperationID {
		if !active || current != target {
			return RuntimeWorkOwnership{}, fmt.Errorf("%w: operation ID is already bound to different ownership", ErrRuntimeWorkOwnershipConflict)
		}
		if err := ensureRuntimeWorkFenceSlots(ctx, tx, target); err != nil {
			return RuntimeWorkOwnership{}, err
		}
		if err := tx.Commit(); err != nil {
			return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: verify activation: %w", err)
		}
		return target, nil
	}
	if found && current.OwnershipEpoch != request.ExpectedOwnershipEpoch {
		return RuntimeWorkOwnership{}, fmt.Errorf("%w: expected epoch %d, current epoch %d", ErrRuntimeWorkOwnershipConflict, request.ExpectedOwnershipEpoch, current.OwnershipEpoch)
	}
	if !found && request.ExpectedOwnershipEpoch != 0 {
		return RuntimeWorkOwnership{}, fmt.Errorf("%w: expected epoch %d for absent work family", ErrRuntimeWorkOwnershipConflict, request.ExpectedOwnershipEpoch)
	}

	if !found {
		_, err = tx.ExecContext(ctx, `INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, ?, 1, ?)`, target.WorkFamily, target.DeploymentID, target.WorkerReleaseDigest, target.EpochDecimal(), request.OperationID)
		if err == nil {
			err = ensureRuntimeWorkFenceSlots(ctx, tx, target)
		}
	} else {
		if err = ensureRuntimeWorkFenceSlots(ctx, tx, current); err != nil {
			return RuntimeWorkOwnership{}, err
		}
		fenceResult, fenceErr := tx.ExecContext(ctx, `UPDATE _System_RuntimeWorkFence
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE WorkFamily = ? AND DeploymentId = ? AND WorkerReleaseDigest = ? AND OwnershipEpoch = ?
  AND FenceSlot >= 0 AND FenceSlot < ?`,
			target.DeploymentID, target.WorkerReleaseDigest, target.EpochDecimal(), target.WorkFamily,
			current.DeploymentID, current.WorkerReleaseDigest, current.EpochDecimal(), RuntimeWorkFenceSlots)
		if fenceErr != nil {
			return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: transfer effect fences: %w", fenceErr)
		}
		fenceRows, fenceErr := fenceResult.RowsAffected()
		if fenceErr != nil || fenceRows != RuntimeWorkFenceSlots {
			if fenceErr == nil {
				fenceErr = ErrRuntimeWorkOwnershipConflict
			}
			return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: transfer effect fences: %w", fenceErr)
		}
		var result sql.Result
		result, err = tx.ExecContext(ctx, `UPDATE _System_RuntimeWorkOwnership
SET DeploymentId = ?, WorkerReleaseDigest = ?, OwnershipEpoch = ?, IsActive = 1,
    OperationId = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE WorkFamily = ? AND OwnershipEpoch = ? AND OperationId = ?`,
			target.DeploymentID, target.WorkerReleaseDigest, target.EpochDecimal(), request.OperationID,
			target.WorkFamily, current.EpochDecimal(), operationID)
		if err == nil {
			var affected int64
			affected, err = result.RowsAffected()
			if err == nil && affected != 1 {
				err = ErrRuntimeWorkOwnershipConflict
			}
		}
	}
	if err != nil {
		return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: activate: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return RuntimeWorkOwnership{}, fmt.Errorf("runtime work ownership control: commit activation: %w", err)
	}
	return target, nil
}

func ensureRuntimeWorkFenceSlots(ctx context.Context, tx *sql.Tx, owner RuntimeWorkOwnership) error {
	total, matching, err := inspectRuntimeWorkFenceSlots(ctx, tx, owner)
	if err != nil {
		return err
	}
	if total == RuntimeWorkFenceSlots && matching == RuntimeWorkFenceSlots {
		return nil
	}
	if total != 0 {
		return fmt.Errorf("%w: effect fences do not exactly match active ownership", ErrRuntimeWorkOwnershipConflict)
	}

	var statement strings.Builder
	statement.WriteString(`INSERT INTO _System_RuntimeWorkFence
(Id, WorkFamily, FenceSlot, DeploymentId, WorkerReleaseDigest, OwnershipEpoch) VALUES `)
	args := make([]interface{}, 0, RuntimeWorkFenceSlots*6)
	for slot := 0; slot < RuntimeWorkFenceSlots; slot++ {
		if slot > 0 {
			statement.WriteString(",")
		}
		statement.WriteString("(?, ?, ?, ?, ?, ?)")
		args = append(args, fmt.Sprintf("%s:%02d", owner.WorkFamily, slot), owner.WorkFamily, slot,
			owner.DeploymentID, owner.WorkerReleaseDigest, owner.EpochDecimal())
	}
	result, err := tx.ExecContext(ctx, statement.String(), args...)
	if err != nil {
		return fmt.Errorf("runtime work ownership control: seed effect fences: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil || rows != RuntimeWorkFenceSlots {
		if err == nil {
			err = ErrRuntimeWorkOwnershipConflict
		}
		return fmt.Errorf("runtime work ownership control: seed effect fences: %w", err)
	}
	total, matching, err = inspectRuntimeWorkFenceSlots(ctx, tx, owner)
	if err != nil {
		return err
	}
	if total != RuntimeWorkFenceSlots || matching != RuntimeWorkFenceSlots {
		return fmt.Errorf("%w: seeded effect fences do not exactly match active ownership", ErrRuntimeWorkOwnershipConflict)
	}
	return nil
}

func inspectRuntimeWorkFenceSlots(ctx context.Context, tx *sql.Tx, owner RuntimeWorkOwnership) (int, int, error) {
	var total, matching int
	err := tx.QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(CASE
  WHEN FenceSlot >= 0 AND FenceSlot < ? AND DeploymentId = ? AND WorkerReleaseDigest = ? AND OwnershipEpoch = ?
  THEN 1 ELSE 0 END), 0)
FROM _System_RuntimeWorkFence WHERE WorkFamily = ?`, RuntimeWorkFenceSlots, owner.DeploymentID,
		owner.WorkerReleaseDigest, owner.EpochDecimal(), owner.WorkFamily).Scan(&total, &matching)
	if err != nil {
		return 0, 0, fmt.Errorf("runtime work ownership control: inspect effect fences: %w", err)
	}
	return total, matching, nil
}

func loadRuntimeWorkOwnershipForUpdate(ctx context.Context, tx *sql.Tx, workFamily string) (RuntimeWorkOwnership, bool, string, bool, error) {
	var current RuntimeWorkOwnership
	var epoch, operationID string
	var active bool
	err := tx.QueryRowContext(ctx, `SELECT DeploymentId, WorkerReleaseDigest, CAST(OwnershipEpoch AS CHAR), IsActive, OperationId
FROM _System_RuntimeWorkOwnership WHERE WorkFamily = ? FOR UPDATE`, workFamily).
		Scan(&current.DeploymentID, &current.WorkerReleaseDigest, &epoch, &active, &operationID)
	if errors.Is(err, sql.ErrNoRows) {
		return RuntimeWorkOwnership{}, false, "", false, nil
	}
	if err != nil {
		return RuntimeWorkOwnership{}, false, "", false, fmt.Errorf("runtime work ownership control: lock current authority: %w", err)
	}
	current.WorkFamily = workFamily
	current.OwnershipEpoch, err = strconv.ParseUint(epoch, 10, 64)
	if err != nil || current.OwnershipEpoch == 0 {
		return RuntimeWorkOwnership{}, false, "", false, errors.New("runtime work ownership control: stored epoch is invalid")
	}
	if err := current.Validate(); err != nil {
		return RuntimeWorkOwnership{}, false, "", false, fmt.Errorf("runtime work ownership control: stored authority is invalid: %w", err)
	}
	if !runtimeDeploymentPattern.MatchString(operationID) {
		return RuntimeWorkOwnership{}, false, "", false, errors.New("runtime work ownership control: stored operation ID is invalid")
	}
	return current, active, operationID, true, nil
}