package persistence

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

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
)

type permissionStateAttemptKey struct{}

type permissionStateAttemptState struct {
	mu        sync.Mutex
	requested bool
	advanced  bool
	receipt   permissionstatev1.AdvanceReceipt
	// Compatibility record writes share the canonical PermissionSet revision.
	// This state belongs to the SQL attempt, never to a request or retry.
	authoringAdvanced map[string]bool
}

func withPermissionStateAttemptState(ctx context.Context) context.Context {
	return context.WithValue(ctx, permissionStateAttemptKey{}, &permissionStateAttemptState{})
}

func permissionStateAttemptFromContext(ctx context.Context) *permissionStateAttemptState {
	if ctx == nil {
		return nil
	}
	state, _ := ctx.Value(permissionStateAttemptKey{}).(*permissionStateAttemptState)
	return state
}

// PermissionStateAuthority owns the one permitted transaction-bound revision
// advance. A caller that needs the receipt immediately acquires its complete
// domain/target lock set, establishes an exact semantic before/after delta,
// invokes AdvanceIfChanged, and acquires any contended evidence/outbox locks
// afterward. Generic nested writers use RequestAdvanceIfChanged so the
// TransactionManager can wait until the callback has acquired every target
// lock before it locks the singleton.
type PermissionStateAuthority struct {
	transactions *TransactionManager
}

func NewPermissionStateAuthority(transactions *TransactionManager) (*PermissionStateAuthority, error) {
	if transactions == nil || transactions.db == nil {
		return nil, errors.New("permission state authority: transaction manager is required")
	}
	return &PermissionStateAuthority{transactions: transactions}, nil
}

// RequestAdvanceIfChanged records a semantic change on the current
// transaction attempt. The TransactionManager flushes the request through
// AdvanceIfChanged after the callback has acquired all mutation target locks
// and before it prepares the commit. Repeated and nested requests coalesce.
func (authority *PermissionStateAuthority) RequestAdvanceIfChanged(
	ctx context.Context,
	changed bool,
) error {
	if authority == nil || authority.transactions == nil {
		return errors.New("permission state authority: unavailable")
	}
	if ctx == nil {
		return errors.New("permission state authority: context is required")
	}
	if !authority.transactions.IsTransactionActive(ctx) {
		return errors.New("permission state authority: active transaction is required")
	}
	if !ports.TransactionAbortSupported(ctx) {
		return errors.New("permission state authority: transaction abort support is required")
	}
	if err := ports.TransactionAbortErr(ctx); err != nil {
		return err
	}
	attempt := permissionStateAttemptFromContext(ctx)
	if attempt == nil {
		return abortPermissionStateTransaction(
			ctx,
			errors.New("permission state authority: transaction attempt state is required"),
		)
	}
	if !changed {
		return nil
	}
	if !authority.transactions.ownsPessimisticTx(ctx) {
		return abortPermissionStateTransaction(
			ctx,
			errors.New("permission state authority: changed mutation requires an owned pessimistic transaction"),
		)
	}
	attempt.mu.Lock()
	attempt.requested = true
	attempt.mu.Unlock()
	return nil
}

// LockPermissionStateAtLeast is the fresh-effect authorization fence. It locks
// the latest committed singleton in the caller-owned pessimistic transaction
// and proves it is in the admitted floor's lineage at or above that revision.
// A concurrent permission writer therefore either commits before this read and
// is observed by the caller's subsequent uncached permission query, or waits
// until this transaction completes.
func (authority *PermissionStateAuthority) LockPermissionStateAtLeast(
	ctx context.Context,
	admittedFloor permissionstatev1.Version,
) (permissionstatev1.Version, error) {
	if authority == nil || authority.transactions == nil {
		return permissionstatev1.Version{}, errors.New("permission state authority: unavailable")
	}
	if err := admittedFloor.Validate(); err != nil {
		return permissionstatev1.Version{}, err
	}
	if ctx == nil || !authority.transactions.IsTransactionActive(ctx) || !authority.transactions.ownsPessimisticTx(ctx) {
		return permissionstatev1.Version{}, errors.New(
			"permission state authority: authorization fence requires an owned pessimistic transaction",
		)
	}
	tx, ok := authority.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return permissionstatev1.Version{}, errors.New("permission state authority: SQL transaction is unavailable")
	}
	state, found, err := readPermissionState(ctx, tx, true)
	if err != nil {
		return permissionstatev1.Version{}, err
	}
	if !found {
		return permissionstatev1.Version{}, errors.New("permission state authority: singleton is missing")
	}
	satisfies, err := state.Version.Satisfies(admittedFloor)
	if err != nil {
		return permissionstatev1.Version{}, err
	}
	if !satisfies {
		return permissionstatev1.Version{}, &permissionstatev1.Failure{
			Category: permissionstatev1.FailureStaleSnapshot,
		}
	}
	return state.Version, nil
}

