package catalogpublication

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

	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	layoutbundleownerverificationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundleownerverificationv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogsystemfence"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogverificationtx"
)

var errSourceVerificationHasNoJSON = catalogverificationtx.ErrCapabilityHasNoJSON

// CandidateSourceVerificationLease retains the exact Catalog source locks and
// writer reservation used by concrete owner verifiers. It is process-local,
// caller-unconstructible, and has no wire form.
type CandidateSourceVerificationLease struct {
	authority    *sourceAuthorityIdentity
	db           *sql.DB
	verification *catalogverificationtx.Lease
	record       candidateSourceRecord
	requiredSet  layoutbundlekeyv1.RequiredBundleKeySet
	required     map[layoutbundlekeyv1.BundleKey]int
	selections   map[string]SourceAppSelection
	ownerMu      sync.Mutex
	ownerClosure *CandidateSourceOwnerVerificationCapability
}

// CandidateSourceOwnerVerification is the complete, exact required-key owner
// closure for one locked Catalog source. The receipt values are immutable
// content; only Capability proves that their source and owner fences remain
// live.
type CandidateSourceOwnerVerification struct {
	ReceiptSet       layoutbundleownerverificationv1.ReceiptSet
	ReceiptSetDigest string
	Capability       *CandidateSourceOwnerVerificationCapability
}

// CandidateSourceOwnerVerificationCapability is an opaque process-local proof
// of complete source/owner closure. It is deliberately not an adoption or
// publication capability: the future routing-security fence must be acquired
// in the prescribed lock order before any Catalog selection can occur.
type CandidateSourceOwnerVerificationCapability struct {
	authority        *sourceAuthorityIdentity
	db               *sql.DB
	source           *CandidateSourceVerificationLease
	receiptSet       layoutbundleownerverificationv1.ReceiptSet
	receiptSetDigest string
}

// CandidateSourceAppVerification is one exact required App key claimed from a
// locked source lease. Only catalogapp can turn it into verified evidence.
type CandidateSourceAppVerification = catalogverificationtx.AppClaim

// CandidateSourceTenantVerification is one exact required tenant key claimed
// from a locked source lease. Only catalogtenant can turn it into verified
// evidence.
type CandidateSourceTenantVerification = catalogverificationtx.TenantClaim

// CandidateSourceSystemVerification is one exact required system key claimed
// from a locked source lease. Only the internal Catalog system verifier can
// turn it into verified evidence.
type CandidateSourceSystemVerification = catalogverificationtx.SystemClaim

// BeginCandidateSourceVerification reacquires and retains the org writer,
// active pointer, exact source row, complete required set, and
// live writer reservation for concrete owner verification.
func (a *Authority) BeginCandidateSourceVerification(
	ctx context.Context,
	handle *CatalogCandidateSourceHandle,
) (_ *CandidateSourceVerificationLease, returnErr error) {
	if err := a.validateHandleIdentity(handle); err != nil {
		return nil, err
	}
	conn, tx, mode, err := beginRetainedSourceTransaction(ctx, a.db)
	if err != nil {
		return nil, err
	}
	token, err := catalogverificationtx.NewOwned(a.db, conn, tx, mode)
	if err != nil {
		rollbackErr := tx.Rollback()
		restoreErr := mode.Restore(context.Background(), conn)
		return nil, errors.Join(err, rollbackErr, restoreErr, conn.Close())
	}
	retained := false
	defer func() {
		if !retained {
			returnErr = errors.Join(returnErr, token.Close())
		}
	}()

	writer, err := lockCandidateSourcePrefix(ctx, tx, handle.orgID)
	if err != nil {
		return nil, err
	}
	state, baseline, err := lockCandidateSourceActiveBaseline(ctx, tx, handle.orgID)
	if err != nil {
		return nil, err
	}
	record, err := readCandidateSourceDB(
		ctx,
		tx,
		handle.orgID,
		handle.writerCommandID,
		true,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrSourceNotFound
	}
	if err != nil {
		return nil, err
	}
	if err := rejectSourceCommandKindCollisionLocked(
		ctx,
		tx,
		handle.orgID,
		handle.writerCommandID,
		true,
	); err != nil {
		return nil, err
	}
	if err := validateHandleRecord(handle, record); err != nil {
		return nil, err
	}
	if err := validateCandidateSourceLifecycle(record, writer, state, baseline); err != nil {
		return nil, err
	}

	canonicalRequired, err := layoutbundlekeyv1.CanonicalizeRequiredBundleKeySet(
		record.RequiredBundleKeySet,
	)
	if err != nil {
		return nil, fmt.Errorf("%w: durable required set: %v", ErrSourceCorrupt, err)
	}
	if err := token.BindValidatedSource(
		record.CatalogSnapshotBinding,
		record.OrgID,
		record.WriterCommandID,
		record.WriterReservationGeneration,
		record.RequestDigest,
		record.RequiredBundleKeySetDigest,
	); err != nil {
		return nil, err
	}
	verification, err := catalogverificationtx.NewLease(token, canonicalRequired)
	if err != nil {
		return nil, err
	}
	lease := &CandidateSourceVerificationLease{
		authority:    a.identity,
		db:           a.db,
		verification: verification,
		record:       cloneCandidateSourceRecordForVerification(record),
		requiredSet:  cloneRequiredSet(canonicalRequired),
		required:     make(map[layoutbundlekeyv1.BundleKey]int, len(canonicalRequired.BundleKeys)),
		selections:   make(map[string]SourceAppSelection, len(record.SourceAppSelectionSet.Entries)),
	}
	for index, key := range canonicalRequired.BundleKeys {
		lease.required[key] = index
	}
	for _, selection := range record.SourceAppSelectionSet.Entries {
		lease.selections[selection.AppID] = selection
	}
	retained = true
	return lease, nil
}

