package serving

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"strconv"
	"time"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/serviceconfiguration"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	applicationresultv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	flowcommandv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

const executionSelect = `
SELECT Id, ExecutionProtocolVersion, ServingScopeDigest, ConnectorAccountId,
       AccountServingGateId, AccountServingFenceToken,
       ServingEpochId, ServingEpochDigest, ServingEpochBindingId, PackGenerationId,
       DefinitionRef, DefinitionVersion, CanonicalInputs, ExternalEffectsMode,
       InputDigest, ExecutionRefDigest, RuntimeCommandId,
       State, ClaimFenceToken, ClaimHolderId, ClaimLeaseUntil,
       ResultDigest, RuntimeCommitReceipt, QuarantineReasonCode
  FROM _System_ConnectorV2Execution
 WHERE Id = ? AND BINARY Id = BINARY ?
 LIMIT 1`

// ExecutionRepository owns the only Connector read/write path for the
// Core-owned V2 Execution and append-only ExecutionAttempt tables.
type ExecutionRepository struct {
	db *sql.DB
}

func NewExecutionRepository(db *sql.DB) (*ExecutionRepository, error) {
	if db == nil {
		return nil, fmt.Errorf("%w: execution database is nil", ErrDependencyMissing)
	}
	return &ExecutionRepository{db: db}, nil
}

