package serving

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

const (
	accountServingGateTable = "_System_ConnectorAccountServingGate"

	accountServingGateSelect = `
SELECT Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
       ServingEpochId, ServingEpochDigest
  FROM _System_ConnectorAccountServingGate
 WHERE ServingScopeDigest = ? AND BINARY ServingScopeDigest = BINARY ?
   AND ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
 LIMIT 1`

	accountServingEpochHeaderSelect = `
SELECT Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State
  FROM _System_ConnectorServingEpoch
 WHERE Id = ? AND BINARY Id = BINARY ?
 LIMIT 1`
)

var (
	errInvalidAccountServingGateInput  = errors.New("connector serving: invalid account serving gate input")
	errAccountServingGateNotFound      = errors.New("connector serving: account serving gate not found")
	errStoredAccountServingGateInvalid = errors.New(
		"connector serving: stored account serving gate invalid",
	)
	errAccountServingGateFenceConflict = errors.New("connector serving: account serving gate fence conflict")
	errAccountServingGateModeConflict  = errors.New("connector serving: account serving gate mode conflict")
	errAccountServingGateEpochConflict = errors.New("connector serving: account serving gate epoch conflict")
)

type accountServingMode string

const (
	accountServingModeV1Serving    accountServingMode = "v1_serving"
	accountServingModeHaltingV1    accountServingMode = "halting_v1"
	accountServingModeV1Stopped    accountServingMode = "v1_stopped"
	accountServingModeV1Drained    accountServingMode = "v1_drained"
	accountServingModeV2Validating accountServingMode = "v2_validating"
	accountServingModeV2Serving    accountServingMode = "v2_serving"
	accountServingModeHaltingV2    accountServingMode = "halting_v2"
	accountServingModeV2Halted     accountServingMode = "v2_halted"

	accountServingEpochStateActive   = "active"
	accountServingEpochStateDraining = "draining"
	accountServingEpochStateRetired  = "retired"
)

// accountServingGate is the provisional read model for one exact account gate.
// It is deliberately private until the complete production transaction mix has
// passed the later capacity certification. This slice exposes no writer or
// transition authority.
type accountServingGate struct {
	id                 string
	servingScopeDigest string
	connectorAccountID string
	mode               accountServingMode
	fenceToken         int64
	servingEpochID     *string
	servingEpochDigest *string
}

func (gate accountServingGate) validateStored() error {
	if !validOpaqueIdentifierWithin(gate.id, connectorDigestProfileV2MaxIdentifierBytes) {
		return storedAccountServingGateInvalid("gate ID is not canonical")
	}
	if !rawSHA256Pattern.MatchString(gate.servingScopeDigest) {
		return storedAccountServingGateInvalid("serving scope digest is not canonical")
	}
	if !validOpaqueIdentifierWithin(gate.connectorAccountID, connectorDigestProfileV2MaxIdentifierBytes) {
		return storedAccountServingGateInvalid("connector account ID is not canonical")
	}
	if gate.fenceToken < 0 {
		return storedAccountServingGateInvalid("fence token is negative")
	}
	if !isClosedAccountServingMode(gate.mode) {
		return storedAccountServingGateInvalid("mode %q is outside the closed account-serving contract", gate.mode)
	}
	if gate.fenceToken == 0 && gate.mode != accountServingModeV1Serving {
		return storedAccountServingGateInvalid("fence zero is valid only for the initial v1_serving state")
	}

	epochPairPresent := gate.servingEpochID != nil && gate.servingEpochDigest != nil
	if (gate.servingEpochID == nil) != (gate.servingEpochDigest == nil) {
		return storedAccountServingGateInvalid("serving epoch ID and digest form a partial pair")
	}
	if accountServingModeRequiresEpoch(gate.mode) != epochPairPresent {
		return storedAccountServingGateInvalid("mode %q has an invalid serving epoch pair", gate.mode)
	}
	if epochPairPresent {
		if !validOpaqueIdentifierWithin(*gate.servingEpochID, connectorDigestProfileV2MaxIdentifierBytes) {
			return storedAccountServingGateInvalid("serving epoch ID is not canonical")
		}
		if !prefixedSHA256Pattern.MatchString(*gate.servingEpochDigest) {
			return storedAccountServingGateInvalid("serving epoch digest is not canonical")
		}
	}
	return nil
}

func isClosedAccountServingMode(mode accountServingMode) bool {
	switch mode {
	case accountServingModeV1Serving,
		accountServingModeHaltingV1,
		accountServingModeV1Stopped,
		accountServingModeV1Drained,
		accountServingModeV2Validating,
		accountServingModeV2Serving,
		accountServingModeHaltingV2,
		accountServingModeV2Halted:
		return true
	default:
		return false
	}
}