// CompleteOwnerVerification proves that every exact required BundleKey has one
// fresh receipt sealed by its fixed-purpose owner verifier under this same
// retained transaction. It accepts no caller receipt, digest, callback, or
// generic proof adapter.
func (lease *CandidateSourceVerificationLease) CompleteOwnerVerification() (
	CandidateSourceOwnerVerification,
	error,
) {
	if lease == nil || lease.verification == nil || !lease.Active() {
		return CandidateSourceOwnerVerification{}, ErrSourceConflict
	}
	lease.ownerMu.Lock()
	defer lease.ownerMu.Unlock()
	if lease.ownerClosure != nil {
		if err := lease.ownerClosure.Validate(); err != nil {
			return CandidateSourceOwnerVerification{}, err
		}
		return CandidateSourceOwnerVerification{
			ReceiptSet:       cloneOwnerReceiptSet(lease.ownerClosure.receiptSet),
			ReceiptSetDigest: lease.ownerClosure.receiptSetDigest,
			Capability:       lease.ownerClosure,
		}, nil
	}
	set, digest, err := lease.verification.SealReceiptSet()
	if errors.Is(err, catalogverificationtx.ErrClaimOrder) {
		return CandidateSourceOwnerVerification{}, fmt.Errorf(
			"%w: owner verification does not cover the complete required BundleKey set",
			ErrSourceNotReady,
		)
	}
	if err != nil {
		closeErr := lease.Close()
		return CandidateSourceOwnerVerification{}, errors.Join(
			fmt.Errorf("%w: complete owner verification: %v", ErrSourceCorrupt, err),
			closeErr,
		)
	}
	capability := &CandidateSourceOwnerVerificationCapability{
		authority:        lease.authority,
		db:               lease.db,
		source:           lease,
		receiptSet:       cloneOwnerReceiptSet(set),
		receiptSetDigest: digest,
	}
	if err := capability.Validate(); err != nil {
		closeErr := lease.Close()
		return CandidateSourceOwnerVerification{}, errors.Join(err, closeErr)
	}
	lease.ownerClosure = capability
	return CandidateSourceOwnerVerification{
		ReceiptSet:       cloneOwnerReceiptSet(set),
		ReceiptSetDigest: digest,
		Capability:       capability,
	}, nil
}

