package catalogtenant

import (
	"context"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"github.com/go-sql-driver/mysql"
	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	tenantauthorityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/tenantlayoutbundleauthorityv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogadmissionfence"
)

func (a *Authority) startCommand(
	ctx context.Context,
	commandID tenantauthorityv1.CommandID,
) (*sql.Tx, uint64, uint64, *terminalCommand, error) {
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, 0, 0, nil, err
	}
	if _, err := catalogadmissionfence.LockExclusive(ctx, tx); err != nil {
		_ = tx.Rollback()
		return nil, 0, 0, nil, fmt.Errorf(
			"%w: publication admission fence: %v",
			ErrCorrupt,
			err,
		)
	}
	sequence, err := lockAuthorityFence(ctx, tx)
	if err != nil {
		_ = tx.Rollback()
		return nil, 0, 0, nil, err
	}
	existing, err := readTerminalCommand(ctx, tx, commandID)
	if err == nil {
		if existing.Sequence > sequence {
			_ = tx.Rollback()
			return nil, 0, 0, nil, fmt.Errorf(
				"%w: command sequence %d exceeds authority fence %d",
				ErrCorrupt,
				existing.Sequence,
				sequence,
			)
		}
		return tx, sequence, 0, &existing, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		_ = tx.Rollback()
		return nil, 0, 0, nil, err
	}
	if sequence == ^uint64(0) {
		_ = tx.Rollback()
		return nil, 0, 0, nil, fmt.Errorf("%w: command sequence exhausted", ErrConflict)
	}
	return tx, sequence, sequence + 1, nil, nil
}

func lockAuthorityFence(ctx context.Context, tx *sql.Tx) (uint64, error) {
	var storedKey string
	var rawSequence []byte
	err := tx.QueryRowContext(ctx, `SELECT FenceKey, CommandSequence
FROM _System_CatalogTenantAuthorityFence WHERE FenceKey = ? FOR UPDATE`,
		authorityFenceKey,
	).Scan(&storedKey, &rawSequence)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, fmt.Errorf("%w: authority fence is missing", ErrCorrupt)
	}
	if err != nil {
		return 0, err
	}
	if storedKey != authorityFenceKey {
		return 0, fmt.Errorf("%w: authority fence identity differs", ErrCorrupt)
	}
	return parseDurableUint64("authority CommandSequence", rawSequence)
}

