package catalogverificationtx

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

	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	tenantauthorityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/tenantlayoutbundleauthorityv1"
)

var (
	ErrTenantOwnerNotFound = errors.New(
		"catalog verification transaction: tenant owner fact not found",
	)
	ErrTenantOwnerConflict = errors.New(
		"catalog verification transaction: tenant owner fact conflict",
	)
	ErrTenantOwnerCorrupt = errors.New(
		"catalog verification transaction: tenant owner durable state is corrupt",
	)
)

// TenantOwnerFact is the complete locked ACTIVE namespace, source-org
// binding, and tenant slot triple. The value digests remain internal inputs to
// the process-local capability and are never caller-authored proof.
type TenantOwnerFact struct {
	SlotCreationCommandSequence uint64
	NamespaceReservation        tenantauthorityv1.TenantNamespaceReservationV1
	NamespaceReservationDigest  string
	OrgBinding                  tenantauthorityv1.TenantOrgNamespaceBindingV1
	OrgBindingDigest            string
	Slot                        tenantauthorityv1.TenantLayoutBundleSlotV1
	SlotDigest                  string
}

type tenantNamespaceCurrent struct {
	value         tenantauthorityv1.TenantNamespaceReservationV1
	digest        string
	lastCommandID tenantauthorityv1.CommandID
}

type tenantBindingCurrent struct {
	value         tenantauthorityv1.TenantOrgNamespaceBindingV1
	digest        string
	lastCommandID tenantauthorityv1.CommandID
}

type tenantSlotCurrent struct {
	value         tenantauthorityv1.TenantLayoutBundleSlotV1
	digest        string
	lastCommandID tenantauthorityv1.CommandID
}

// VerifyTenantOwner holds the exact namespace, source-org binding, and slot
// under FOR UPDATE locks in the retained source transaction. Those current
// rows serialize the verifier with every command that could retire or replace
// the triple, while unrelated tenant namespaces remain independent.
func VerifyTenantOwner(
	ctx context.Context,
	token *Token,
	db *sql.DB,
	request TenantOwnerRequest,
) (TenantOwnerFact, error) {
	if err := validateTenantOwnerRequest(request); err != nil {
		return TenantOwnerFact{}, err
	}
	tx, err := token.acquire(db)
	if err != nil {
		return TenantOwnerFact{}, err
	}
	defer token.release()

	namespace, err := lockExactActiveTenantNamespace(
		ctx,
		tx,
		request.BundleKey.SlotOwnerNamespace,
	)
	if err != nil {
		return TenantOwnerFact{}, err
	}
	binding, err := lockExactActiveTenantBinding(ctx, tx, request)
	if err != nil {
		return TenantOwnerFact{}, err
	}
	slot, err := lockExactActiveTenantSlot(ctx, tx, request.BundleKey)
	if err != nil {
		return TenantOwnerFact{}, err
	}
	sequence, err := verifyTenantOwnerLineage(
		ctx,
		tx,
		request,
		namespace,
		binding,
		slot,
	)
	if err != nil {
		return TenantOwnerFact{}, err
	}

	return TenantOwnerFact{
		SlotCreationCommandSequence: sequence,
		NamespaceReservation:        namespace.value,
		NamespaceReservationDigest:  namespace.digest,
		OrgBinding:                  binding.value,
		OrgBindingDigest:            binding.digest,
		Slot:                        slot.value,
		SlotDigest:                  slot.digest,
	}, nil
}

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

func validateTenantOwnerRequest(request TenantOwnerRequest) error {
	if err := tenantauthorityv1.ValidateCatalogOrgID(request.OrgID); err != nil {
		return fmt.Errorf("%w: OrgID: %v", ErrTenantOwnerConflict, err)
	}
	if err := layoutbundlekeyv1.ValidateBundleKey(request.BundleKey); err != nil {
		return fmt.Errorf("%w: BundleKey: %v", ErrTenantOwnerConflict, err)
	}
	if request.BundleKey.SlotOwnerKind != layoutbundlekeyv1.SlotOwnerKindTenant {
		return fmt.Errorf("%w: BundleKey is not tenant-owned", ErrTenantOwnerConflict)
	}
	return nil
}

