// Package catalogsystemfence owns the one session-scoped advisory lock shared
// by Catalog system-slot writers and retained owner verifiers.
package catalogsystemfence

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"sync"
	"time"
)

const (
	// Name is the exact cluster-wide Catalog system-slot authority fence.
	Name = "nexuscrm_catalog_system_layout_bundle_authority"

	acquireTimeoutSeconds = 600
	cleanupTimeout        = 5 * time.Second
)

var (
	// ErrNotAcquired means GET_LOCK completed without granting the fence.
	ErrNotAcquired = errors.New("catalog system authority fence was not acquired")
	// ErrOwnershipLost means the retained transaction's exact session no
	// longer owns the fence it acquired.
	ErrOwnershipLost = errors.New(
		"catalog system authority fence is not owned by the retained session",
	)
	// ErrNotReleased means RELEASE_LOCK completed without releasing the fence.
	ErrNotReleased = errors.New("catalog system authority fence was not released")
	// ErrStateUnknown means a session might still own the advisory lock and
	// must be discarded instead of returned to the pool.
	ErrStateUnknown = errors.New("catalog system authority fence state is unknown")
	// ErrClosed means the retained fence has already been released.
	ErrClosed = errors.New("catalog system authority fence is closed")
)

type queryRower interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

// Fence records ownership of the advisory lock without owning a second
// database connection. Its caller must use the same *sql.Conn or *sql.Tx for
// both acquisition and release.
type Fence struct {
	mu           sync.Mutex
	sessionID    int64
	sessionBound bool
	active       bool
}

// PessimisticModeGuard records the prior TiDB transaction mode for one
// explicitly retained session. Restore returns the session to its prior mode;
// if that is ambiguous, it marks the connection bad so it cannot re-enter the
// pool.
type PessimisticModeGuard struct {
	mu       sync.Mutex
	previous string
	active   bool
}

// ForcePessimistic snapshots one retained session's current mode and explicitly
// selects TiDB pessimistic transactions before a transaction begins.
func ForcePessimistic(
	ctx context.Context,
	conn *sql.Conn,
) (*PessimisticModeGuard, error) {
	if conn == nil {
		return nil, errors.New("catalog system authority session: connection is required")
	}
	var previous string
	if err := conn.QueryRowContext(
		ctx,
		"SELECT @@SESSION.tidb_txn_mode",
	).Scan(&previous); err != nil {
		return nil, fmt.Errorf("snapshot TiDB transaction mode: %w", err)
	}
	if _, err := conn.ExecContext(
		ctx,
		"SET SESSION tidb_txn_mode = 'pessimistic'",
	); err != nil {
		return nil, errors.Join(
			fmt.Errorf("select TiDB pessimistic transaction mode: %w", err),
			Discard(conn),
		)
	}
	return &PessimisticModeGuard{previous: previous, active: true}, nil
}

// Restore restores the exact prior TiDB transaction mode. It is idempotent
// after a successful restore. A failed restore discards the session.
func (guard *PessimisticModeGuard) Restore(
	ctx context.Context,
	conn *sql.Conn,
) error {
	if guard == nil {
		return nil
	}
	guard.mu.Lock()
	defer guard.mu.Unlock()
	if !guard.active {
		return nil
	}
	guard.active = false
	if conn == nil {
		return ErrClosed
	}
	cleanupContext, cancel := cleanupContext(ctx)
	defer cancel()
	if _, err := conn.ExecContext(
		cleanupContext,
		"SET SESSION tidb_txn_mode = ?",
		guard.previous,
	); err != nil {
		return errors.Join(
			fmt.Errorf("restore TiDB transaction mode: %w", err),
			Discard(conn),
		)
	}
	return nil
}

// AcquireOnConn obtains the shared system authority fence on an explicitly
// retained session before its writer transaction begins.
func AcquireOnConn(ctx context.Context, conn *sql.Conn) (*Fence, error) {
	if conn == nil {
		return nil, errors.New("catalog system authority fence: connection is required")
	}
	return acquire(ctx, conn)
}

