// Package catalogverificationtx owns the process-local transaction retained
// while Catalog source and owner-authority fences are live.
//
// It is internal so no caller outside the control-plane module can obtain a
// database transaction, inject a query callback, or manufacture an owner
// verification path.
package catalogverificationtx

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

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogsystemfence"
)

var (
	ErrClosed            = errors.New("catalog verification transaction: closed")
	ErrAuthorityMismatch = errors.New("catalog verification transaction: authority mismatch")
)

// Token is an opaque retained Catalog verification transaction. Its only
// database operations are the fixed-purpose authority checks in this package.
//
// Production source verification uses an owned Token, which retains the exact
// *sql.Conn, transaction, and TiDB session-mode guard for its whole lifetime.
// The legacy transaction-only form exists only for App/tenant focused tests and
// cannot acquire the session-scoped system authority fence.
type Token struct {
	mu          sync.RWMutex
	db          *sql.DB
	conn        *sql.Conn
	tx          *sql.Tx
	mode        *catalogsystemfence.PessimisticModeGuard
	systemFence *catalogsystemfence.Fence
	source      sourceBinding
	owned       bool
	active      bool
}

// NewOwned binds one retained connection, its live transaction, and its TiDB
// pessimistic-mode guard. The Token assumes sole responsibility for ending the
// transaction, restoring or discarding the session, and closing the connection.
func NewOwned(
	db *sql.DB,
	conn *sql.Conn,
	tx *sql.Tx,
	mode *catalogsystemfence.PessimisticModeGuard,
) (*Token, error) {
	if db == nil || conn == nil || tx == nil || mode == nil {
		return nil, errors.New(
			"catalog verification transaction: database, connection, transaction, and session guard are required",
		)
	}
	return &Token{
		db:     db,
		conn:   conn,
		tx:     tx,
		mode:   mode,
		owned:  true,
		active: true,
	}, nil
}

// ActiveFor reports whether the exact originating database and retained
// transaction/session lifetime are still owned.
func (token *Token) ActiveFor(db *sql.DB) bool {
	if token == nil || db == nil {
		return false
	}
	token.mu.RLock()
	defer token.mu.RUnlock()
	return token.activeForLocked(db)
}

func (token *Token) activeForLocked(db *sql.DB) bool {
	if !token.active || token.db != db || token.tx == nil {
		return false
	}
	return !token.owned || (token.conn != nil && token.mode != nil)
}

func (token *Token) systemAuthorityFenceActive() bool {
	if token == nil {
		return false
	}
	token.mu.Lock()
	if !token.activeForLocked(token.db) ||
		!token.owned ||
		token.conn == nil ||
		token.mode == nil ||
		token.systemFence == nil {
		token.mu.Unlock()
		return false
	}
	if err := token.systemFence.ValidateOnTx(
		context.Background(),
		token.tx,
	); err == nil {
		token.mu.Unlock()
		return true
	}

	// A failed ownership probe means the retained transaction, its row locks,
	// and the advisory fence can no longer be treated as one live authority
	// scope. Invalidate first, then discard the session so no ambiguous
	// connection can return to the pool.
	resources := token.detachLocked()
	token.mu.Unlock()
	_ = discardResources(resources, catalogsystemfence.ErrStateUnknown)
	return false
}

type tokenResources struct {
	conn  *sql.Conn
	tx    *sql.Tx
	mode  *catalogsystemfence.PessimisticModeGuard
	fence *catalogsystemfence.Fence
	owned bool
}

func (token *Token) detachLocked() tokenResources {
	resources := tokenResources{
		conn:  token.conn,
		tx:    token.tx,
		mode:  token.mode,
		fence: token.systemFence,
		owned: token.owned,
	}
	token.active = false
	token.conn = nil
	token.tx = nil
	token.mode = nil
	token.systemFence = nil
	token.source = sourceBinding{}
	return resources
}

// Close invalidates the token before releasing its locks. It waits for any
// in-flight fixed-purpose verifier to finish before rolling the transaction
// back.
func (token *Token) Close() error {
	if token == nil {
		return nil
	}
	token.mu.Lock()
	if !token.active {
		token.mu.Unlock()
		return nil
	}
	resources := token.detachLocked()
	token.mu.Unlock()
	return closeResources(resources)
}

// Commit invalidates the token before committing. It exists for the
// source-authority's lock-and-revalidate compatibility path; owner verifiers
// cannot commit through their public API.
func (token *Token) Commit() error {
	if token == nil {
		return ErrClosed
	}
	token.mu.Lock()
	if !token.active || token.tx == nil {
		token.mu.Unlock()
		return ErrClosed
	}
	resources := token.detachLocked()
	token.mu.Unlock()
	return commitResources(resources)
}

