package catalogsystem

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

	"github.com/go-sql-driver/mysql"
	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/catalogsystemprotocol"
)

func insertDurableCommand(
	ctx context.Context,
	tx *sql.Tx,
	command durableCommand,
) error {
	request := command.Request
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogSystemLayoutBundleSlotCommand
(CommandId, PrincipalId, PrincipalAuditDigest, RequestDigest, Operation,
TargetBundleKeyVersion, TargetSlotOwnerKind, TargetSlotOwnerNamespace,
TargetLocalComponentKey, DesiredValueVersion, DesiredBundleKeyVersion,
DesiredSlotOwnerKind, DesiredSlotOwnerNamespace, DesiredLocalComponentKey,
DesiredState, ExpectedPriorVariant, ExpectedPriorDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		string(command.CommandID),
		command.Principal.PrincipalID,
		command.Principal.AuditDigest,
		command.RequestDigest,
		string(request.Operation),
		request.TargetBundleKey.BundleKeyVersion,
		string(request.TargetBundleKey.SlotOwnerKind),
		request.TargetBundleKey.SlotOwnerNamespace,
		string(request.TargetBundleKey.LocalComponentKey),
		request.DesiredSlotValue.ValueVersion,
		request.DesiredSlotValue.BundleKey.BundleKeyVersion,
		string(request.DesiredSlotValue.BundleKey.SlotOwnerKind),
		request.DesiredSlotValue.BundleKey.SlotOwnerNamespace,
		string(request.DesiredSlotValue.BundleKey.LocalComponentKey),
		string(request.DesiredSlotValue.State),
		string(request.ExpectedPrior.Variant),
		nullableDigest(request.ExpectedPrior.SlotValueDigest),
	)
	if err != nil {
		return err
	}
	return requireInsertedOne(result)
}

func readDurableCommand(
	ctx context.Context,
	tx *sql.Tx,
	commandID systemauthorityv1.CommandID,
) (durableCommand, error) {
	var command durableCommand
	var storedCommandID, rawOperation, rawTargetOwner, rawDesiredOwner string
	var principalID, targetNamespace, targetLocalKey []byte
	var desiredNamespace, desiredLocalKey []byte
	var rawDesiredState, rawPriorVariant string
	var priorDigest sql.NullString
	err := tx.QueryRowContext(ctx, `SELECT CommandId, PrincipalId,
PrincipalAuditDigest, RequestDigest, Operation,
TargetBundleKeyVersion, TargetSlotOwnerKind, TargetSlotOwnerNamespace,
TargetLocalComponentKey, DesiredValueVersion, DesiredBundleKeyVersion,
DesiredSlotOwnerKind, DesiredSlotOwnerNamespace, DesiredLocalComponentKey,
DesiredState, ExpectedPriorVariant, ExpectedPriorDigest
FROM _System_CatalogSystemLayoutBundleSlotCommand
WHERE CommandId = ? FOR UPDATE`,
		string(commandID),
	).Scan(
		&storedCommandID,
		&principalID,
		&command.Principal.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,
	)
	if err != nil {
		return durableCommand{}, err
	}
	command.CommandID = systemauthorityv1.CommandID(storedCommandID)
	command.Principal.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 command.CommandID != commandID {
		return durableCommand{}, fmt.Errorf(
			"%w: command identity differs from lookup",
			ErrCorrupt,
		)
	}
	if err := validateDurableCommand(command); err != nil {
		return durableCommand{}, err
	}
	return command, nil
}

func validateDurableCommand(command durableCommand) error {
	if err := systemauthorityv1.ValidateCommandID(command.CommandID); err != nil {
		return fmt.Errorf("%w: invalid CommandID: %v", ErrCorrupt, err)
	}
	if err := validatePrincipal(command.Principal); err != nil {
		return fmt.Errorf("%w: invalid principal binding: %v", ErrCorrupt, err)
	}
	if err := systemauthorityv1.ValidateCommandRequest(command.Request); err != nil {
		return fmt.Errorf("%w: invalid stored request: %v", ErrCorrupt, err)
	}
	if err := validateDigest(command.RequestDigest); err != nil {
		return fmt.Errorf("%w: invalid request digest: %v", ErrCorrupt, err)
	}
	computed, err := systemauthorityv1.CommandRequestDigest(command.Request)
	if err != nil {
		return fmt.Errorf("%w: cannot recompute request digest: %v", ErrCorrupt, err)
	}
	if computed != command.RequestDigest {
		return fmt.Errorf("%w: request digest differs", ErrCorrupt)
	}
	return nil
}