func finishCommand(
	ctx context.Context,
	tx *sql.Tx,
	priorSequence uint64,
	command terminalCommand,
) error {
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogTenantAuthorityFence
SET CommandSequence = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE FenceKey = ? AND CommandSequence = ?`,
		uintText(command.Sequence),
		authorityFenceKey,
		uintText(priorSequence),
	)
	if err != nil {
		return err
	}
	if err := requireOne(result); err != nil {
		return err
	}
	if err := insertTerminalCommand(ctx, tx, command); err != nil {
		return err
	}
	return tx.Commit()
}

func insertTerminalCommand(
	ctx context.Context,
	tx *sql.Tx,
	command terminalCommand,
) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogTenantAuthorityCommand
(CommandId, CommandKind, RequestDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingValuePresent, ResultingValueDigest,
ObservedNamespacePresent, ObservedNamespaceDigest,
ObservedBindingPresent, ObservedBindingDigest,
CommandSequence, OutcomeDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		string(command.CommandID),
		string(command.Kind),
		command.RequestDigest,
		string(command.Result),
		command.ObservedPrior.Present,
		nullableReference(command.ObservedPrior),
		command.ResultingValue.Present,
		nullableReference(command.ResultingValue),
		command.ObservedNamespace.Present,
		nullableReference(command.ObservedNamespace),
		command.ObservedBinding.Present,
		nullableReference(command.ObservedBinding),
		uintText(command.Sequence),
		command.OutcomeDigest,
	)
	if isDuplicate(err) {
		return ErrConflict
	}
	if err != nil {
		return err
	}
	return requireOne(result)
}

func readTerminalCommand(
	ctx context.Context,
	tx *sql.Tx,
	commandID tenantauthorityv1.CommandID,
) (terminalCommand, error) {
	var command terminalCommand
	var storedCommandID, rawKind, rawResult string
	var priorPresent, resultingPresent, namespacePresent, bindingPresent bool
	var priorDigest, resultingDigest, namespaceDigest, bindingDigest sql.NullString
	var rawSequence []byte
	err := tx.QueryRowContext(ctx, `SELECT CommandId, CommandKind, RequestDigest, Result,
ObservedPriorPresent, ObservedPriorDigest, ResultingValuePresent, ResultingValueDigest,
ObservedNamespacePresent, ObservedNamespaceDigest,
ObservedBindingPresent, ObservedBindingDigest,
CommandSequence, OutcomeDigest
FROM _System_CatalogTenantAuthorityCommand WHERE CommandId = ? FOR UPDATE`,
		string(commandID),
	).Scan(
		&storedCommandID,
		&rawKind,
		&command.RequestDigest,
		&rawResult,
		&priorPresent,
		&priorDigest,
		&resultingPresent,
		&resultingDigest,
		&namespacePresent,
		&namespaceDigest,
		&bindingPresent,
		&bindingDigest,
		&rawSequence,
		&command.OutcomeDigest,
	)
	if err != nil {
		return terminalCommand{}, err
	}
	if storedCommandID != string(commandID) {
		return terminalCommand{}, fmt.Errorf(
			"%w: durable command identity differs from its lookup",
			ErrCorrupt,
		)
	}
	command.CommandID = tenantauthorityv1.CommandID(storedCommandID)
	command.Kind = commandKind(rawKind)
	command.Result = Result(rawResult)
	command.ObservedPrior, err = durableReference(
		"ObservedPrior",
		priorPresent,
		priorDigest,
	)
	if err != nil {
		return terminalCommand{}, err
	}
	command.ResultingValue, err = durableReference(
		"ResultingValue",
		resultingPresent,
		resultingDigest,
	)
	if err != nil {
		return terminalCommand{}, err
	}
	command.ObservedNamespace, err = durableReference(
		"ObservedNamespace",
		namespacePresent,
		namespaceDigest,
	)
	if err != nil {
		return terminalCommand{}, err
	}
	command.ObservedBinding, err = durableReference(
		"ObservedBinding",
		bindingPresent,
		bindingDigest,
	)
	if err != nil {
		return terminalCommand{}, err
	}
	command.Sequence, err = parseDurableUint64("command CommandSequence", rawSequence)
	if err != nil {
		return terminalCommand{}, err
	}
	if err := validateDurableTerminalCommand(command); err != nil {
		return terminalCommand{}, err
	}
	return command, nil
}

func lockNamespaceCurrent(
	ctx context.Context,
	tx *sql.Tx,
	tenantNamespaceID string,
) (namespaceCurrent, bool, error) {
	var current namespaceCurrent
	var storedNamespaceID []byte
	var rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT TenantNamespaceId, ValueVersion, State,
ValueDigest, LastCommandId
FROM _System_CatalogTenantNamespaceReservation
WHERE TenantNamespaceId = ? FOR UPDATE`,
		tenantNamespaceID,
	).Scan(
		&storedNamespaceID,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return namespaceCurrent{}, false, nil
	}
	if err != nil {
		return namespaceCurrent{}, false, err
	}
	current.Value.TenantNamespaceID = string(storedNamespaceID)
	current.Value.State = tenantauthorityv1.NamespaceState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.TenantNamespaceID != tenantNamespaceID {
		return namespaceCurrent{}, false, fmt.Errorf(
			"%w: namespace current identity differs from its lookup",
			ErrCorrupt,
		)
	}
	if err := validateNamespaceCurrent(current); err != nil {
		return namespaceCurrent{}, false, err
	}
	return current, true, nil
}

func validateNamespaceCurrent(current namespaceCurrent) error {
	computed, err := tenantauthorityv1.NamespaceValueDigest(current.Value)
	if err != nil {
		return fmt.Errorf("%w: invalid namespace current value: %v", ErrCorrupt, err)
	}
	if computed != current.Digest {
		return fmt.Errorf("%w: namespace current digest differs", ErrCorrupt)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.LastCommandID); err != nil {
		return fmt.Errorf("%w: invalid namespace current command: %v", ErrCorrupt, err)
	}
	return nil
}

