package catalogverificationtx

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

	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	systemauthorityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/systemlayoutbundleauthorityv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogsystemfence"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogsystemprotocol"
)

var (
	ErrSystemOwnerNotFound = errors.New(
		"catalog verification transaction: system owner fact not found",
	)
	ErrSystemOwnerConflict = errors.New(
		"catalog verification transaction: system owner fact conflict",
	)
	ErrSystemOwnerCorrupt = errors.New(
		"catalog verification transaction: system owner durable state is corrupt",
	)
)

// SystemOwnerRequest is the exact required system-owned key selected from the
// retained Catalog source. Only catalogpublication can place it in a claim.
type SystemOwnerRequest struct {
	BundleKey layoutbundlekeyv1.BundleKey
}

// SystemOwnerFact is the complete locked ACTIVE system slot. SlotDigest is
// retained as the private authority-fact identity; it is not an evidence field.
type SystemOwnerFact struct {
	Slot       systemauthorityv1.SystemLayoutBundleSlotV1
	SlotDigest string
}

type systemCurrent struct {
	slot          systemauthorityv1.SystemLayoutBundleSlotV1
	digest        string
	lastCommandID systemauthorityv1.CommandID
}

type systemHistory struct {
	commandID systemauthorityv1.CommandID
	ordinal   uint8
	slot      systemauthorityv1.SystemLayoutBundleSlotV1
	digest    string
	prior     sql.NullString
}

type systemCommand struct {
	commandID     systemauthorityv1.CommandID
	principalID   string
	auditDigest   string
	request       systemauthorityv1.SystemLayoutBundleSlotCommandRequest
	requestDigest string
}

type systemOutcome struct {
	commandID     systemauthorityv1.CommandID
	requestDigest string
	principalID   string
	auditDigest   string
	result        string
	outcomeDigest string
}

// VerifySystemOwner holds the shared system-authority advisory fence and locks
// the exact current row plus its complete history, commands, and outcomes in
// the retained source transaction. The advisory fence remains owned by token
// until that transaction is committed or rolled back.
func VerifySystemOwner(
	ctx context.Context,
	token *Token,
	db *sql.DB,
	request SystemOwnerRequest,
) (SystemOwnerFact, error) {
	key, err := validateSystemOwnerRequest(request)
	if err != nil {
		return SystemOwnerFact{}, err
	}
	if err := token.acquireSystemAuthorityFence(ctx, db); err != nil {
		if errors.Is(err, catalogsystemfence.ErrNotAcquired) {
			return SystemOwnerFact{}, fmt.Errorf(
				"%w: system authority fence is unavailable",
				ErrSystemOwnerConflict,
			)
		}
		return SystemOwnerFact{}, err
	}
	tx, err := token.acquire(db)
	if err != nil {
		return SystemOwnerFact{}, err
	}
	defer token.release()

	current, err := lockExactActiveSystemCurrent(ctx, tx, key)
	if err != nil {
		return SystemOwnerFact{}, err
	}
	history, err := lockCompleteSystemHistory(ctx, tx, key)
	if err != nil {
		return SystemOwnerFact{}, err
	}
	if err := validateSystemCurrentHistory(current, history); err != nil {
		return SystemOwnerFact{}, err
	}
	commands, err := lockCompleteSystemCommands(ctx, tx, key)
	if err != nil {
		return SystemOwnerFact{}, err
	}
	if err := lockAndValidateSystemOutcomes(
		ctx,
		tx,
		current,
		history[0],
		commands,
	); err != nil {
		return SystemOwnerFact{}, err
	}

	return SystemOwnerFact{
		Slot:       current.slot,
		SlotDigest: current.digest,
	}, nil
}

