package catalogsystem

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

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

const authorityLockName = catalogsystemfence.Name

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

func (a *Authority) transact(
	ctx context.Context,
	run func(*sql.Tx) (commandOutcome, error),
) (commandOutcome, error) {
	session, err := a.beginAuthoritySession(ctx)
	if err != nil {
		return commandOutcome{}, err
	}
	outcome, runErr := run(session.tx)
	if runErr != nil {
		rollbackErr := session.tx.Rollback()
		session.tx = nil
		releaseErr := session.release(ctx, rollbackErr != nil)
		return commandOutcome{}, errors.Join(runErr, rollbackErr, releaseErr)
	}
	commitErr := session.tx.Commit()
	session.tx = nil
	releaseErr := session.release(ctx, commitErr != nil)
	if commitErr != nil || releaseErr != nil {
		return commandOutcome{}, errors.Join(commitErr, releaseErr)
	}
	return outcome, nil
}

func (a *Authority) beginAuthoritySession(ctx context.Context) (*authoritySession, error) {
	if a == nil || a.db == nil {
		return nil, ErrConflict
	}
	conn, err := a.db.Conn(ctx)
	if err != nil {
		return nil, err
	}
	mode, err := catalogsystemfence.ForcePessimistic(ctx, conn)
	if err != nil {
		return nil, errors.Join(err, conn.Close())
	}
	session := &authoritySession{conn: conn, mode: mode}

	fence, err := catalogsystemfence.AcquireOnConn(ctx, conn)
	if err != nil {
		unsafe := !errors.Is(err, catalogsystemfence.ErrNotAcquired)
		releaseErr := session.release(ctx, unsafe)
		if errors.Is(err, catalogsystemfence.ErrNotAcquired) {
			return nil, errors.Join(
				fmt.Errorf("%w: global authority fence was not acquired", ErrConflict),
				releaseErr,
			)
		}
		return nil, errors.Join(err, releaseErr)
	}
	session.fence = fence

	// TiDB does not implement SQL SERIALIZABLE. The fixed session-level
	// authority lock admits exactly one cooperating writer at a time, while
	// explicit pessimistic REPEATABLE READ supplies the transaction snapshot
	// and current-row locks. Complete-key uniqueness and exact CAS predicates
	// close write races, giving this authority protocol one serial command
	// order.
	tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
	if err != nil {
		return nil, errors.Join(err, session.release(ctx, true))
	}
	session.tx = tx
	return session, nil
}

// release ends any remaining transaction while the advisory fence is still
// held, releases that fence on the exact retained session, then restores a
// clean session or discards an ambiguous one.
func (session *authoritySession) release(ctx context.Context, unsafe bool) error {
	if session == nil || session.conn == nil {
		return nil
	}
	var rollbackErr error
	if session.tx != nil {
		rollbackErr = session.tx.Rollback()
		if errors.Is(rollbackErr, sql.ErrTxDone) {
			rollbackErr = nil
		}
		if rollbackErr != nil {
			unsafe = true
		}
		session.tx = nil
	}

	var fenceErr error
	if session.fence != nil {
		fenceErr = session.fence.ReleaseOnConn(ctx, session.conn)
		session.fence = nil
		if fenceErr != nil {
			unsafe = true
		}
	}
	if errors.Is(fenceErr, catalogsystemfence.ErrNotReleased) {
		fenceErr = fmt.Errorf(
			"%w: global authority fence was not released: %v",
			ErrCorrupt,
			fenceErr,
		)
	}

	var modeErr error
	if unsafe {
		modeErr = catalogsystemfence.Discard(session.conn)
	} else {
		modeErr = session.mode.Restore(ctx, session.conn)
	}
	closeErr := session.conn.Close()
	session.conn = nil
	session.mode = nil
	return errors.Join(rollbackErr, fenceErr, modeErr, closeErr)
}