func lockExactActiveTenantNamespace(
	ctx context.Context,
	tx *sql.Tx,
	tenantNamespaceID string,
) (tenantNamespaceCurrent, error) {
	var current tenantNamespaceCurrent
	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 tenantNamespaceCurrent{}, ErrTenantOwnerNotFound
	}
	if err != nil {
		return tenantNamespaceCurrent{}, err
	}
	current.value.TenantNamespaceID = string(storedNamespaceID)
	current.value.State = tenantauthorityv1.NamespaceState(rawState)
	current.lastCommandID = tenantauthorityv1.CommandID(storedCommandID)
	if current.value.TenantNamespaceID != tenantNamespaceID {
		return tenantNamespaceCurrent{}, fmt.Errorf(
			"%w: tenant namespace durable identity differs from its exact lookup",
			ErrTenantOwnerCorrupt,
		)
	}
	computed, err := tenantauthorityv1.NamespaceValueDigest(current.value)
	if err != nil || computed != current.digest {
		return tenantNamespaceCurrent{}, fmt.Errorf(
			"%w: tenant namespace value or digest is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.lastCommandID); err != nil {
		return tenantNamespaceCurrent{}, fmt.Errorf(
			"%w: tenant namespace LastCommandId is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if current.value.State != tenantauthorityv1.NamespaceStateActive {
		return tenantNamespaceCurrent{}, fmt.Errorf(
			"%w: tenant namespace is not ACTIVE",
			ErrTenantOwnerConflict,
		)
	}
	return current, nil
}

func lockExactActiveTenantBinding(
	ctx context.Context,
	tx *sql.Tx,
	request TenantOwnerRequest,
) (tenantBindingCurrent, error) {
	var current tenantBindingCurrent
	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`,
		request.OrgID,
	).Scan(
		&storedOrgID,
		&storedNamespaceID,
		&current.value.ValueVersion,
		&rawState,
		&current.digest,
		&storedCommandID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return tenantBindingCurrent{}, ErrTenantOwnerNotFound
	}
	if err != nil {
		return tenantBindingCurrent{}, 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 != request.OrgID {
		return tenantBindingCurrent{}, fmt.Errorf(
			"%w: tenant binding durable identity differs from its exact lookup",
			ErrTenantOwnerCorrupt,
		)
	}
	computed, err := tenantauthorityv1.OrgBindingValueDigest(current.value)
	if err != nil || computed != current.digest {
		return tenantBindingCurrent{}, fmt.Errorf(
			"%w: tenant binding value or digest is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.lastCommandID); err != nil {
		return tenantBindingCurrent{}, fmt.Errorf(
			"%w: tenant binding LastCommandId is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if current.value.State != tenantauthorityv1.OrgBindingStateActive ||
		current.value.TenantNamespaceID != request.BundleKey.SlotOwnerNamespace {
		return tenantBindingCurrent{}, fmt.Errorf(
			"%w: source OrgID is not actively bound to the tenant namespace",
			ErrTenantOwnerConflict,
		)
	}
	return current, nil
}

func lockExactActiveTenantSlot(
	ctx context.Context,
	tx *sql.Tx,
	key layoutbundlekeyv1.BundleKey,
) (tenantSlotCurrent, error) {
	var current tenantSlotCurrent
	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 tenantSlotCurrent{}, ErrTenantOwnerNotFound
	}
	if err != nil {
		return tenantSlotCurrent{}, 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)
	expected := tenantBundleKey(key)
	if current.value.BundleKey != expected {
		return tenantSlotCurrent{}, fmt.Errorf(
			"%w: tenant slot durable identity differs from its exact lookup",
			ErrTenantOwnerCorrupt,
		)
	}
	computed, err := tenantauthorityv1.SlotValueDigest(current.value)
	if err != nil || computed != current.digest {
		return tenantSlotCurrent{}, fmt.Errorf(
			"%w: tenant slot value or digest is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if err := tenantauthorityv1.ValidateCommandID(current.lastCommandID); err != nil {
		return tenantSlotCurrent{}, fmt.Errorf(
			"%w: tenant slot LastCommandId is invalid",
			ErrTenantOwnerCorrupt,
		)
	}
	if current.value.State != tenantauthorityv1.SlotStateActive {
		return tenantSlotCurrent{}, fmt.Errorf(
			"%w: tenant slot is not ACTIVE",
			ErrTenantOwnerConflict,
		)
	}
	return current, nil
}

func tenantBundleKey(key layoutbundlekeyv1.BundleKey) tenantauthorityv1.BundleKey {
	return tenantauthorityv1.BundleKey{
		BundleKeyVersion:   layoutbundlekeyv1.BundleKeyVersion,
		SlotOwnerKind:      key.SlotOwnerKind,
		SlotOwnerNamespace: key.SlotOwnerNamespace,
		LocalComponentKey:  key.LocalComponentKey,
	}
}

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