func lockBindingCurrent(
	ctx context.Context,
	tx *sql.Tx,
	orgID string,
) (bindingCurrent, bool, error) {
	var current bindingCurrent
	var storedOrgID, storedNamespaceID []byte
	var rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding WHERE OrgId = ? FOR UPDATE`,
		orgID,
	).Scan(
		&storedOrgID,
		&storedNamespaceID,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return bindingCurrent{}, false, nil
	}
	if err != nil {
		return bindingCurrent{}, false, err
	}
	current.Value.OrgID = string(storedOrgID)
	current.Value.TenantNamespaceID = string(storedNamespaceID)
	current.Value.State = tenantauthorityv1.OrgBindingState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.OrgID != orgID {
		return bindingCurrent{}, false, fmt.Errorf(
			"%w: binding current identity differs from its lookup",
			ErrCorrupt,
		)
	}
	if err := validateBindingCurrent(current); err != nil {
		return bindingCurrent{}, false, err
	}
	return current, true, nil
}

func validateBindingCurrent(current bindingCurrent) error {
	computed, err := tenantauthorityv1.OrgBindingValueDigest(current.Value)
	if err != nil {
		return fmt.Errorf("%w: invalid binding current value: %v", ErrCorrupt, err)
	}
	if computed != current.Digest {
		return fmt.Errorf("%w: binding current digest differs", ErrCorrupt)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.LastCommandID); err != nil {
		return fmt.Errorf("%w: invalid binding current command: %v", ErrCorrupt, err)
	}
	return nil
}

func lockSlotCurrent(
	ctx context.Context,
	tx *sql.Tx,
	key tenantauthorityv1.BundleKey,
) (slotCurrent, bool, error) {
	var current slotCurrent
	var storedNamespaceID, storedLocalKey []byte
	var rawOwnerKind, rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT TenantNamespaceId, LocalComponentKey,
BundleKeyVersion, SlotOwnerKind, ValueVersion, State, ValueDigest, LastCommandId
FROM _System_CatalogTenantLayoutBundleSlot
WHERE TenantNamespaceId = ? AND LocalComponentKey = ? FOR UPDATE`,
		key.SlotOwnerNamespace,
		string(key.LocalComponentKey),
	).Scan(
		&storedNamespaceID,
		&storedLocalKey,
		&current.Value.BundleKey.BundleKeyVersion,
		&rawOwnerKind,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return slotCurrent{}, false, nil
	}
	if err != nil {
		return slotCurrent{}, false, err
	}
	current.Value.BundleKey.SlotOwnerKind = layoutbundlekeyv1.SlotOwnerKind(rawOwnerKind)
	current.Value.BundleKey.SlotOwnerNamespace = string(storedNamespaceID)
	current.Value.BundleKey.LocalComponentKey = layoutbundlekeyv1.LocalComponentKey(
		storedLocalKey,
	)
	current.Value.State = tenantauthorityv1.SlotState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.BundleKey != key {
		return slotCurrent{}, false, fmt.Errorf(
			"%w: slot current identity differs from its lookup",
			ErrCorrupt,
		)
	}
	if err := validateSlotCurrent(current); err != nil {
		return slotCurrent{}, false, err
	}
	return current, true, nil
}

func validateSlotCurrent(current slotCurrent) error {
	computed, err := tenantauthorityv1.SlotValueDigest(current.Value)
	if err != nil {
		return fmt.Errorf("%w: invalid slot current value: %v", ErrCorrupt, err)
	}
	if computed != current.Digest {
		return fmt.Errorf("%w: slot current digest differs", ErrCorrupt)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.LastCommandID); err != nil {
		return fmt.Errorf("%w: invalid slot current command: %v", ErrCorrupt, err)
	}
	return nil
}

func lockAnyActiveBinding(
	ctx context.Context,
	tx *sql.Tx,
	tenantNamespaceID string,
) (bool, error) {
	var current bindingCurrent
	var storedOrgID, storedNamespaceID []byte
	var rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding
WHERE TenantNamespaceId = ? AND State = 'ACTIVE'
ORDER BY OrgId LIMIT 1 FOR UPDATE`,
		tenantNamespaceID,
	).Scan(
		&storedOrgID,
		&storedNamespaceID,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	current.Value.OrgID = string(storedOrgID)
	current.Value.TenantNamespaceID = string(storedNamespaceID)
	current.Value.State = tenantauthorityv1.OrgBindingState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.TenantNamespaceID != tenantNamespaceID ||
		current.Value.State != tenantauthorityv1.OrgBindingStateActive {
		return false, fmt.Errorf("%w: active binding query returned another value", ErrCorrupt)
	}
	if err := validateBindingCurrent(current); err != nil {
		return false, err
	}
	return true, nil
}

func lockAnyOtherActiveBinding(
	ctx context.Context,
	tx *sql.Tx,
	tenantNamespaceID string,
	excludedOrgID string,
) (bool, error) {
	var current bindingCurrent
	var storedOrgID, storedNamespaceID []byte
	var rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT OrgId, TenantNamespaceId, ValueVersion,
State, ValueDigest, LastCommandId
FROM _System_CatalogTenantOrgNamespaceBinding
WHERE TenantNamespaceId = ? AND State = 'ACTIVE' AND OrgId <> ?
ORDER BY OrgId LIMIT 1 FOR UPDATE`,
		tenantNamespaceID,
		excludedOrgID,
	).Scan(
		&storedOrgID,
		&storedNamespaceID,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	current.Value.OrgID = string(storedOrgID)
	current.Value.TenantNamespaceID = string(storedNamespaceID)
	current.Value.State = tenantauthorityv1.OrgBindingState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.OrgID == excludedOrgID ||
		current.Value.TenantNamespaceID != tenantNamespaceID ||
		current.Value.State != tenantauthorityv1.OrgBindingStateActive {
		return false, fmt.Errorf(
			"%w: other active binding query returned another value",
			ErrCorrupt,
		)
	}
	if err := validateBindingCurrent(current); err != nil {
		return false, err
	}
	return true, nil
}