// Validate re-proves that the exact sealed receipt set and its retained source
// transaction remain live. Values alone never grant authority.
func (capability *CandidateSourceOwnerVerificationCapability) Validate() error {
	if capability == nil ||
		capability.authority == nil ||
		capability.db == nil ||
		capability.source == nil ||
		capability.source.authority != capability.authority ||
		capability.source.db != capability.db ||
		!capability.source.Active() {
		return ErrSourceConflict
	}
	set, digest, err := capability.source.verification.ValidateSealedReceiptSet()
	if err != nil {
		return fmt.Errorf("%w: sealed owner verification: %v", ErrSourceConflict, err)
	}
	if err := layoutbundleownerverificationv1.ValidateReceiptSet(
		set,
		capability.source.requiredSet,
	); err != nil {
		return fmt.Errorf("%w: sealed owner receipt set: %v", ErrSourceCorrupt, err)
	}
	if digest != capability.receiptSetDigest ||
		!equalOwnerReceiptSets(set, capability.receiptSet) {
		return fmt.Errorf("%w: sealed owner verification differs", ErrSourceCorrupt)
	}
	return nil
}

// ReceiptSet returns a defensive copy only while the closure capability is
// live. The value is non-authorizing.
func (capability *CandidateSourceOwnerVerificationCapability) ReceiptSet() (
	layoutbundleownerverificationv1.ReceiptSet,
	error,
) {
	if err := capability.Validate(); err != nil {
		return layoutbundleownerverificationv1.ReceiptSet{}, err
	}
	return cloneOwnerReceiptSet(capability.receiptSet), nil
}

// ReceiptSetDigest returns the immutable content identity only while the
// closure capability is live. The digest is not authority by itself.
func (capability *CandidateSourceOwnerVerificationCapability) ReceiptSetDigest() (
	string,
	error,
) {
	if err := capability.Validate(); err != nil {
		return "", err
	}
	return capability.receiptSetDigest, nil
}

// MarshalJSON rejects persistence of process-scoped owner closure authority.
func (*CandidateSourceOwnerVerificationCapability) MarshalJSON() ([]byte, error) {
	return nil, errSourceVerificationHasNoJSON
}

// UnmarshalJSON rejects caller construction of owner closure authority.
func (*CandidateSourceOwnerVerificationCapability) UnmarshalJSON([]byte) error {
	return errSourceVerificationHasNoJSON
}

func cloneOwnerReceiptSet(
	set layoutbundleownerverificationv1.ReceiptSet,
) layoutbundleownerverificationv1.ReceiptSet {
	set.Receipts = append([]layoutbundleownerverificationv1.ReceiptEntry{}, set.Receipts...)
	return set
}

func equalOwnerReceiptSets(
	left layoutbundleownerverificationv1.ReceiptSet,
	right layoutbundleownerverificationv1.ReceiptSet,
) bool {
	if left.ReceiptSetVersion != right.ReceiptSetVersion ||
		left.ReceiptVersion != right.ReceiptVersion ||
		left.OrgID != right.OrgID ||
		left.WriterCommandID != right.WriterCommandID ||
		left.WriterReservationGeneration != right.WriterReservationGeneration ||
		left.SourceRequestDigest != right.SourceRequestDigest ||
		left.RequiredBundleKeySetDigest != right.RequiredBundleKeySetDigest ||
		(left.Receipts == nil) != (right.Receipts == nil) ||
		len(left.Receipts) != len(right.Receipts) {
		return false
	}
	for index := range left.Receipts {
		if left.Receipts[index] != right.Receipts[index] {
			return false
		}
	}
	return true
}

func beginRetainedSourceTransaction(
	ctx context.Context,
	db *sql.DB,
) (
	*sql.Conn,
	*sql.Tx,
	*catalogsystemfence.PessimisticModeGuard,
	error,
) {
	if db == nil {
		return nil, nil, nil, errors.New(
			"begin retained Catalog source transaction: database is required",
		)
	}
	conn, err := db.Conn(ctx)
	if err != nil {
		return nil, nil, nil, err
	}
	mode, err := catalogsystemfence.ForcePessimistic(ctx, conn)
	if err != nil {
		return nil, nil, nil, errors.Join(err, conn.Close())
	}

	// database/sql binds a transaction's lifetime to the BeginTx context. The
	// verification lease, not request cancellation, owns this retained
	// transaction, so its begin context deliberately has no cancellation or
	// deadline while all individual queries still use the caller's context.
	tx, err := conn.BeginTx(context.WithoutCancel(ctx), nil)
	if err != nil {
		return nil, nil, nil, errors.Join(
			err,
			catalogsystemfence.Discard(conn),
			conn.Close(),
		)
	}
	return conn, tx, mode, nil
}