func accountServingModeRequiresEpoch(mode accountServingMode) bool {
	switch mode {
	case accountServingModeV2Serving, accountServingModeHaltingV2, accountServingModeV2Halted:
		return true
	default:
		return false
	}
}

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

// loadAccountServingGateExact has one indexed lookup path: exact serving scope
// plus exact connector account. The ordinary equalities preserve use of the
// composite index; the binary residuals make the identity comparison explicit.
func loadAccountServingGateExact(
	ctx context.Context,
	queryer accountServingGateQueryer,
	servingScopeDigest string,
	connectorAccountID string,
	forUpdate bool,
) (accountServingGate, bool, error) {
	if ctx == nil || queryer == nil {
		return accountServingGate{}, false, invalidAccountServingGateInput("context or queryer is unavailable")
	}
	if !rawSHA256Pattern.MatchString(servingScopeDigest) {
		return accountServingGate{}, false, invalidAccountServingGateInput("serving scope digest is not canonical")
	}
	if !validOpaqueIdentifierWithin(connectorAccountID, connectorDigestProfileV2MaxIdentifierBytes) {
		return accountServingGate{}, false, invalidAccountServingGateInput("connector account ID is not canonical")
	}

	query := accountServingGateSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	var (
		gate        accountServingGate
		epochID     sql.NullString
		epochDigest sql.NullString
	)
	err := queryer.QueryRowContext(
		ctx,
		query,
		servingScopeDigest,
		servingScopeDigest,
		connectorAccountID,
		connectorAccountID,
	).Scan(
		&gate.id,
		&gate.servingScopeDigest,
		&gate.connectorAccountID,
		&gate.mode,
		&gate.fenceToken,
		&epochID,
		&epochDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return accountServingGate{}, false, nil
	}
	if err != nil {
		return accountServingGate{}, false, fmt.Errorf("load exact account serving gate: %w", err)
	}
	if epochID.Valid {
		gate.servingEpochID = stringPointer(epochID.String)
	}
	if epochDigest.Valid {
		gate.servingEpochDigest = stringPointer(epochDigest.String)
	}
	if gate.servingScopeDigest != servingScopeDigest || gate.connectorAccountID != connectorAccountID {
		return accountServingGate{}, false, storedAccountServingGateInvalid(
			"indexed lookup returned a non-exact scope or account",
		)
	}
	if err := gate.validateStored(); err != nil {
		return accountServingGate{}, false, err
	}
	return gate, true, nil
}

type accountServingEpochHeader struct {
	id                 string
	epochDigest        string
	servingScopeDigest string
	activeScopeDigest  *string
	state              string
}

func (header accountServingEpochHeader) validateStored() error {
	if !validOpaqueIdentifierWithin(header.id, connectorDigestProfileV2MaxIdentifierBytes) {
		return storedAccountServingEpochHeaderInvalid("epoch ID is not canonical")
	}
	if !prefixedSHA256Pattern.MatchString(header.epochDigest) {
		return storedAccountServingEpochHeaderInvalid("epoch digest is not canonical")
	}
	if !rawSHA256Pattern.MatchString(header.servingScopeDigest) {
		return storedAccountServingEpochHeaderInvalid("serving scope digest is not canonical")
	}
	if header.activeScopeDigest != nil && !rawSHA256Pattern.MatchString(*header.activeScopeDigest) {
		return storedAccountServingEpochHeaderInvalid("active scope digest is not canonical")
	}

	switch header.state {
	case preparedState, verifiedState, rejectedState,
		accountServingEpochStateDraining, accountServingEpochStateRetired:
		if header.activeScopeDigest != nil {
			return storedAccountServingEpochHeaderInvalid(
				"state %q must not retain an active scope digest",
				header.state,
			)
		}
	case accountServingEpochStateActive:
		if header.activeScopeDigest == nil || *header.activeScopeDigest != header.servingScopeDigest {
			return storedAccountServingEpochHeaderInvalid(
				"active state must bind the exact serving scope digest",
			)
		}
	default:
		return storedAccountServingEpochHeaderInvalid(
			"state %q is outside the closed ServingEpoch lifecycle",
			header.state,
		)
	}
	return nil
}