// AcquireOnTx obtains the shared system authority fence inside the retained
// source transaction, hence on the exact session already owning source locks.
func AcquireOnTx(ctx context.Context, tx *sql.Tx) (*Fence, error) {
	if tx == nil {
		return nil, errors.New("catalog system authority fence: transaction is required")
	}
	fence, err := acquire(ctx, tx)
	if err != nil {
		return nil, err
	}
	if err := fence.bindSession(ctx, tx); err != nil {
		return nil, err
	}
	return fence, nil
}

func acquire(ctx context.Context, session queryRower) (*Fence, error) {
	var acquired sql.NullInt64
	err := session.QueryRowContext(
		ctx,
		fmt.Sprintf("SELECT GET_LOCK('%s', %d)", Name, acquireTimeoutSeconds),
	).Scan(&acquired)
	if err != nil {
		return nil, errors.Join(ErrStateUnknown, err)
	}
	if !acquired.Valid || (acquired.Int64 != 0 && acquired.Int64 != 1) {
		return nil, ErrStateUnknown
	}
	if acquired.Int64 == 0 {
		return nil, ErrNotAcquired
	}
	return &Fence{active: true}, nil
}

func (fence *Fence) bindSession(ctx context.Context, tx *sql.Tx) error {
	var sessionID int64
	if err := tx.QueryRowContext(ctx, "SELECT CONNECTION_ID()").Scan(
		&sessionID,
	); err != nil {
		fence.active = false
		return errors.Join(ErrStateUnknown, err)
	}
	if sessionID <= 0 {
		fence.active = false
		return ErrStateUnknown
	}
	fence.sessionID = sessionID
	fence.sessionBound = true
	return nil
}

// ValidateOnTx re-proves that the exact transaction session bound at
// acquisition is still reachable and still owns the shared advisory lock.
// Any ambiguous or negative result permanently invalidates the fence. The
// caller that owns the retained connection must then discard that session;
// validation deliberately cannot release an advisory lock whose state is no
// longer trustworthy.
func (fence *Fence) ValidateOnTx(ctx context.Context, tx *sql.Tx) error {
	if tx == nil {
		return ErrClosed
	}
	if fence == nil {
		return ErrClosed
	}
	fence.mu.Lock()
	defer fence.mu.Unlock()
	if !fence.active || !fence.sessionBound || fence.sessionID <= 0 {
		return ErrClosed
	}

	probeContext, cancel := cleanupContext(ctx)
	defer cancel()
	var ownerSessionID sql.NullInt64
	var currentSessionID int64
	if err := tx.QueryRowContext(
		probeContext,
		"SELECT IS_USED_LOCK('"+Name+"'), CONNECTION_ID()",
	).Scan(&ownerSessionID, &currentSessionID); err != nil {
		fence.active = false
		return errors.Join(ErrStateUnknown, err)
	}
	if !ownerSessionID.Valid ||
		ownerSessionID.Int64 != fence.sessionID ||
		currentSessionID != fence.sessionID {
		fence.active = false
		return ErrOwnershipLost
	}
	return nil
}

// ReleaseOnConn releases a writer or verifier fence on the same retained
// session after its SQL transaction has ended.
func (fence *Fence) ReleaseOnConn(ctx context.Context, conn *sql.Conn) error {
	if conn == nil {
		return ErrClosed
	}
	return fence.release(ctx, conn)
}

func (fence *Fence) release(ctx context.Context, session queryRower) error {
	if fence == nil {
		return nil
	}
	fence.mu.Lock()
	defer fence.mu.Unlock()
	if !fence.active {
		return nil
	}
	fence.active = false

	cleanupContext, cancel := cleanupContext(ctx)
	defer cancel()
	var released sql.NullInt64
	err := session.QueryRowContext(
		cleanupContext,
		"SELECT RELEASE_LOCK('"+Name+"')",
	).Scan(&released)
	if err != nil {
		return errors.Join(ErrNotReleased, ErrStateUnknown, err)
	}
	if !released.Valid || released.Int64 != 1 {
		return errors.Join(ErrNotReleased, ErrStateUnknown)
	}
	return nil
}

// Discard marks a retained session bad so database/sql closes it instead of
// returning a possibly lock-owning or session-mutated connection to the pool.
func Discard(conn *sql.Conn) error {
	if conn == nil {
		return nil
	}
	err := conn.Raw(func(any) error {
		return driver.ErrBadConn
	})
	if errors.Is(err, driver.ErrBadConn) {
		return nil
	}
	return err
}

func cleanupContext(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout)
}