// BeginAppBundleKeyVerification claims one exact App-owned required key in
// canonical BundleKey order.
func (lease *CandidateSourceVerificationLease) BeginAppBundleKeyVerification(
	key layoutbundlekeyv1.BundleKey,
) (*CandidateSourceAppVerification, error) {
	if lease == nil {
		return nil, ErrSourceConflict
	}
	if err := layoutbundlekeyv1.ValidateBundleKey(key); err != nil {
		return nil, fmt.Errorf("%w: invalid BundleKey: %v", ErrSourceConflict, err)
	}
	if key.SlotOwnerKind != layoutbundlekeyv1.SlotOwnerKindApp {
		return nil, fmt.Errorf("%w: BundleKey is not App-owned", ErrSourceConflict)
	}

	if !lease.Active() {
		return nil, ErrSourceConflict
	}
	index, required := lease.required[key]
	if !required {
		return nil, fmt.Errorf("%w: BundleKey is not required by the source", ErrSourceConflict)
	}
	selection, selected := lease.selections[key.SlotOwnerNamespace]
	if !selected ||
		selection.AppID != key.SlotOwnerNamespace ||
		selection.SelectionKind != SourceAppSelectionKindActive ||
		selection.AppLifecycleCandidateCommandID != "" {
		return nil, fmt.Errorf(
			"%w: required App key has no exact ACTIVE source selection",
			ErrSourceConflict,
		)
	}
	claim, err := lease.verification.BeginAppClaim(
		index,
		catalogverificationtx.AppSourceFacts{
			SourceCatalogSnapshot:       lease.record.CatalogSnapshotBinding,
			OrgID:                       lease.record.OrgID,
			WriterCommandID:             lease.record.WriterCommandID,
			WriterReservationGeneration: lease.record.WriterReservationGeneration,
			SourceRequestDigest:         lease.record.RequestDigest,
			RequiredBundleKeySetDigest:  lease.record.RequiredBundleKeySetDigest,
			BundleKey:                   key,
			AppID:                       selection.AppID,
			AppInstallationID:           selection.AppInstallationID,
			AppReleaseDigest:            selection.AppReleaseDigest,
		},
		catalogverificationtx.AppOwnerRequest{
			OrgID:             lease.record.OrgID,
			AppID:             selection.AppID,
			AppInstallationID: selection.AppInstallationID,
			AppReleaseDigest:  selection.AppReleaseDigest,
			LocalComponentKey: key.LocalComponentKey,
		},
	)
	if errors.Is(err, catalogverificationtx.ErrClaimOrder) {
		return nil, fmt.Errorf(
			"%w: owner fences must be acquired in canonical BundleKey order",
			ErrSourceConflict,
		)
	}
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	return claim, nil
}

// BeginTenantBundleKeyVerification claims one exact tenant-owned required key
// in canonical BundleKey order.
func (lease *CandidateSourceVerificationLease) BeginTenantBundleKeyVerification(
	key layoutbundlekeyv1.BundleKey,
) (*CandidateSourceTenantVerification, error) {
	if lease == nil {
		return nil, ErrSourceConflict
	}
	if err := layoutbundlekeyv1.ValidateBundleKey(key); err != nil {
		return nil, fmt.Errorf("%w: invalid BundleKey: %v", ErrSourceConflict, err)
	}
	if key.SlotOwnerKind != layoutbundlekeyv1.SlotOwnerKindTenant {
		return nil, fmt.Errorf("%w: BundleKey is not tenant-owned", ErrSourceConflict)
	}

	if !lease.Active() {
		return nil, ErrSourceConflict
	}
	index, required := lease.required[key]
	if !required {
		return nil, fmt.Errorf("%w: BundleKey is not required by the source", ErrSourceConflict)
	}
	claim, err := lease.verification.BeginTenantClaim(
		index,
		catalogverificationtx.TenantSourceFacts{
			SourceCatalogSnapshot:       lease.record.CatalogSnapshotBinding,
			OrgID:                       lease.record.OrgID,
			WriterCommandID:             lease.record.WriterCommandID,
			WriterReservationGeneration: lease.record.WriterReservationGeneration,
			SourceRequestDigest:         lease.record.RequestDigest,
			RequiredBundleKeySetDigest:  lease.record.RequiredBundleKeySetDigest,
			BundleKey:                   key,
		},
		catalogverificationtx.TenantOwnerRequest{
			OrgID:     lease.record.OrgID,
			BundleKey: key,
		},
	)
	if errors.Is(err, catalogverificationtx.ErrClaimOrder) {
		return nil, fmt.Errorf(
			"%w: owner fences must be acquired in canonical BundleKey order",
			ErrSourceConflict,
		)
	}
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	return claim, nil
}