func insertDurableOutcome(
	ctx context.Context,
	tx *sql.Tx,
	outcome durableOutcome,
) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogSystemLayoutBundleSlotCommandOutcome
(CommandId, RequestDigest, PrincipalId, PrincipalAuditDigest, Result,
OutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		string(outcome.CommandID),
		outcome.RequestDigest,
		outcome.Principal.PrincipalID,
		outcome.Principal.AuditDigest,
		string(outcome.Result),
		outcome.OutcomeDigest,
	)
	if err != nil {
		return err
	}
	return requireInsertedOne(result)
}

func readDurableOutcome(
	ctx context.Context,
	tx *sql.Tx,
	command durableCommand,
) (durableOutcome, error) {
	var outcome durableOutcome
	var storedCommandID, rawResult 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.Principal.AuditDigest,
		&rawResult,
		&outcome.OutcomeDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return durableOutcome{}, fmt.Errorf("%w: command outcome is missing", ErrCorrupt)
	}
	if err != nil {
		return durableOutcome{}, err
	}
	outcome.CommandID = systemauthorityv1.CommandID(storedCommandID)
	outcome.Principal.PrincipalID = string(principalID)
	outcome.Result = Result(rawResult)
	if err := validateDurableOutcome(outcome, command); err != nil {
		return durableOutcome{}, err
	}
	return outcome, nil
}

func lockCurrentSlot(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) (currentSlot, bool, error) {
	var current currentSlot
	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.Value.BundleKey.BundleKeyVersion,
		&rawOwner,
		&storedNamespace,
		&storedLocalKey,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return currentSlot{}, false, nil
	}
	if err != nil {
		return currentSlot{}, false, err
	}
	current.Value.BundleKey.SlotOwnerKind = layoutbundlekeyv1.SlotOwnerKind(rawOwner)
	current.Value.BundleKey.SlotOwnerNamespace = string(storedNamespace)
	current.Value.BundleKey.LocalComponentKey =
		layoutbundlekeyv1.LocalComponentKey(storedLocalKey)
	current.Value.State = systemauthorityv1.SlotState(rawState)
	current.LastCommandID = systemauthorityv1.CommandID(storedCommandID)
	if current.Value.BundleKey != key {
		return currentSlot{}, true, fmt.Errorf(
			"%w: current slot identity differs from lookup",
			ErrCorrupt,
		)
	}
	if err := validateCurrentSlot(current); err != nil {
		return currentSlot{}, true, err
	}
	return current, true, nil
}

func validateCurrentSlot(current currentSlot) error {
	if err := systemauthorityv1.ValidateSlotValue(current.Value); err != nil {
		return fmt.Errorf("%w: invalid current value: %v", ErrCorrupt, err)
	}
	if err := validateDigest(current.Digest); err != nil {
		return fmt.Errorf("%w: invalid current digest: %v", ErrCorrupt, err)
	}
	computed, err := systemauthorityv1.SlotValueDigest(current.Value)
	if err != nil || computed != current.Digest {
		return fmt.Errorf("%w: current value digest differs", ErrCorrupt)
	}
	if err := systemauthorityv1.ValidateCommandID(current.LastCommandID); err != nil {
		return fmt.Errorf("%w: invalid current command identity: %v", ErrCorrupt, err)
	}
	return nil
}

func lockAnyHistory(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) (bool, error) {
	var rawOrdinal []byte
	err := tx.QueryRowContext(ctx, `SELECT TransitionOrdinal
FROM _System_CatalogSystemLayoutBundleSlotHistory
WHERE BundleKeyVersion = ? AND SlotOwnerKind = ?
AND SlotOwnerNamespace = ? AND LocalComponentKey = ?
ORDER BY TransitionOrdinal LIMIT 1 FOR UPDATE`,
		key.BundleKeyVersion,
		string(key.SlotOwnerKind),
		key.SlotOwnerNamespace,
		string(key.LocalComponentKey),
	).Scan(&rawOrdinal)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return true, nil
}

func lockHistory(
	ctx context.Context,
	tx *sql.Tx,
	key systemauthorityv1.BundleKey,
) ([]historyEntry, 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
	}
	defer rows.Close()
	var history []historyEntry
	for rows.Next() {
		var entry historyEntry
		var rawOwner, rawState, storedCommandID string
		var storedNamespace, storedLocalKey, rawOrdinal []byte
		if err := rows.Scan(
			&entry.Value.BundleKey.BundleKeyVersion,
			&rawOwner,
			&storedNamespace,
			&storedLocalKey,
			&rawOrdinal,
			&storedCommandID,
			&entry.Value.ValueVersion,
			&rawState,
			&entry.Digest,
			&entry.PriorDigest,
		); err != nil {
			return nil, err
		}
		entry.Value.BundleKey.SlotOwnerKind =
			layoutbundlekeyv1.SlotOwnerKind(rawOwner)
		entry.Value.BundleKey.SlotOwnerNamespace = string(storedNamespace)
		entry.Value.BundleKey.LocalComponentKey =
			layoutbundlekeyv1.LocalComponentKey(storedLocalKey)
		entry.Value.State = systemauthorityv1.SlotState(rawState)
		entry.CommandID = systemauthorityv1.CommandID(storedCommandID)
		ordinal, err := parseOrdinal(rawOrdinal)
		if err != nil {
			return nil, err
		}
		entry.TransitionOrdinal = ordinal
		if entry.Value.BundleKey != key {
			return nil, fmt.Errorf("%w: history identity differs from lookup", ErrCorrupt)
		}
		if err := validateHistoryEntry(entry); err != nil {
			return nil, err
		}
		history = append(history, entry)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return history, nil
}