// VerifySystemOwnerClaim is the sole bridge from an opaque source system claim
// to the fixed-purpose system authority query.
func VerifySystemOwnerClaim(
	ctx context.Context,
	db *sql.DB,
	claim *SystemClaim,
) (SystemOwnerFact, error) {
	token, request, err := claim.acquireInput()
	if err != nil {
		return SystemOwnerFact{}, err
	}
	fact, err := VerifySystemOwner(ctx, token, db, request)
	if err != nil {
		claim.Abort()
		return SystemOwnerFact{}, err
	}
	if err := claim.lease.markClaimVerified(claim.lifecycle); err != nil {
		claim.Abort()
		return SystemOwnerFact{}, err
	}
	return fact, nil
}

func validateSystemOwnerRequest(
	request SystemOwnerRequest,
) (systemauthorityv1.BundleKey, error) {
	if err := layoutbundlekeyv1.ValidateBundleKey(request.BundleKey); err != nil {
		return systemauthorityv1.BundleKey{}, fmt.Errorf(
			"%w: BundleKey: %v",
			ErrSystemOwnerConflict,
			err,
		)
	}
	if request.BundleKey.SlotOwnerKind != layoutbundlekeyv1.SlotOwnerKindSystem ||
		request.BundleKey.SlotOwnerNamespace != layoutbundlekeyv1.SystemSlotOwnerNamespace {
		return systemauthorityv1.BundleKey{}, fmt.Errorf(
			"%w: BundleKey is not system-owned",
			ErrSystemOwnerConflict,
		)
	}
	key := systemauthorityv1.BundleKey{
		BundleKeyVersion:   layoutbundlekeyv1.BundleKeyVersion,
		SlotOwnerKind:      request.BundleKey.SlotOwnerKind,
		SlotOwnerNamespace: request.BundleKey.SlotOwnerNamespace,
		LocalComponentKey:  request.BundleKey.LocalComponentKey,
	}
	if err := systemauthorityv1.ValidateBundleKey(key); err != nil {
		return systemauthorityv1.BundleKey{}, fmt.Errorf(
			"%w: BundleKey: %v",
			ErrSystemOwnerConflict,
			err,
		)
	}
	return key, nil
}

func lockExactActiveSystemCurrent(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) (systemCurrent, error) {
	var current systemCurrent
	var rawOwner, rawState, storedCommandID string
	var storedNamespace, storedLocalKey []byte
	err := tx.QueryRowContext(ctx, `SELECT BundleKeyVersion, SlotOwnerKind,
SlotOwnerNamespace, LocalComponentKey, ValueVersion, State, ValueDigest,
LastCommandId
FROM _System_CatalogSystemLayoutBundleSlot
WHERE BundleKeyVersion = ? AND SlotOwnerKind = ?
AND SlotOwnerNamespace = ? AND LocalComponentKey = ? FOR UPDATE`,
		key.BundleKeyVersion,
		string(key.SlotOwnerKind),
		key.SlotOwnerNamespace,
		string(key.LocalComponentKey),
	).Scan(
		&current.slot.BundleKey.BundleKeyVersion,
		&rawOwner,
		&storedNamespace,
		&storedLocalKey,
		&current.slot.ValueVersion,
		&rawState,
		&current.digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return systemCurrent{}, ErrSystemOwnerNotFound
	}
	if err != nil {
		return systemCurrent{}, err
	}
	current.slot.BundleKey.SlotOwnerKind = layoutbundlekeyv1.SlotOwnerKind(rawOwner)
	current.slot.BundleKey.SlotOwnerNamespace = string(storedNamespace)
	current.slot.BundleKey.LocalComponentKey =
		layoutbundlekeyv1.LocalComponentKey(storedLocalKey)
	current.slot.State = systemauthorityv1.SlotState(rawState)
	current.lastCommandID = systemauthorityv1.CommandID(storedCommandID)

	if current.slot.BundleKey != key {
		return systemCurrent{}, fmt.Errorf(
			"%w: current slot identity differs from its exact lookup",
			ErrSystemOwnerCorrupt,
		)
	}
	if err := systemauthorityv1.ValidateSlotValue(current.slot); err != nil {
		return systemCurrent{}, fmt.Errorf(
			"%w: current slot value: %v",
			ErrSystemOwnerCorrupt,
			err,
		)
	}
	computed, err := systemauthorityv1.SlotValueDigest(current.slot)
	if err != nil || computed != current.digest {
		return systemCurrent{}, fmt.Errorf(
			"%w: current slot value digest differs",
			ErrSystemOwnerCorrupt,
		)
	}
	if err := systemauthorityv1.ValidateCommandID(current.lastCommandID); err != nil {
		return systemCurrent{}, fmt.Errorf(
			"%w: current LastCommandId is invalid",
			ErrSystemOwnerCorrupt,
		)
	}
	if current.slot.State != systemauthorityv1.SlotStateActive {
		return systemCurrent{}, fmt.Errorf(
			"%w: system slot is not ACTIVE",
			ErrSystemOwnerConflict,
		)
	}
	return current, nil
}