// BeginSystemBundleKeyVerification claims one exact system-owned required key
// in canonical BundleKey order.
func (lease *CandidateSourceVerificationLease) BeginSystemBundleKeyVerification(
	key layoutbundlekeyv1.BundleKey,
) (*CandidateSourceSystemVerification, error) {
	if lease == nil {
		return nil, ErrSourceConflict
	}
	if err := layoutbundlekeyv1.ValidateBundleKey(key); err != nil {
		return nil, fmt.Errorf("%w: invalid BundleKey: %v", ErrSourceConflict, err)
	}
	if key.SlotOwnerKind != layoutbundlekeyv1.SlotOwnerKindSystem {
		return nil, fmt.Errorf("%w: BundleKey is not system-owned", ErrSourceConflict)
	}

	if !lease.Active() {
		return nil, ErrSourceConflict
	}
	index, required := lease.required[key]
	if !required {
		return nil, fmt.Errorf("%w: BundleKey is not required by the source", ErrSourceConflict)
	}
	claim, err := lease.verification.BeginSystemClaim(
		index,
		catalogverificationtx.SystemSourceFacts{
			SourceCatalogSnapshot:       lease.record.CatalogSnapshotBinding,
			OrgID:                       lease.record.OrgID,
			WriterCommandID:             lease.record.WriterCommandID,
			WriterReservationGeneration: lease.record.WriterReservationGeneration,
			SourceRequestDigest:         lease.record.RequestDigest,
			RequiredBundleKeySetDigest:  lease.record.RequiredBundleKeySetDigest,
			BundleKey:                   key,
		},
		catalogverificationtx.SystemOwnerRequest{BundleKey: key},
	)
	if errors.Is(err, catalogverificationtx.ErrClaimOrder) {
		return nil, fmt.Errorf(
			"%w: owner fences must be acquired in canonical BundleKey order",
			ErrSourceConflict,
		)
	}
	if err != nil {
		return nil, fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	return claim, nil
}

// Active reports whether the source lease and all retained locks are live.
func (lease *CandidateSourceVerificationLease) Active() bool {
	if lease == nil {
		return false
	}
	return lease.authority != nil &&
		lease.db != nil &&
		lease.verification != nil &&
		lease.verification.Active()
}

// Close invalidates the source lease and rolls back its retained read-fence
// transaction.
func (lease *CandidateSourceVerificationLease) Close() error {
	if lease == nil {
		return nil
	}
	if lease.verification == nil {
		return nil
	}
	return lease.verification.Close()
}

func (lease *CandidateSourceVerificationLease) commitRevalidation() error {
	if lease == nil || lease.verification == nil {
		return ErrSourceConflict
	}
	if err := lease.verification.Commit(); err != nil {
		return fmt.Errorf("%w: %v", ErrSourceConflict, err)
	}
	return nil
}

// MarshalJSON rejects persistence of process-scoped source locks.
func (*CandidateSourceVerificationLease) MarshalJSON() ([]byte, error) {
	return nil, errSourceVerificationHasNoJSON
}

// UnmarshalJSON rejects caller construction of a source lease.
func (*CandidateSourceVerificationLease) UnmarshalJSON([]byte) error {
	return errSourceVerificationHasNoJSON
}

func cloneCandidateSourceRecordForVerification(
	record candidateSourceRecord,
) candidateSourceRecord {
	record.RequiredBundleKeySet = cloneRequiredSet(record.RequiredBundleKeySet)
	record.RequiredBundleKeySetCanonical = append(
		[]byte(nil),
		record.RequiredBundleKeySetCanonical...,
	)
	record.SourceAppSelectionSet = cloneSourceAppSelectionSet(record.SourceAppSelectionSet)
	record.SourceAppSelectionCanonical = append(
		[]byte(nil),
		record.SourceAppSelectionCanonical...,
	)
	return record
}