func lockAnyActiveSlot(
	ctx context.Context,
	tx *sql.Tx,
	tenantNamespaceID string,
) (bool, error) {
	var current slotCurrent
	var storedNamespaceID, storedLocalKey []byte
	var rawOwnerKind, rawState, storedCommandID string
	err := tx.QueryRowContext(ctx, `SELECT TenantNamespaceId, LocalComponentKey,
BundleKeyVersion, SlotOwnerKind, ValueVersion, State, ValueDigest, LastCommandId
FROM _System_CatalogTenantLayoutBundleSlot
WHERE TenantNamespaceId = ? AND State = 'ACTIVE'
ORDER BY LocalComponentKey LIMIT 1 FOR UPDATE`,
		tenantNamespaceID,
	).Scan(
		&storedNamespaceID,
		&storedLocalKey,
		&current.Value.BundleKey.BundleKeyVersion,
		&rawOwnerKind,
		&current.Value.ValueVersion,
		&rawState,
		&current.Digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	current.Value.BundleKey.SlotOwnerKind = layoutbundlekeyv1.SlotOwnerKind(rawOwnerKind)
	current.Value.BundleKey.SlotOwnerNamespace = string(storedNamespaceID)
	current.Value.BundleKey.LocalComponentKey = layoutbundlekeyv1.LocalComponentKey(
		storedLocalKey,
	)
	current.Value.State = tenantauthorityv1.SlotState(rawState)
	current.LastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.Value.BundleKey.SlotOwnerNamespace != tenantNamespaceID ||
		current.Value.State != tenantauthorityv1.SlotStateActive {
		return false, fmt.Errorf("%w: active slot query returned another value", ErrCorrupt)
	}
	if err := validateSlotCurrent(current); err != nil {
		return false, err
	}
	return true, nil
}

func referenceForNamespace(current namespaceCurrent, present bool) ValueReference {
	if !present {
		return ValueReference{}
	}
	return ValueReference{Present: true, Digest: current.Digest}
}

func referenceForBinding(current bindingCurrent, present bool) ValueReference {
	if !present {
		return ValueReference{}
	}
	return ValueReference{Present: true, Digest: current.Digest}
}

func referenceForSlot(current slotCurrent, present bool) ValueReference {
	if !present {
		return ValueReference{}
	}
	return ValueReference{Present: true, Digest: current.Digest}
}

func durableReference(
	label string,
	present bool,
	digest sql.NullString,
) (ValueReference, error) {
	if present != digest.Valid {
		return ValueReference{}, fmt.Errorf(
			"%w: %s presence and NULL state disagree",
			ErrCorrupt,
			label,
		)
	}
	reference := ValueReference{Present: present}
	if digest.Valid {
		reference.Digest = digest.String
	}
	if err := validateReference(reference); err != nil {
		return ValueReference{}, fmt.Errorf("%w: invalid %s: %v", ErrCorrupt, label, err)
	}
	return reference, nil
}

func validateReference(reference ValueReference) error {
	if !reference.Present {
		if reference.Digest != "" {
			return fmt.Errorf("absent reference must not carry a digest")
		}
		return nil
	}
	return validateDigest(reference.Digest)
}

func validateDigest(value string) error {
	const prefix = "sha256:"
	if len(value) != len(prefix)+64 || !strings.HasPrefix(value, prefix) {
		return fmt.Errorf("must use sha256:<64 lowercase hex>")
	}
	hexText := value[len(prefix):]
	if strings.ToLower(hexText) != hexText {
		return fmt.Errorf("digest hex must be lowercase")
	}
	if _, err := hex.DecodeString(hexText); err != nil {
		return fmt.Errorf("digest hex is invalid")
	}
	return nil
}

func parseDurableUint64(label string, raw []byte) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("%w: %s is empty", ErrCorrupt, label)
	}
	for _, character := range raw {
		if character < '0' || character > '9' {
			return 0, fmt.Errorf("%w: %s is not an unsigned decimal", ErrCorrupt, label)
		}
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%w: %s exceeds uint64: %v", ErrCorrupt, label, err)
	}
	if string(raw) != strconv.FormatUint(value, 10) {
		return 0, fmt.Errorf("%w: %s is not a canonical unsigned decimal", ErrCorrupt, label)
	}
	return value, nil
}

func uintText(value uint64) string {
	return strconv.FormatUint(value, 10)
}

func nullableReference(reference ValueReference) any {
	if !reference.Present {
		return nil
	}
	return reference.Digest
}

func requireOne(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return ErrConflict
	}
	return nil
}

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

func rollbackTx(tx *sql.Tx) {
	_ = tx.Rollback()
}