func lockCompleteSystemHistory(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) ([]systemHistory, error) {
	rows, err := tx.QueryContext(ctx, `SELECT BundleKeyVersion, SlotOwnerKind,
SlotOwnerNamespace, LocalComponentKey, TransitionOrdinal, CommandId,
ValueVersion, State, ValueDigest, PriorValueDigest
FROM _System_CatalogSystemLayoutBundleSlotHistory
WHERE BundleKeyVersion = ? AND SlotOwnerKind = ?
AND SlotOwnerNamespace = ? AND LocalComponentKey = ?
ORDER BY TransitionOrdinal FOR UPDATE`,
		key.BundleKeyVersion,
		string(key.SlotOwnerKind),
		key.SlotOwnerNamespace,
		string(key.LocalComponentKey),
	)
	if err != nil {
		return nil, err
	}
	var history []systemHistory
	for rows.Next() {
		var entry systemHistory
		var rawOwner, rawState, storedCommandID string
		var storedNamespace, storedLocalKey, rawOrdinal []byte
		if err := rows.Scan(
			&entry.slot.BundleKey.BundleKeyVersion,
			&rawOwner,
			&storedNamespace,
			&storedLocalKey,
			&rawOrdinal,
			&storedCommandID,
			&entry.slot.ValueVersion,
			&rawState,
			&entry.digest,
			&entry.prior,
		); err != nil {
			_ = rows.Close()
			return nil, err
		}
		entry.slot.BundleKey.SlotOwnerKind =
			layoutbundlekeyv1.SlotOwnerKind(rawOwner)
		entry.slot.BundleKey.SlotOwnerNamespace = string(storedNamespace)
		entry.slot.BundleKey.LocalComponentKey =
			layoutbundlekeyv1.LocalComponentKey(storedLocalKey)
		entry.slot.State = systemauthorityv1.SlotState(rawState)
		entry.commandID = systemauthorityv1.CommandID(storedCommandID)
		entry.ordinal, err = parseSystemOrdinal(rawOrdinal)
		if err != nil {
			_ = rows.Close()
			return nil, err
		}
		if entry.slot.BundleKey != key {
			_ = rows.Close()
			return nil, fmt.Errorf(
				"%w: history identity differs from its exact lookup",
				ErrSystemOwnerCorrupt,
			)
		}
		if err := validateSystemHistory(entry); err != nil {
			_ = rows.Close()
			return nil, err
		}
		history = append(history, entry)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	return history, nil
}

func validateSystemHistory(entry systemHistory) error {
	if err := systemauthorityv1.ValidateCommandID(entry.commandID); err != nil {
		return fmt.Errorf(
			"%w: history CommandId is invalid",
			ErrSystemOwnerCorrupt,
		)
	}
	if err := systemauthorityv1.ValidateSlotValue(entry.slot); err != nil {
		return fmt.Errorf(
			"%w: history slot value: %v",
			ErrSystemOwnerCorrupt,
			err,
		)
	}
	computed, err := systemauthorityv1.SlotValueDigest(entry.slot)
	if err != nil || computed != entry.digest {
		return fmt.Errorf(
			"%w: history slot value digest differs",
			ErrSystemOwnerCorrupt,
		)
	}
	switch entry.ordinal {
	case 1:
		if entry.slot.State != systemauthorityv1.SlotStateActive ||
			entry.prior.Valid {
			return fmt.Errorf(
				"%w: creation history is not ACTIVE from ABSENT",
				ErrSystemOwnerCorrupt,
			)
		}
	case 2:
		if entry.slot.State != systemauthorityv1.SlotStateTombstone ||
			!entry.prior.Valid ||
			validateSystemDigest(entry.prior.String) != nil {
			return fmt.Errorf(
				"%w: retirement history is not TOMBSTONE from EXACT",
				ErrSystemOwnerCorrupt,
			)
		}
	default:
		return fmt.Errorf(
			"%w: invalid history ordinal %d",
			ErrSystemOwnerCorrupt,
			entry.ordinal,
		)
	}
	return nil
}

func validateSystemCurrentHistory(
	current systemCurrent,
	history []systemHistory,
) error {
	if len(history) != 1 {
		return fmt.Errorf(
			"%w: ACTIVE current must have exactly one creation-history row",
			ErrSystemOwnerCorrupt,
		)
	}
	creation := history[0]
	if creation.ordinal != 1 ||
		creation.commandID != current.lastCommandID ||
		creation.slot != current.slot ||
		creation.digest != current.digest {
		return fmt.Errorf(
			"%w: ACTIVE current and creation history differ",
			ErrSystemOwnerCorrupt,
		)
	}
	return nil
}

func lockCompleteSystemCommands(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) ([]systemCommand, error) {
	rows, err := tx.QueryContext(ctx, `SELECT CommandId, PrincipalId,
PrincipalAuditDigest, RequestDigest, Operation,
TargetBundleKeyVersion, TargetSlotOwnerKind, TargetSlotOwnerNamespace,
TargetLocalComponentKey, DesiredValueVersion, DesiredBundleKeyVersion,
DesiredSlotOwnerKind, DesiredSlotOwnerNamespace, DesiredLocalComponentKey,
DesiredState, ExpectedPriorVariant, ExpectedPriorDigest
FROM _System_CatalogSystemLayoutBundleSlotCommand
WHERE TargetBundleKeyVersion = ? AND TargetSlotOwnerKind = ?
AND TargetSlotOwnerNamespace = ? AND TargetLocalComponentKey = ?
ORDER BY CommandId FOR UPDATE`,
		key.BundleKeyVersion,
		string(key.SlotOwnerKind),
		key.SlotOwnerNamespace,
		string(key.LocalComponentKey),
	)
	if err != nil {
		return nil, err
	}
	var commands []systemCommand
	for rows.Next() {
		command, err := scanSystemCommand(rows)
		if err != nil {
			_ = rows.Close()
			return nil, err
		}
		if command.request.TargetBundleKey != key {
			_ = rows.Close()
			return nil, fmt.Errorf(
				"%w: command target differs from its exact lookup",
				ErrSystemOwnerCorrupt,
			)
		}
		commands = append(commands, command)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	return commands, nil
}

type systemCommandScanner interface {
	Scan(dest ...any) error
}

func scanSystemCommand(scanner systemCommandScanner) (systemCommand, error) {
	var command systemCommand
	var storedCommandID, rawOperation, rawTargetOwner, rawDesiredOwner string
	var principalID, targetNamespace, targetLocalKey []byte
	var desiredNamespace, desiredLocalKey []byte
	var rawDesiredState, rawPriorVariant string
	var priorDigest sql.NullString
	if err := scanner.Scan(
		&storedCommandID,
		&principalID,
		&command.auditDigest,
		&command.requestDigest,
		&rawOperation,
		&command.request.TargetBundleKey.BundleKeyVersion,
		&rawTargetOwner,
		&targetNamespace,
		&targetLocalKey,
		&command.request.DesiredSlotValue.ValueVersion,
		&command.request.DesiredSlotValue.BundleKey.BundleKeyVersion,
		&rawDesiredOwner,
		&desiredNamespace,
		&desiredLocalKey,
		&rawDesiredState,
		&rawPriorVariant,
		&priorDigest,
	); err != nil {
		return systemCommand{}, err
	}
	command.commandID = systemauthorityv1.CommandID(storedCommandID)
	command.principalID = string(principalID)
	command.request.Operation = systemauthorityv1.Operation(rawOperation)
	command.request.TargetBundleKey.SlotOwnerKind =
		layoutbundlekeyv1.SlotOwnerKind(rawTargetOwner)
	command.request.TargetBundleKey.SlotOwnerNamespace = string(targetNamespace)
	command.request.TargetBundleKey.LocalComponentKey =
		layoutbundlekeyv1.LocalComponentKey(targetLocalKey)
	command.request.DesiredSlotValue.BundleKey.SlotOwnerKind =
		layoutbundlekeyv1.SlotOwnerKind(rawDesiredOwner)
	command.request.DesiredSlotValue.BundleKey.SlotOwnerNamespace =
		string(desiredNamespace)
	command.request.DesiredSlotValue.BundleKey.LocalComponentKey =
		layoutbundlekeyv1.LocalComponentKey(desiredLocalKey)
	command.request.DesiredSlotValue.State =
		systemauthorityv1.SlotState(rawDesiredState)
	command.request.ExpectedPrior.Variant =
		systemauthorityv1.ExpectedPriorVariant(rawPriorVariant)
	if priorDigest.Valid {
		command.request.ExpectedPrior.SlotValueDigest = priorDigest.String
	}
	if err := validateSystemCommand(command); err != nil {
		return systemCommand{}, err
	}
	return command, nil
}

func validateSystemCommand(command systemCommand) error {
	if err := systemauthorityv1.ValidateCommandID(command.commandID); err != nil {
		return fmt.Errorf("%w: invalid command identity", ErrSystemOwnerCorrupt)
	}
	if err := validateSystemPrincipal(command.principalID, command.auditDigest); err != nil {
		return fmt.Errorf(
			"%w: invalid command principal binding: %v",
			ErrSystemOwnerCorrupt,
			err,
		)
	}
	if err := systemauthorityv1.ValidateCommandRequest(command.request); err != nil {
		return fmt.Errorf(
			"%w: invalid command request: %v",
			ErrSystemOwnerCorrupt,
			err,
		)
	}
	if err := validateSystemDigest(command.requestDigest); err != nil {
		return fmt.Errorf(
			"%w: invalid command request digest",
			ErrSystemOwnerCorrupt,
		)
	}
	computed, err := systemauthorityv1.CommandRequestDigest(command.request)
	if err != nil || computed != command.requestDigest {
		return fmt.Errorf(
			"%w: command request digest differs",
			ErrSystemOwnerCorrupt,
		)
	}
	return nil
}

func lockAndValidateSystemOutcomes(
	ctx context.Context,
	tx *sql.Tx,
	current systemCurrent,
	creationHistory systemHistory,
	commands []systemCommand,
) error {
	appliedCreates := 0
	for _, command := range commands {
		outcome, err := lockExactSystemOutcome(ctx, tx, command)
		if err != nil {
			return err
		}
		if outcome.result != "APPLIED" {
			continue
		}
		switch command.request.Operation {
		case systemauthorityv1.OperationCreate:
			appliedCreates++
			if command.commandID != creationHistory.commandID ||
				command.commandID != current.lastCommandID ||
				command.request.DesiredSlotValue != current.slot {
				return fmt.Errorf(
					"%w: APPLIED creation differs from current history",
					ErrSystemOwnerCorrupt,
				)
			}
		case systemauthorityv1.OperationRetire:
			return fmt.Errorf(
				"%w: ACTIVE slot has an APPLIED retirement outcome",
				ErrSystemOwnerCorrupt,
			)
		default:
			return fmt.Errorf(
				"%w: APPLIED command has an invalid operation",
				ErrSystemOwnerCorrupt,
			)
		}
	}
	if appliedCreates != 1 {
		return fmt.Errorf(
			"%w: ACTIVE slot must have exactly one APPLIED creation outcome",
			ErrSystemOwnerCorrupt,
		)
	}
	return nil
}

func lockExactSystemOutcome(
	ctx context.Context,
	tx *sql.Tx,
	command systemCommand,
) (systemOutcome, error) {
	var outcome systemOutcome
	var storedCommandID string
	var principalID []byte
	err := tx.QueryRowContext(ctx, `SELECT CommandId, RequestDigest, PrincipalId,
PrincipalAuditDigest, Result, OutcomeDigest
FROM _System_CatalogSystemLayoutBundleSlotCommandOutcome
WHERE CommandId = ? FOR UPDATE`,
		string(command.commandID),
	).Scan(
		&storedCommandID,
		&outcome.requestDigest,
		&principalID,
		&outcome.auditDigest,
		&outcome.result,
		&outcome.outcomeDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return systemOutcome{}, fmt.Errorf(
			"%w: command outcome is missing",
			ErrSystemOwnerCorrupt,
		)
	}
	if err != nil {
		return systemOutcome{}, err
	}
	outcome.commandID = systemauthorityv1.CommandID(storedCommandID)
	outcome.principalID = string(principalID)
	if err := validateSystemOutcome(outcome, command); err != nil {
		return systemOutcome{}, err
	}
	return outcome, nil
}

func validateSystemOutcome(
	outcome systemOutcome,
	command systemCommand,
) error {
	if outcome.commandID != command.commandID ||
		outcome.requestDigest != command.requestDigest ||
		outcome.principalID != command.principalID ||
		outcome.auditDigest != command.auditDigest {
		return fmt.Errorf(
			"%w: outcome binding differs from command",
			ErrSystemOwnerCorrupt,
		)
	}
	if !validSystemResult(outcome.result) {
		return fmt.Errorf(
			"%w: invalid outcome result %q",
			ErrSystemOwnerCorrupt,
			outcome.result,
		)
	}
	if err := validateSystemDigest(outcome.outcomeDigest); err != nil {
		return fmt.Errorf(
			"%w: invalid outcome digest",
			ErrSystemOwnerCorrupt,
		)
	}
	computed, err := systemOutcomeDigest(outcome)
	if err != nil || computed != outcome.outcomeDigest {
		return fmt.Errorf(
			"%w: outcome digest differs",
			ErrSystemOwnerCorrupt,
		)
	}
	return nil
}

func systemOutcomeDigest(outcome systemOutcome) (string, error) {
	return catalogsystemprotocol.OutcomeDigest(
		outcome.commandID,
		outcome.requestDigest,
		outcome.principalID,
		outcome.auditDigest,
		catalogsystemprotocol.Result(outcome.result),
	)
}

func validateSystemPrincipal(principalID, auditDigest string) error {
	return catalogsystemprotocol.ValidatePrincipal(principalID, auditDigest)
}

func validSystemResult(result string) bool {
	return catalogsystemprotocol.ValidateResult(
		catalogsystemprotocol.Result(result),
	) == nil
}

func parseSystemOrdinal(raw []byte) (uint8, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("%w: history ordinal is empty", ErrSystemOwnerCorrupt)
	}
	value, err := strconv.ParseUint(string(raw), 10, 8)
	if err != nil || strconv.FormatUint(value, 10) != string(raw) {
		return 0, fmt.Errorf(
			"%w: history ordinal is not canonical decimal",
			ErrSystemOwnerCorrupt,
		)
	}
	return uint8(value), nil
}

func validateSystemDigest(value string) error {
	return catalogsystemprotocol.ValidateDigest(value)
}