func closeResources(resources tokenResources) error {
	if resources.tx == nil {
		return closeOwnedConnection(resources, ErrClosed)
	}
	if !resources.owned {
		rollbackErr := resources.tx.Rollback()
		if errors.Is(rollbackErr, sql.ErrTxDone) {
			rollbackErr = nil
		}
		return rollbackErr
	}

	rollbackErr := resources.tx.Rollback()
	if errors.Is(rollbackErr, sql.ErrTxDone) {
		rollbackErr = nil
	}
	var fenceErr error
	if resources.fence != nil {
		fenceErr = resources.fence.ReleaseOnConn(context.Background(), resources.conn)
	}
	if fenceErr != nil || rollbackErr != nil {
		return errors.Join(
			fenceErr,
			rollbackErr,
			closeOwnedConnection(resources, catalogsystemfence.ErrStateUnknown),
		)
	}
	return closeOwnedConnection(resources, nil)
}

func discardResources(resources tokenResources, cause error) error {
	var rollbackErr error
	if resources.tx != nil {
		rollbackErr = resources.tx.Rollback()
		if errors.Is(rollbackErr, sql.ErrTxDone) {
			rollbackErr = nil
		}
	}
	return errors.Join(
		cause,
		rollbackErr,
		closeOwnedConnection(resources, catalogsystemfence.ErrStateUnknown),
	)
}

func commitResources(resources tokenResources) error {
	if resources.tx == nil {
		return closeOwnedConnection(resources, ErrClosed)
	}
	if !resources.owned {
		return resources.tx.Commit()
	}

	commitErr := resources.tx.Commit()
	var fenceErr error
	if resources.fence != nil {
		fenceErr = resources.fence.ReleaseOnConn(context.Background(), resources.conn)
	}
	if commitErr != nil || fenceErr != nil {
		return errors.Join(
			commitErr,
			fenceErr,
			closeOwnedConnection(resources, catalogsystemfence.ErrStateUnknown),
		)
	}
	return closeOwnedConnection(resources, nil)
}

// closeOwnedConnection either restores a clean session or discards an
// ambiguous one, then releases the explicitly retained *sql.Conn.
func closeOwnedConnection(resources tokenResources, unsafe error) error {
	if !resources.owned || resources.conn == nil {
		return unsafe
	}
	if unsafe != nil {
		return errors.Join(
			unsafe,
			catalogsystemfence.Discard(resources.conn),
			resources.conn.Close(),
		)
	}
	restoreErr := resources.mode.Restore(context.Background(), resources.conn)
	return errors.Join(restoreErr, resources.conn.Close())
}

// acquireSystemAuthorityFence acquires the shared session advisory lock once
// inside this retained source transaction. The fence remains held across every
// later system-key verification and is released only by Close or Commit.
func (token *Token) acquireSystemAuthorityFence(
	ctx context.Context,
	db *sql.DB,
) error {
	if token == nil {
		return ErrClosed
	}
	token.mu.Lock()
	if !token.active || token.tx == nil {
		token.mu.Unlock()
		return ErrClosed
	}
	if db == nil || token.db != db {
		token.mu.Unlock()
		return ErrAuthorityMismatch
	}
	if !token.owned || token.conn == nil || token.mode == nil {
		token.mu.Unlock()
		return ErrAuthorityMismatch
	}
	if token.systemFence != nil {
		token.mu.Unlock()
		return nil
	}
	fence, err := catalogsystemfence.AcquireOnTx(ctx, token.tx)
	if err == nil {
		token.systemFence = fence
		token.mu.Unlock()
		return nil
	}
	if errors.Is(err, catalogsystemfence.ErrNotAcquired) {
		token.mu.Unlock()
		return err
	}

	// GET_LOCK may have succeeded even if its response was lost. End the
	// transaction and discard the owning session before reporting the error.
	resources := token.detachLocked()
	token.mu.Unlock()
	return discardResources(resources, err)
}

func (token *Token) acquire(db *sql.DB) (*sql.Tx, error) {
	if token == nil {
		return nil, ErrClosed
	}
	token.mu.RLock()
	if !token.activeForLocked(db) {
		token.mu.RUnlock()
		if db == nil || token.db != db {
			return nil, ErrAuthorityMismatch
		}
		return nil, ErrClosed
	}
	return token.tx, nil
}

func (token *Token) release() {
	token.mu.RUnlock()
}