func (transactions *TransactionManager) advanceRequestedPermissionState(ctx context.Context) error {
	attempt := permissionStateAttemptFromContext(ctx)
	if attempt == nil {
		return nil
	}
	attempt.mu.Lock()
	requested := attempt.requested
	attempt.mu.Unlock()
	if !requested {
		return nil
	}
	authority := &PermissionStateAuthority{transactions: transactions}
	_, _, err := authority.AdvanceIfChanged(ctx, true)
	return err
}

// AdvanceIfChanged advances Permission State once for the active semantic
// transaction. Nested changed calls reuse the first receipt. A semantic no-op
// neither locks nor advances Permission State.
//
// A changed call requires a TransactionManager-owned pessimistic transaction.
// Any revision failure aborts that outer transaction, so swallowing the returned
// error cannot commit semantic writes without their revision.
func (authority *PermissionStateAuthority) AdvanceIfChanged(
	ctx context.Context,
	changed bool,
) (permissionstatev1.AdvanceReceipt, bool, error) {
	if authority == nil || authority.transactions == nil {
		return permissionstatev1.AdvanceReceipt{}, false, errors.New("permission state authority: unavailable")
	}
	if ctx == nil {
		return permissionstatev1.AdvanceReceipt{}, false, errors.New("permission state authority: context is required")
	}
	if !authority.transactions.IsTransactionActive(ctx) {
		return permissionstatev1.AdvanceReceipt{}, false, errors.New("permission state authority: active transaction is required")
	}
	if !ports.TransactionAbortSupported(ctx) {
		return permissionstatev1.AdvanceReceipt{}, false, errors.New("permission state authority: transaction abort support is required")
	}
	if err := ports.TransactionAbortErr(ctx); err != nil {
		return permissionstatev1.AdvanceReceipt{}, false, err
	}
	attempt := permissionStateAttemptFromContext(ctx)
	if attempt == nil {
		return permissionstatev1.AdvanceReceipt{}, false, abortPermissionStateTransaction(
			ctx,
			errors.New("permission state authority: transaction attempt state is required"),
		)
	}
	if !changed {
		return permissionstatev1.AdvanceReceipt{}, false, nil
	}
	if !authority.transactions.ownsPessimisticTx(ctx) {
		return permissionstatev1.AdvanceReceipt{}, false, abortPermissionStateTransaction(
			ctx,
			errors.New("permission state authority: changed mutation requires an owned pessimistic transaction"),
		)
	}
	tx, ok := authority.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return permissionstatev1.AdvanceReceipt{}, false, abortPermissionStateTransaction(
			ctx,
			errors.New("permission state authority: SQL transaction is unavailable"),
		)
	}

	attempt.mu.Lock()
	defer attempt.mu.Unlock()
	if attempt.advanced {
		return attempt.receipt, true, nil
	}

	receipt, err := advancePermissionStateOnTransaction(ctx, tx)
	if err != nil {
		return permissionstatev1.AdvanceReceipt{}, false, abortPermissionStateTransaction(ctx, err)
	}

	attempt.receipt = receipt
	attempt.advanced = true
	return receipt, true, nil
}

func abortPermissionStateTransaction(ctx context.Context, cause error) error {
	if ports.AbortTransaction(ctx, cause) {
		if err := ports.TransactionAbortErr(ctx); err != nil {
			return err
		}
	}
	return cause
}