// lockExpectedV2Serving is a read/lock primitive, not an admission or mutation
// API. Its caller owns the transaction. The fixed order is gate first, then the
// gate's narrow exact ServingEpoch header. No other lock order is exposed here.
func lockExpectedV2Serving(
	ctx context.Context,
	tx *sql.Tx,
	servingScopeDigest string,
	connectorAccountID string,
	expectedFenceToken int64,
	expectedEpochID string,
	expectedEpochDigest string,
) (accountServingGate, error) {
	if ctx == nil || tx == nil {
		return accountServingGate{}, invalidAccountServingGateInput("context or transaction is unavailable")
	}
	if expectedFenceToken < 0 {
		return accountServingGate{}, invalidAccountServingGateInput("expected fence token is negative")
	}
	if !validOpaqueIdentifierWithin(expectedEpochID, connectorDigestProfileV2MaxIdentifierBytes) {
		return accountServingGate{}, invalidAccountServingGateInput("expected epoch ID is not canonical")
	}
	if !prefixedSHA256Pattern.MatchString(expectedEpochDigest) {
		return accountServingGate{}, invalidAccountServingGateInput("expected epoch digest is not canonical")
	}

	gate, found, err := loadAccountServingGateExact(
		ctx,
		tx,
		servingScopeDigest,
		connectorAccountID,
		true,
	)
	if err != nil {
		return accountServingGate{}, err
	}
	if !found {
		return accountServingGate{}, fmt.Errorf(
			"%w: scope=%s account=%s",
			errAccountServingGateNotFound,
			servingScopeDigest,
			connectorAccountID,
		)
	}
	// Fence is classified first so a waiter that observes a completed transition
	// receives the stale-observation error even when that transition also changed
	// the gate mode.
	if gate.fenceToken != expectedFenceToken {
		return accountServingGate{}, fmt.Errorf(
			"%w: stored=%d expected=%d",
			errAccountServingGateFenceConflict,
			gate.fenceToken,
			expectedFenceToken,
		)
	}
	if gate.mode != accountServingModeV2Serving {
		return accountServingGate{}, fmt.Errorf(
			"%w: stored=%s required=%s",
			errAccountServingGateModeConflict,
			gate.mode,
			accountServingModeV2Serving,
		)
	}
	if gate.servingEpochID == nil || gate.servingEpochDigest == nil ||
		*gate.servingEpochID != expectedEpochID || *gate.servingEpochDigest != expectedEpochDigest {
		return accountServingGate{}, fmt.Errorf(
			"%w: gate does not name the expected exact epoch",
			errAccountServingGateEpochConflict,
		)
	}

	header, found, err := loadAccountServingEpochHeaderExact(ctx, tx, expectedEpochID, true)
	if err != nil {
		return accountServingGate{}, err
	}
	if !found {
		return accountServingGate{}, fmt.Errorf(
			"%w: referenced epoch %q is absent",
			errAccountServingGateEpochConflict,
			expectedEpochID,
		)
	}
	if header.id != expectedEpochID ||
		header.epochDigest != expectedEpochDigest ||
		header.servingScopeDigest != servingScopeDigest ||
		header.activeScopeDigest == nil ||
		*header.activeScopeDigest != servingScopeDigest ||
		header.state != accountServingEpochStateActive {
		return accountServingGate{}, fmt.Errorf(
			"%w: referenced epoch is not the exact active serving scope",
			errAccountServingGateEpochConflict,
		)
	}
	return gate, nil
}

func loadAccountServingEpochHeaderExact(
	ctx context.Context,
	queryer accountServingGateQueryer,
	epochID string,
	forUpdate bool,
) (accountServingEpochHeader, bool, error) {
	query := accountServingEpochHeaderSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	var (
		header      accountServingEpochHeader
		activeScope sql.NullString
	)
	err := queryer.QueryRowContext(ctx, query, epochID, epochID).Scan(
		&header.id,
		&header.epochDigest,
		&header.servingScopeDigest,
		&activeScope,
		&header.state,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return accountServingEpochHeader{}, false, nil
	}
	if err != nil {
		return accountServingEpochHeader{}, false, fmt.Errorf("load exact account serving epoch header: %w", err)
	}
	if activeScope.Valid {
		header.activeScopeDigest = stringPointer(activeScope.String)
	}
	if header.id != epochID {
		return accountServingEpochHeader{}, false, storedAccountServingEpochHeaderInvalid(
			"indexed lookup returned a non-exact epoch ID",
		)
	}
	if err := header.validateStored(); err != nil {
		return accountServingEpochHeader{}, false, err
	}
	return header, true, nil
}

func invalidAccountServingGateInput(format string, args ...any) error {
	return fmt.Errorf("%w: %s", errInvalidAccountServingGateInput, fmt.Sprintf(format, args...))
}

func storedAccountServingGateInvalid(format string, args ...any) error {
	return fmt.Errorf("%w: %s", errStoredAccountServingGateInvalid, fmt.Sprintf(format, args...))
}

func storedAccountServingEpochHeaderInvalid(format string, args ...any) error {
	return fmt.Errorf(
		"%w: account serving epoch header: %s",
		ErrStoredCandidateInvalid,
		fmt.Sprintf(format, args...),
	)
}