func validateHistoryEntry(entry historyEntry) error {
	if err := systemauthorityv1.ValidateCommandID(entry.CommandID); err != nil {
		return fmt.Errorf("%w: invalid history command: %v", ErrCorrupt, err)
	}
	if err := systemauthorityv1.ValidateSlotValue(entry.Value); err != nil {
		return fmt.Errorf("%w: invalid history value: %v", ErrCorrupt, err)
	}
	if err := validateDigest(entry.Digest); err != nil {
		return fmt.Errorf("%w: invalid history digest: %v", ErrCorrupt, err)
	}
	computed, err := systemauthorityv1.SlotValueDigest(entry.Value)
	if err != nil || computed != entry.Digest {
		return fmt.Errorf("%w: history value digest differs", ErrCorrupt)
	}
	switch entry.TransitionOrdinal {
	case 1:
		if entry.Value.State != systemauthorityv1.SlotStateActive ||
			entry.PriorDigest.Valid {
			return fmt.Errorf("%w: creation history is not ACTIVE from ABSENT", ErrCorrupt)
		}
	case 2:
		if entry.Value.State != systemauthorityv1.SlotStateTombstone ||
			!entry.PriorDigest.Valid {
			return fmt.Errorf("%w: retirement history is not TOMBSTONE from EXACT", ErrCorrupt)
		}
		if err := validateDigest(entry.PriorDigest.String); err != nil {
			return fmt.Errorf("%w: invalid retirement prior digest: %v", ErrCorrupt, err)
		}
	default:
		return fmt.Errorf("%w: invalid history ordinal %d", ErrCorrupt, entry.TransitionOrdinal)
	}
	return nil
}

func validateActiveLineage(
	ctx context.Context,
	tx *sql.Tx,
	current currentSlot,
	history []historyEntry,
) error {
	if current.Value.State != systemauthorityv1.SlotStateActive ||
		len(history) != 1 {
		return fmt.Errorf("%w: ACTIVE current must have exactly one history entry", ErrCorrupt)
	}
	creation := history[0]
	if creation.TransitionOrdinal != 1 ||
		creation.CommandID != current.LastCommandID ||
		creation.Value != current.Value ||
		creation.Digest != current.Digest {
		return fmt.Errorf("%w: ACTIVE current and creation history differ", ErrCorrupt)
	}
	command, err := readDurableCommand(ctx, tx, creation.CommandID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return fmt.Errorf("%w: creation command is missing", ErrCorrupt)
		}
		return err
	}
	if command.Request.Operation != systemauthorityv1.OperationCreate ||
		command.Request.TargetBundleKey != current.Value.BundleKey ||
		command.Request.DesiredSlotValue != current.Value ||
		command.Request.ExpectedPrior.Variant != systemauthorityv1.ExpectedPriorAbsent ||
		command.Request.ExpectedPrior.SlotValueDigest != "" {
		return fmt.Errorf("%w: creation command and history differ", ErrCorrupt)
	}
	outcome, err := readDurableOutcome(ctx, tx, command)
	if err != nil {
		return err
	}
	if outcome.Result != ResultApplied {
		return fmt.Errorf("%w: creation history has no APPLIED outcome", ErrCorrupt)
	}
	return nil
}

func parseOrdinal(raw []byte) (uint8, error) {
	value, err := strconv.ParseUint(string(raw), 10, 8)
	if err != nil {
		return 0, fmt.Errorf("%w: invalid history ordinal: %v", ErrCorrupt, err)
	}
	return uint8(value), nil
}

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

func nullableDigest(value string) any {
	if value == "" {
		return nil
	}
	return value
}

func requireInsertedOne(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return fmt.Errorf(
			"catalog system slot authority: durable insert affected %d rows",
			affected,
		)
	}
	return nil
}

func isDuplicate(err error) bool {
	var mysqlError *mysql.MySQLError
	return errors.As(err, &mysqlError) && mysqlError.Number == 1062
}