// VerifySchema fails closed on any drift in the complete execution authority.
func (repository *ExecutionRepository) VerifySchema(ctx context.Context) error {
	if repository == nil || repository.db == nil {
		return fmt.Errorf("%w: execution repository is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	return verifyExecutionSchema(ctx, repository.db)
}

func (repository *ExecutionRepository) LoadExecution(
	ctx context.Context,
	id string,
) (Execution, bool, error) {
	if repository == nil || repository.db == nil {
		return Execution{}, false, fmt.Errorf("%w: execution repository is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return Execution{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validOpaqueIdentifierWithin(id, connectorDigestProfileV2MaxIdentifierBytes) {
		return Execution{}, false, invalidExecutionCandidate("execution ID is not canonical")
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Execution{}, false, err
	}
	return loadExecutionExact(ctx, repository.db, id, false)
}

// AdmitConformance inserts one immutable dependency-free execution only after
// locking and validating its complete active authority tuple. Exact retries
// return the existing lifecycle without resetting it.
func (repository *ExecutionRepository) AdmitConformance(
	ctx context.Context,
	candidate ConformanceExecutionCandidate,
) (Execution, bool, error) {
	if repository == nil || repository.db == nil {
		return Execution{}, false, fmt.Errorf("%w: execution repository is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return Execution{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	execution := candidate.execution.copy()
	if err := execution.validateStored(); err != nil {
		return Execution{}, false, invalidExecutionCandidate("candidate is invalid: %v", err)
	}
	if execution.state != executionStateReady || execution.claimFenceToken != 0 ||
		execution.claimHolderID != nil || execution.claimLeaseUntil != nil {
		return Execution{}, false, invalidExecutionCandidate("candidate is not a pristine ready execution")
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Execution{}, false, err
	}

	// Immutable exact retries remain resolvable after the serving gate moves.
	if stored, found, err := loadExecutionExact(ctx, repository.db, execution.id, false); err != nil {
		return Execution{}, false, err
	} else if found {
		if !stored.equalImmutable(execution) {
			return Execution{}, false, executionIdentityConflict("execution ID names different immutable bytes")
		}
		return stored.copy(), true, nil
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return Execution{}, false, fmt.Errorf("begin execution admission: %w", err)
	}
	rollback := func(cause error) (Execution, bool, error) {
		return Execution{}, false, errors.Join(cause, tx.Rollback(), cleanup())
	}
	commit := func() error { return errors.Join(tx.Commit(), cleanup()) }

	if _, err := lockExecutionAuthority(ctx, tx, execution); err != nil {
		return rollback(err)
	}
	if err := insertExecution(ctx, tx, execution); err != nil {
		if !isMySQLDuplicateKey(err) {
			return rollback(fmt.Errorf("insert execution: %w", err))
		}
		rollbackErr := errors.Join(tx.Rollback(), cleanup())
		stored, found, loadErr := loadExecutionExact(ctx, repository.db, execution.id, false)
		if loadErr != nil {
			return Execution{}, false, errors.Join(loadErr, rollbackErr)
		}
		if !found || !stored.equalImmutable(execution) {
			return Execution{}, false, errors.Join(
				executionIdentityConflict("execution ID, ExecutionRef, or Runtime command already names different bytes"),
				rollbackErr,
			)
		}
		return stored.copy(), true, rollbackErr
	}
	stored, found, err := loadExecutionExact(ctx, tx, execution.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found || !stored.equalImmutable(execution) || stored.state != executionStateReady {
		return rollback(storedExecutionInvalid("inserted execution cannot be reloaded byte-exactly"))
	}
	if err := commit(); err != nil {
		return Execution{}, false, fmt.Errorf("commit execution admission: %w", err)
	}
	return stored.copy(), false, nil
}

// ClaimConformance acquires one DB-time lease and appends immutable attempt
// evidence in the same transaction. A live lease or terminal row fails fast.
func (repository *ExecutionRepository) ClaimConformance(
	ctx context.Context,
	executionID string,
	holderID string,
	lease time.Duration,
) (ExecutionClaim, error) {
	if repository == nil || repository.db == nil {
		return ExecutionClaim{}, fmt.Errorf("%w: execution repository is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return ExecutionClaim{}, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validOpaqueIdentifierWithin(executionID, connectorDigestProfileV2MaxIdentifierBytes) ||
		!validOpaqueIdentifierWithin(holderID, connectorDigestProfileV2MaxIdentifierBytes) {
		return ExecutionClaim{}, invalidExecutionCandidate("execution or claim-holder ID is not canonical")
	}
	leaseMicros := lease.Microseconds()
	if leaseMicros <= 0 || lease > maxExecutionLease {
		return ExecutionClaim{}, invalidExecutionCandidate("claim lease must be between one microsecond and %s", maxExecutionLease)
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return ExecutionClaim{}, err
	}
	observed, found, err := loadExecutionExact(ctx, repository.db, executionID, false)
	if err != nil {
		return ExecutionClaim{}, err
	}
	if !found {
		return ExecutionClaim{}, fmt.Errorf("%w: %s", ErrExecutionNotFound, executionID)
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return ExecutionClaim{}, fmt.Errorf("begin execution claim: %w", err)
	}
	rollback := func(cause error) (ExecutionClaim, error) {
		return ExecutionClaim{}, errors.Join(cause, tx.Rollback(), cleanup())
	}
	commit := func() error { return errors.Join(tx.Commit(), cleanup()) }

	catalogRevision, err := lockExecutionAuthority(ctx, tx, observed)
	if err != nil {
		return rollback(err)
	}
	locked, found, err := loadExecutionExact(ctx, tx, executionID, true)
	if err != nil {
		return rollback(err)
	}
	if !found || !locked.equalImmutable(observed) {
		return rollback(executionAuthorityConflict("execution changed after its authority tuple was selected"))
	}
	if locked.state != executionStateReady {
		return rollback(fmt.Errorf("%w: state=%s", ErrExecutionNotClaimable, locked.state))
	}
	var databaseNow time.Time
	if err := tx.QueryRowContext(ctx, `SELECT CURRENT_TIMESTAMP(6)`).Scan(&databaseNow); err != nil {
		return rollback(fmt.Errorf("read database time for execution claim: %w", err))
	}
	if locked.claimLeaseUntil != nil && locked.claimLeaseUntil.After(databaseNow) {
		return rollback(fmt.Errorf("%w: live holder=%s lease_until=%s",
			ErrExecutionNotClaimable, optionalExecutionString(locked.claimHolderID),
			locked.claimLeaseUntil.UTC().Format(time.RFC3339Nano)))
	}
	if locked.claimFenceToken == math.MaxInt64 {
		return rollback(storedExecutionInvalid("claim fence is exhausted"))
	}
	newFence := locked.claimFenceToken + 1
	leaseUntil := databaseNow.Add(time.Duration(leaseMicros) * time.Microsecond).Truncate(time.Microsecond)
	result, err := tx.ExecContext(ctx, `UPDATE _System_ConnectorV2Execution
SET ClaimFenceToken = ?, ClaimHolderId = ?, ClaimLeaseUntil = ?,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND BINARY Id = BINARY ?
  AND State = ? AND BINARY State = BINARY ?
  AND ClaimFenceToken = ?
  AND (ClaimLeaseUntil IS NULL OR ClaimLeaseUntil <= CURRENT_TIMESTAMP(6))`,
		newFence, holderID, leaseUntil,
		executionID, executionID,
		executionStateReady, executionStateReady,
		locked.claimFenceToken,
	)
	if err != nil {
		return rollback(fmt.Errorf("compare-and-swap execution claim: %w", err))
	}
	if err := requireOneExecutionRow(result, "execution claim"); err != nil {
		return rollback(fmt.Errorf("%w: %v", ErrExecutionNotClaimable, err))
	}
	attemptID := stableExecutionIdentity(
		"connector-v2-attempt",
		fmt.Sprintf("%s\x00%d", executionID, newFence),
	)
	result, err = tx.ExecContext(ctx, `INSERT INTO _System_ConnectorV2ExecutionAttempt
(Id, ExecutionId, AttemptOrdinal, ClaimFenceToken, ClaimHolderId, LeaseUntil)
VALUES (?, ?, ?, ?, ?, ?)`,
		attemptID, executionID, newFence, newFence, holderID, leaseUntil,
	)
	if err != nil {
		return rollback(fmt.Errorf("append execution claim attempt: %w", err))
	}
	if err := requireOneExecutionRow(result, "execution claim attempt"); err != nil {
		return rollback(err)
	}
	locked.claimFenceToken = newFence
	locked.claimHolderID = copyExecutionString(&holderID)
	locked.claimLeaseUntil = copyExecutionTime(&leaseUntil)
	if err := locked.validateStored(); err != nil {
		return rollback(err)
	}
	if err := commit(); err != nil {
		return ExecutionClaim{}, fmt.Errorf("commit execution claim: %w", err)
	}
	return ExecutionClaim{
		execution:       locked.copy(),
		holderID:        holderID,
		fenceToken:      newFence,
		leaseUntil:      leaseUntil,
		catalogRevision: catalogRevision,
	}, nil
}

// completeFromApplicationResult records Runtime's immutable terminal evidence.
// A claim fences command submission; it is not completion authority after
// Runtime has committed. Finalization therefore converges independently of a
// live claim while the original epoch remains the gate's predecessor authority.
func (repository *ExecutionRepository) completeFromApplicationResult(
	ctx context.Context,
	observed Execution,
	result applicationresultv1.ApplicationResult,
) (Execution, error) {
	if repository == nil || repository.db == nil {
		return Execution{}, fmt.Errorf("%w: execution repository is unavailable", ErrDependencyMissing)
	}
	if ctx == nil {
		return Execution{}, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if err := observed.validateStored(); err != nil {
		return Execution{}, fmt.Errorf("invalid execution result observation: %w", err)
	}
	if observed.state != executionStateReady && observed.state != executionStateCompleted {
		return Execution{}, fmt.Errorf("execution result observation has non-finalizable state %q", observed.state)
	}
	if err := applicationresultv1.ValidateApplicationResult(result); err != nil {
		return Execution{}, fmt.Errorf("invalid Runtime ApplicationResult: %w", err)
	}
	if result.CommandID != observed.runtimeCommandID ||
		result.ExecutionRefDigest != observed.executionRefDigest {
		return Execution{}, executionIdentityConflict("Runtime result names a different command or ExecutionRef")
	}
	if err := repository.VerifySchema(ctx); err != nil {
		return Execution{}, err
	}

	tx, cleanup, err := sqltx.BeginPessimisticConstraintCheckingTx(ctx, repository.db)
	if err != nil {
		return Execution{}, fmt.Errorf("begin execution completion: %w", err)
	}
	rollback := func(cause error) (Execution, error) {
		return Execution{}, errors.Join(cause, tx.Rollback(), cleanup())
	}
	commit := func() error { return errors.Join(tx.Commit(), cleanup()) }

	if err := lockExecutionResultFinalizationAuthority(ctx, tx, observed); err != nil {
		return rollback(err)
	}
	locked, found, err := loadExecutionExact(ctx, tx, observed.id, true)
	if err != nil {
		return rollback(err)
	}
	if !found || !locked.equalImmutable(observed) {
		return rollback(executionIdentityConflict("execution identity changed before result finalization"))
	}
	if locked.state == executionStateCompleted {
		if locked.resultDigest == nil || locked.runtimeCommitReceipt == nil ||
			*locked.resultDigest != result.ResultDigest ||
			*locked.runtimeCommitReceipt != result.RuntimeCommitReceipt {
			return rollback(executionIdentityConflict("completed Runtime correlation disagrees"))
		}
		if err := commit(); err != nil {
			return Execution{}, fmt.Errorf("commit exact execution completion read: %w", err)
		}
		return locked.copy(), nil
	}
	if locked.state != executionStateReady {
		return rollback(fmt.Errorf("execution state %q cannot record a Runtime result", locked.state))
	}
	if locked.claimFenceToken == 0 {
		return rollback(storedExecutionInvalid("Runtime result exists for an execution with no claim attempt"))
	}
	resultUpdate, err := tx.ExecContext(ctx, `UPDATE _System_ConnectorV2Execution
SET State = ?, ResultDigest = ?, RuntimeCommitReceipt = ?,
    ClaimHolderId = NULL, ClaimLeaseUntil = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND BINARY Id = BINARY ?
  AND State = ? AND BINARY State = BINARY ?
  AND ClaimFenceToken = ?`,
		executionStateCompleted, result.ResultDigest, result.RuntimeCommitReceipt,
		locked.id, locked.id,
		executionStateReady, executionStateReady,
		locked.claimFenceToken,
	)
	if err != nil {
		return rollback(fmt.Errorf("compare-and-swap execution completion: %w", err))
	}
	if err := requireOneExecutionRow(resultUpdate, "execution completion"); err != nil {
		return rollback(err)
	}
	locked.state = executionStateCompleted
	locked.resultDigest = copyExecutionString(&result.ResultDigest)
	locked.runtimeCommitReceipt = copyExecutionString(&result.RuntimeCommitReceipt)
	locked.claimHolderID = nil
	locked.claimLeaseUntil = nil
	if err := locked.validateStored(); err != nil {
		return rollback(err)
	}
	if err := commit(); err != nil {
		return Execution{}, fmt.Errorf("commit execution completion: %w", err)
	}
	return locked.copy(), nil
}

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

func loadExecutionExact(
	ctx context.Context,
	queryer executionQueryer,
	id string,
	forUpdate bool,
) (Execution, bool, error) {
	query := executionSelect
	if forUpdate {
		query += " FOR UPDATE"
	}
	var (
		execution  Execution
		holder     sql.NullString
		lease      sql.NullTime
		result     sql.NullString
		receipt    sql.NullString
		quarantine sql.NullString
	)
	err := queryer.QueryRowContext(ctx, query, id, id).Scan(
		&execution.id,
		&execution.executionProtocolVersion,
		&execution.servingScopeDigest,
		&execution.connectorAccountID,
		&execution.accountServingGateID,
		&execution.accountServingFenceToken,
		&execution.servingEpochID,
		&execution.servingEpochDigest,
		&execution.servingEpochBindingID,
		&execution.packGenerationID,
		&execution.definitionRef,
		&execution.definitionVersion,
		&execution.canonicalInputs,
		&execution.externalEffectsMode,
		&execution.inputDigest,
		&execution.executionRefDigest,
		&execution.runtimeCommandID,
		&execution.state,
		&execution.claimFenceToken,
		&holder,
		&lease,
		&result,
		&receipt,
		&quarantine,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return Execution{}, false, nil
	}
	if err != nil {
		return Execution{}, false, fmt.Errorf("load exact execution: %w", err)
	}
	if execution.id != id {
		return Execution{}, false, storedExecutionInvalid("indexed lookup returned a non-exact ID")
	}
	execution.canonicalInputs = append([]byte(nil), execution.canonicalInputs...)
	if holder.Valid {
		execution.claimHolderID = copyExecutionString(&holder.String)
	}
	if lease.Valid {
		execution.claimLeaseUntil = copyExecutionTime(&lease.Time)
	}
	if result.Valid {
		execution.resultDigest = copyExecutionString(&result.String)
	}
	if receipt.Valid {
		execution.runtimeCommitReceipt = copyExecutionString(&receipt.String)
	}
	if quarantine.Valid {
		execution.quarantineReasonCode = copyExecutionString(&quarantine.String)
	}
	if err := execution.validateStored(); err != nil {
		return Execution{}, false, err
	}
	return execution.copy(), true, nil
}

func insertExecution(ctx context.Context, tx *sql.Tx, execution Execution) error {
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_ConnectorV2Execution
(Id, ExecutionProtocolVersion, ServingScopeDigest, ConnectorAccountId,
 AccountServingGateId, AccountServingFenceToken,
 ServingEpochId, ServingEpochDigest, ServingEpochBindingId, PackGenerationId,
 DefinitionRef, DefinitionVersion, CanonicalInputs, ExternalEffectsMode,
 InputDigest, ExecutionRefDigest, RuntimeCommandId)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		execution.id,
		execution.executionProtocolVersion,
		execution.servingScopeDigest,
		execution.connectorAccountID,
		execution.accountServingGateID,
		execution.accountServingFenceToken,
		execution.servingEpochID,
		execution.servingEpochDigest,
		execution.servingEpochBindingID,
		execution.packGenerationID,
		execution.definitionRef,
		execution.definitionVersion,
		execution.canonicalInputs,
		execution.externalEffectsMode,
		execution.inputDigest,
		execution.executionRefDigest,
		execution.runtimeCommandID,
	)
	if err != nil {
		return err
	}
	return requireOneExecutionRow(result, "execution admission")
}

type executionBindingAuthority struct {
	id                          string
	servingEpochID              string
	kind                        catalogv1.IntegrationBindingKind
	ownerPack                   string
	packInstallationID          string
	packPayloadDigest           string
	businessKey                 string
	definitionVersion           uint64
	runtimeCommandKind          sql.NullString
	runtimePayloadFormatVersion sql.NullInt64
	runtimeDefinitionRef        sql.NullString
	runtimeDefinitionVersion    sql.NullString
	owner                       storedBindingOwnerProof
}

// lockExecutionResultFinalizationAuthority serializes result recording with a
// halt or successor transition. A higher fence is valid while the gate still
// names the execution's original active epoch in a V2 serving or halted mode.
// Once successor activation changes the epoch or drains the predecessor, local
// finalization loses the race and must not cross the OrderingHandoff boundary.
func lockExecutionResultFinalizationAuthority(
	ctx context.Context,
	tx *sql.Tx,
	execution Execution,
) error {
	gate, found, err := loadAccountServingGateExact(
		ctx,
		tx,
		execution.servingScopeDigest,
		execution.connectorAccountID,
		true,
	)
	if err != nil {
		return fmt.Errorf("lock execution result account gate: %w", err)
	}
	if !found {
		return executionAuthorityConflict("result account gate is absent")
	}
	if gate.id != execution.accountServingGateID {
		return executionAuthorityConflict("result account gate ID disagrees")
	}
	if gate.fenceToken < execution.accountServingFenceToken {
		return executionAuthorityConflict("result account gate fence trails the admitted execution")
	}
	switch gate.mode {
	case accountServingModeV2Serving, accountServingModeHaltingV2, accountServingModeV2Halted:
	default:
		return executionAuthorityConflict("result account gate mode %q forbids V2 finalization", gate.mode)
	}
	if gate.servingEpochID == nil || gate.servingEpochDigest == nil ||
		*gate.servingEpochID != execution.servingEpochID ||
		*gate.servingEpochDigest != execution.servingEpochDigest {
		return executionAuthorityConflict("result account gate no longer names the execution epoch")
	}

	header, found, err := loadAccountServingEpochHeaderExact(
		ctx,
		tx,
		execution.servingEpochID,
		true,
	)
	if err != nil {
		return fmt.Errorf("lock execution result epoch: %w", err)
	}
	if !found || header.id != execution.servingEpochID ||
		header.epochDigest != execution.servingEpochDigest ||
		header.servingScopeDigest != execution.servingScopeDigest ||
		header.activeScopeDigest == nil ||
		*header.activeScopeDigest != execution.servingScopeDigest ||
		header.state != accountServingEpochStateActive {
		return executionAuthorityConflict("result epoch is not the retained active predecessor")
	}
	if _, err := lockExecutionRetainedAuthority(ctx, tx, execution); err != nil {
		return err
	}
	return nil
}

// lockExecutionAuthority is the sole pre-write lock order: account gate,
// active epoch, exact binding/owner, PackGeneration parent, then definition-set
// header. It returns the immutable Runtime CatalogRevision.
func lockExecutionAuthority(
	ctx context.Context,
	tx *sql.Tx,
	execution Execution,
) (uint64, error) {
	gate, err := lockExpectedV2Serving(
		ctx,
		tx,
		execution.servingScopeDigest,
		execution.connectorAccountID,
		execution.accountServingFenceToken,
		execution.servingEpochID,
		execution.servingEpochDigest,
	)
	if err != nil {
		if errors.Is(err, errAccountServingGateNotFound) ||
			errors.Is(err, errAccountServingGateFenceConflict) ||
			errors.Is(err, errAccountServingGateModeConflict) ||
			errors.Is(err, errAccountServingGateEpochConflict) {
			return 0, executionAuthorityConflict("account gate: %v", err)
		}
		return 0, fmt.Errorf("lock execution account gate and epoch: %w", err)
	}
	if gate.id != execution.accountServingGateID {
		return 0, executionAuthorityConflict("account gate ID disagrees")
	}
	return lockExecutionRetainedAuthority(ctx, tx, execution)
}

// lockExecutionRetainedAuthority byte-compares the immutable epoch, binding,
// owner, PackGeneration, and definition-set authority after its caller has
// locked and classified the account gate and original epoch.
func lockExecutionRetainedAuthority(
	ctx context.Context,
	tx *sql.Tx,
	execution Execution,
) (uint64, error) {
	var (
		err                             error
		catalogRevision                 uint64
		profileID                       ConnectorDigestProfileID
		configurationGenerationProtocol serviceconfiguration.GenerationProtocolVersion
		configurationReceiptProtocol    serviceconfiguration.ReceiptProtocolVersion
	)
	if err := tx.QueryRowContext(ctx, `SELECT
e.CatalogRevision, r.ConnectorDigestProfileId,
r.ServiceConfigurationGenerationProtocolVersion,
r.ServiceConfigurationReceiptProtocolVersion
FROM _System_ConnectorServingEpoch AS e
JOIN _System_ConnectorServingEpochReproducibility AS r
  ON r.ServingEpochId = e.Id
 AND BINARY r.ServingEpochId = BINARY e.Id
WHERE e.Id = ? AND BINARY e.Id = BINARY ?
LIMIT 1
FOR UPDATE`, execution.servingEpochID, execution.servingEpochID).Scan(
		&catalogRevision,
		&profileID,
		&configurationGenerationProtocol,
		&configurationReceiptProtocol,
	); errors.Is(err, sql.ErrNoRows) {
		return 0, executionAuthorityConflict("active epoch lacks its exact Profile V2 reproducibility header")
	} else if err != nil {
		return 0, fmt.Errorf("lock execution epoch reproducibility header: %w", err)
	}
	if catalogRevision == 0 || profileID != ConnectorDigestProfileV2ID ||
		configurationGenerationProtocol != serviceconfiguration.GenerationProtocolV1 ||
		configurationReceiptProtocol != serviceconfiguration.ReceiptProtocolV1 {
		return 0, executionAuthorityConflict("active epoch reproducibility profile is invalid")
	}

	var binding executionBindingAuthority
	err = tx.QueryRowContext(ctx, `SELECT
b.Id, b.ServingEpochId, b.Kind, b.OwnerPack, b.PackInstallationId,
b.PackPayloadDigest, b.BusinessKey, b.DefinitionVersion,
b.RuntimeCommandKind, b.RuntimePayloadFormatVersion,
b.RuntimeDefinitionRef, b.RuntimeDefinitionVersion,
o.ServingEpochBindingId, o.PackGenerationId,
o.DefinitionEnvelopeVersion, o.DefinitionReceiptVersion,
o.CanonicalDefinitionDigest, o.DefinitionOwnerReceipt,
o.DefinitionOwnerReceiptDigest
FROM _System_ConnectorServingEpochBinding AS b
JOIN _System_ConnectorServingEpochBindingOwner AS o
  ON o.ServingEpochBindingId = b.Id
 AND BINARY o.ServingEpochBindingId = BINARY b.Id
WHERE b.Id = ? AND BINARY b.Id = BINARY ?
LIMIT 1
FOR UPDATE`, execution.servingEpochBindingID, execution.servingEpochBindingID).Scan(
		&binding.id,
		&binding.servingEpochID,
		&binding.kind,
		&binding.ownerPack,
		&binding.packInstallationID,
		&binding.packPayloadDigest,
		&binding.businessKey,
		&binding.definitionVersion,
		&binding.runtimeCommandKind,
		&binding.runtimePayloadFormatVersion,
		&binding.runtimeDefinitionRef,
		&binding.runtimeDefinitionVersion,
		&binding.owner.servingEpochBindingID,
		&binding.owner.packGenerationID,
		&binding.owner.definitionEnvelopeVersion,
		&binding.owner.definitionReceiptVersion,
		&binding.owner.canonicalDefinitionDigest,
		&binding.owner.definitionOwnerReceipt,
		&binding.owner.definitionOwnerReceiptDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, executionAuthorityConflict("exact serving binding or owner is absent")
	}
	if err != nil {
		return 0, fmt.Errorf("lock execution binding and owner: %w", err)
	}
	if binding.id != execution.servingEpochBindingID ||
		binding.owner.servingEpochBindingID != execution.servingEpochBindingID ||
		binding.servingEpochID != execution.servingEpochID ||
		binding.owner.packGenerationID != execution.packGenerationID ||
		!binding.runtimeCommandKind.Valid ||
		binding.runtimeCommandKind.String != flowcommandv3.CommandKind ||
		!binding.runtimePayloadFormatVersion.Valid ||
		binding.runtimePayloadFormatVersion.Int64 != int64(flowcommandv3.PayloadFormatVersion) ||
		!binding.runtimeDefinitionRef.Valid ||
		binding.runtimeDefinitionRef.String != execution.definitionRef ||
		!binding.runtimeDefinitionVersion.Valid ||
		binding.runtimeDefinitionVersion.String != execution.definitionVersion {
		return 0, executionAuthorityConflict("exact serving binding disagrees with execution")
	}
	ownerIdentity := connectorbindingownerv1.Identity{
		Kind:               binding.kind,
		OwnerPack:          binding.ownerPack,
		PackInstallationID: binding.packInstallationID,
		PackPayloadDigest:  binding.packPayloadDigest,
		BusinessKey:        binding.businessKey,
		DefinitionVersion:  binding.definitionVersion,
	}
	proof := BindingOwnerProof{
		identity:                     ownerIdentity,
		packGenerationID:             binding.owner.packGenerationID,
		definitionEnvelopeVersion:    binding.owner.definitionEnvelopeVersion,
		definitionReceiptVersion:     binding.owner.definitionReceiptVersion,
		canonicalDefinitionDigest:    binding.owner.canonicalDefinitionDigest,
		definitionOwnerReceipt:       binding.owner.definitionOwnerReceipt,
		definitionOwnerReceiptDigest: binding.owner.definitionOwnerReceiptDigest,
	}
	if err := validateBindingOwnerProof(proof); err != nil {
		return 0, executionAuthorityConflict("binding-owner proof is invalid: %v", err)
	}

	var (
		packGenerationID string
		packName         string
		packInstallID    string
		appliedHash      sql.NullString
	)
	err = tx.QueryRowContext(ctx, `SELECT Id, PackName, PackInstallId, AppliedPayloadHash
FROM _System_PackGeneration
WHERE Id = ? AND BINARY Id = BINARY ?
LIMIT 1
FOR UPDATE`, execution.packGenerationID, execution.packGenerationID).Scan(
		&packGenerationID,
		&packName,
		&packInstallID,
		&appliedHash,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, executionAuthorityConflict("exact PackGeneration parent is absent")
	}
	if err != nil {
		return 0, fmt.Errorf("lock execution PackGeneration parent: %w", err)
	}
	if packGenerationID != execution.packGenerationID || packName != binding.ownerPack ||
		packInstallID != binding.packInstallationID || !appliedHash.Valid ||
		appliedHash.String != binding.packPayloadDigest {
		return 0, executionAuthorityConflict("PackGeneration parent disagrees with binding owner")
	}

	var (
		setPackGenerationID string
		setProtocol         string
		definitionCountRaw  []byte
		definitionSetDigest string
	)
	err = tx.QueryRowContext(ctx, `SELECT PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest
FROM _System_PackGenerationConnectorBindingDefinitionSet
WHERE PackGenerationId = ? AND BINARY PackGenerationId = BINARY ?
LIMIT 1
FOR UPDATE`, execution.packGenerationID, execution.packGenerationID).Scan(
		&setPackGenerationID,
		&setProtocol,
		&definitionCountRaw,
		&definitionSetDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, executionAuthorityConflict("exact PackGeneration definition-set header is absent")
	}
	if err != nil {
		return 0, fmt.Errorf("lock execution PackGeneration definition set: %w", err)
	}
	definitionCount, parseErr := strconv.ParseUint(string(definitionCountRaw), 10, 64)
	if setPackGenerationID != execution.packGenerationID ||
		setProtocol != connectorbindingownerv1.DefinitionSetProtocolV1 ||
		parseErr != nil || definitionCount == 0 ||
		definitionCount > uint64(connectorbindingownerv1.MaxDefinitionCount) ||
		!prefixedSHA256Pattern.MatchString(definitionSetDigest) {
		return 0, executionAuthorityConflict("PackGeneration definition-set header is invalid")
	}
	return catalogRevision, nil
}

func requireOneExecutionRow(result sql.Result, operation string) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("%s rows affected: %w", operation, err)
	}
	if affected != 1 {
		return fmt.Errorf("%s rows affected = %d, want 1", operation, affected)
	}
	return nil
}

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

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

func optionalExecutionString(value *string) string {
	if value == nil {
		return "<none>"
	}
	return *value
}