package controlplane

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-controlplane/componentbehavior"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/connectorbindingdefinitionauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

const (
	firstPackActivationEpoch  = uint64(1)
	packGenerationPreparing   = "preparing"
	packGenerationActive      = "active"
	packGenerationRetired     = "retired"
	packGenerationFailedClean = "failed_clean"
	packGenerationFailedDirty = "failed_dirty"

	packGenerationCommitReconciliationTimeout = 30 * time.Second
)

var (
	ErrPackGenerationPrecondition = errors.New(
		"source-backed pack generation precondition failed",
	)
	ErrPackGenerationConflict = errors.New(
		"source-backed pack generation authority conflict",
	)
)

type sourceBackedPackGeneration struct {
	id                      string
	packInstallID           string
	packName                string
	category                string
	status                  string
	packVersion             sql.NullString
	manifestFormatVersion   sql.NullString
	platformCompatibility   sql.NullString
	desiredPayloadHash      sql.NullString
	appliedPayloadHash      sql.NullString
	payloadHashAlgorithm    sql.NullString
	sourceKind              sql.NullString
	sourceURI               sql.NullString
	sourceRevision          sql.NullString
	hostRequirementsDigest  sql.NullString
	runtimeObjectSpecDecl   sql.NullString
	runtimeRelatedLists     sql.NullString
	runtimeSearchScope      sql.NullString
	runtimeSearchAllDefault sql.NullString
	runtimeFormulaBundles   sql.NullString
	runtimeStartupShallow   sql.NullString
	lastError               sql.NullString
	isDeleted               int
}

type packActivation struct {
	id                  string
	packName            string
	activePackInstallID string
	// previousPackInstallID is reserved for a future explicitly designed
	// draining/rollback protocol. Stopped-upgrade authority requires it NULL.
	previousPackInstallID sql.NullString
	activationEpoch       uint64
	activatedDate         sql.NullTime
	isDeleted             int
}

type packGenerationCompatibility struct {
	packInstallID      string
	status             string
	isDeleted          int
	appliedPayloadHash sql.NullString
}

type sourceBackedPackGenerationAuthority struct {
	compatibility       packGenerationCompatibility
	compatibilityExists bool
	activePackInstallID string
}

// packGenerationActivationTransaction is the narrow transaction-bound store
// used by first-generation publication and exact verification.
type packGenerationActivationTransaction interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

// activatedSourceBackedPackRepairInstallID authorizes an ordinary source-backed
// apply before it mutates pack-owned state. A genuinely fresh install may mint
// an incarnation. A pre-existing active row without immutable authority is a
// stopped-upgrade adoption and must use publishFirstSourceBackedPackGeneration;
// ordinary apply/reapply must not replace its incarnation. With authority, only
// byte-for-byte/semantic equality across InstalledPack, Activation, and
// Generation is a repair and reuses that generation's PackInstallId.
func activatedSourceBackedPackRepairInstallID(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
) (string, error) {
	authority, err := classifySourceBackedPackGenerationAuthority(ctx, db, applied)
	if err != nil {
		return "", err
	}
	if authority.activePackInstallID == "" &&
		authority.compatibilityExists &&
		authority.compatibility.isDeleted == 0 &&
		authority.compatibility.status == installedPackStatusActive {
		return "", fmt.Errorf(
			"%w: active pack %q has no immutable generation authority; run explicit stopped-upgrade adoption before apply",
			ErrPackGenerationConflict,
			applied.pack.Name(),
		)
	}
	return authority.activePackInstallID, nil
}

func sourceBackedPackInstallIDForExactSourceAuthorityAdoption(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
	admittedPackInstallID string,
) (string, error) {
	authority, err := classifySourceBackedPackGenerationAuthority(ctx, db, applied)
	if err != nil {
		return "", err
	}
	admittedPackInstallID = strings.TrimSpace(admittedPackInstallID)
	if authority.activePackInstallID != "" ||
		!authority.compatibilityExists ||
		authority.compatibility.isDeleted != 0 ||
		authority.compatibility.status != installedPackStatusActive ||
		strings.TrimSpace(authority.compatibility.packInstallID) == "" ||
		strings.TrimSpace(authority.compatibility.packInstallID) != admittedPackInstallID {
		return "", fmt.Errorf(
			"%w: pack %q is not the exact admitted source-authority adoption incarnation %q",
			ErrPackGenerationConflict,
			applied.pack.Name(),
			admittedPackInstallID,
		)
	}
	if !authority.compatibility.appliedPayloadHash.Valid ||
		strings.TrimSpace(authority.compatibility.appliedPayloadHash.String) == "" {
		return "", fmt.Errorf(
			"%w: pack %q source-authority adoption incarnation %q has no applied payload identity",
			ErrPackGenerationConflict,
			applied.pack.Name(),
			admittedPackInstallID,
		)
	}
	if _, err := packs.ValidateStoredPackPayloadHash(
		applied.pack,
		authority.compatibility.appliedPayloadHash.String,
	); err != nil {
		return "", fmt.Errorf(
			"%w: pack %q source-authority adoption incarnation %q payload identity differs from the resolved source: %v",
			ErrPackGenerationConflict,
			applied.pack.Name(),
			admittedPackInstallID,
			err,
		)
	}
	return admittedPackInstallID, nil
}

// classifySourceBackedPackGenerationAuthority reads the complete authority
// tuple under one transaction. Lock ordering matches activation publication:
// InstalledPack -> Activation -> Generation. Absence of authority is returned
// as a classification, while partial or mixed authority always fails closed.
func classifySourceBackedPackGenerationAuthority(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
) (sourceBackedPackGenerationAuthority, error) {
	if ctx == nil || db == nil {
		return sourceBackedPackGenerationAuthority{}, fmt.Errorf(
			"%w: context and database are required",
			ErrPackGenerationPrecondition,
		)
	}
	expected, err := sourceBackedPackGenerationFromVerifiedPack(applied)
	if err != nil {
		return sourceBackedPackGenerationAuthority{}, err
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return sourceBackedPackGenerationAuthority{}, fmt.Errorf("begin pack %q generation classification: %w", expected.packName, err)
	}
	defer func() { _ = tx.Rollback() }()

	compatibility, compatibilityExists, err := readPackGenerationCompatibilityForUpdate(
		ctx,
		tx,
		expected.packName,
	)
	if err != nil {
		return sourceBackedPackGenerationAuthority{}, err
	}
	authority := sourceBackedPackGenerationAuthority{
		compatibility:       compatibility,
		compatibilityExists: compatibilityExists,
	}
	activation, exists, err := readPackActivationForUpdate(ctx, tx, expected.packName)
	if err != nil {
		return sourceBackedPackGenerationAuthority{}, err
	}
	if !exists {
		installIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(ctx, tx, expected.packName)
		if err != nil {
			return sourceBackedPackGenerationAuthority{}, err
		}
		if len(installIDs) != 0 {
			return sourceBackedPackGenerationAuthority{}, fmt.Errorf(
				"%w: pack %q has generation %q without its activation",
				ErrPackGenerationConflict,
				expected.packName,
				installIDs[0],
			)
		}
		return authority, nil
	}

	expected.packInstallID = strings.TrimSpace(activation.activePackInstallID)
	expected.id = packGenerationID(expected.packName, expected.packInstallID)
	if expected.packInstallID == "" {
		return sourceBackedPackGenerationAuthority{}, fmt.Errorf(
			"%w: pack %q activation has a blank generation identity",
			ErrPackGenerationConflict,
			expected.packName,
		)
	}
	if !compatibilityExists ||
		compatibility.isDeleted != 0 ||
		strings.TrimSpace(compatibility.packInstallID) != expected.packInstallID {
		return sourceBackedPackGenerationAuthority{}, fmt.Errorf(
			"%w: pack %q InstalledPack incarnation %q does not match active generation %q",
			ErrPackGenerationConflict,
			expected.packName,
			compatibility.packInstallID,
			expected.packInstallID,
		)
	}
	if err := verifyExistingPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		applied,
		expected,
		activation,
	); err != nil {
		return sourceBackedPackGenerationAuthority{}, fmt.Errorf(
			"%w: pack %q apply is not an exact repair of its active generation: %v",
			ErrPackGenerationConflict,
			expected.packName,
			err,
		)
	}
	authority.activePackInstallID = expected.packInstallID
	return authority, nil
}

func loadActivePackGenerationRuntimeForInstalledRow(
	ctx context.Context,
	db *sql.DB,
	packName string,
	legacyConnectorProjectionSource packs.Pack,
) (packs.Pack, error) {
	expected, err := loadActivePackGenerationForInstalledRow(ctx, db, packName)
	if err != nil {
		return nil, err
	}
	declaration := packs.PersistedRuntimeDeclarationJSON{
		ObjectSpecs:            expected.runtimeObjectSpecDecl.String,
		RelatedLists:           expected.runtimeRelatedLists.String,
		SearchScope:            expected.runtimeSearchScope.String,
		SearchAllDefaultFields: expected.runtimeSearchAllDefault.String,
		FormulaFunctionBundles: expected.runtimeFormulaBundles.String,
		StartupShallow:         expected.runtimeStartupShallow.String,
	}
	runtimePack, err := declaration.LoadPack(expected.packName)
	if err != nil && legacyConnectorProjectionSource != nil {
		runtimePack, err = declaration.LoadPackWithLegacyConnectorCalloutProfileProjectionSource(
			expected.packName,
			legacyConnectorProjectionSource,
		)
	}
	if err != nil {
		return nil, fmt.Errorf("%w: pack %q active generation Runtime declaration is malformed: %v", ErrPackGenerationConflict, expected.packName, err)
	}
	return runtimePack, nil
}

func loadActivePackGenerationObjectOwnershipForInstalledRow(
	ctx context.Context,
	db *sql.DB,
	packName string,
) ([]string, error) {
	expected, err := loadActivePackGenerationForInstalledRow(ctx, db, packName)
	if err != nil {
		return nil, err
	}
	return packGenerationObjectOwnership(expected)
}

func loadActivePackGenerationForInstalledRow(
	ctx context.Context,
	db *sql.DB,
	packName string,
) (sourceBackedPackGeneration, error) {
	packName = strings.TrimSpace(packName)
	if ctx == nil || db == nil || packName == "" {
		return sourceBackedPackGeneration{}, fmt.Errorf("%w: context, database, and installed pack name are required", ErrPackGenerationPrecondition)
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return sourceBackedPackGeneration{}, fmt.Errorf("begin pack %q InstalledPack/generation comparison: %w", packName, err)
	}
	defer func() { _ = tx.Rollback() }()

	expected, exists, err := readInstalledPackGenerationPreimageForUpdate(ctx, tx, packName)
	if err != nil {
		return sourceBackedPackGeneration{}, err
	}
	if !exists {
		return sourceBackedPackGeneration{}, fmt.Errorf("%w: pack %q active InstalledPack row is missing", ErrPackGenerationConflict, packName)
	}
	activation, exists, err := readPackActivationForUpdate(ctx, tx, packName)
	if err != nil {
		return sourceBackedPackGeneration{}, err
	}
	if !exists || expected.packInstallID != activation.activePackInstallID {
		return sourceBackedPackGeneration{}, fmt.Errorf("%w: pack %q active InstalledPack incarnation does not match its activation", ErrPackGenerationConflict, packName)
	}
	if err := verifyExistingPackGeneration(ctx, tx, expected, activation); err != nil {
		return sourceBackedPackGeneration{}, fmt.Errorf("%w: pack %q InstalledPack runtime declaration does not match its active generation: %v", ErrPackGenerationConflict, packName, err)
	}
	return expected, nil
}

func packGenerationObjectOwnership(generation sourceBackedPackGeneration) ([]string, error) {
	runtimePack, err := packs.LoadPersistedRuntimePack(
		generation.packName,
		generation.runtimeObjectSpecDecl.String,
		"[]",
		"[]",
	)
	if err != nil {
		return nil, fmt.Errorf(
			"%w: pack %q active generation object ownership declaration is malformed: %v",
			ErrPackGenerationConflict,
			generation.packName,
			err,
		)
	}
	return runtimePack.OwnedObjectAPINames(), nil
}

func verifySourceBackedPackGenerationAuthorityForInstalledRow(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
	row installedPackRow,
) (string, error) {
	authority, err := classifySourceBackedPackGenerationAuthority(ctx, db, applied)
	if err != nil {
		return "", err
	}
	if !authority.compatibilityExists ||
		authority.compatibility.packInstallID != strings.TrimSpace(row.PackInstallID) ||
		authority.compatibility.status != strings.TrimSpace(row.Status) ||
		authority.compatibility.isDeleted != row.IsDeleted {
		return "", fmt.Errorf(
			"%w: pack %q InstalledPack lifecycle/incarnation changed during generation preflight",
			ErrPackGenerationConflict,
			row.Name,
		)
	}
	return authority.activePackInstallID, nil
}

// verifySourceRevisionAdvanceGenerationAuthority verifies the immutable side of
// one ApplyAll revision edge without requiring the mutable InstalledPack row to
// keep the old incarnation during retry. A failed/installing retry row already
// names the target revision, while PackActivation intentionally continues to
// select the last committed generation until finalization. An active transition
// validates G1 as stored and binds its source identity to the revision-bound
// current source; it must not replay current compiler projections into G1.
func verifySourceRevisionAdvanceGenerationAuthority(
	ctx context.Context,
	db *sql.DB,
	advance normalizedApplyAllSourceRevisionAdvance,
	current *packs.SourceBackedPack,
	target *packs.SourceBackedPack,
	row installedPackRow,
	requiresTransition bool,
) error {
	if ctx == nil || db == nil || current == nil || target == nil {
		return fmt.Errorf(
			"%w: source-revision generation preflight requires context, database, current pack, and target pack",
			ErrPackGenerationPrecondition,
		)
	}
	payloadHash, err := packs.PackPayloadHash(current)
	if err != nil {
		return fmt.Errorf("build current source-revision generation payload hash: %w", err)
	}
	expected, err := sourceBackedPackGenerationFromVerifiedPack(appliedAppPack{
		pack: current, payloadHash: payloadHash,
	})
	if err != nil {
		return err
	}

	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin pack %q source-revision generation preflight: %w", advance.Name, err)
	}
	defer func() { _ = tx.Rollback() }()

	compatibility, exists, err := readPackGenerationCompatibilityForUpdate(
		ctx, tx, advance.Name,
	)
	if err != nil {
		return err
	}
	if !exists ||
		compatibility.packInstallID != strings.TrimSpace(row.PackInstallID) ||
		compatibility.status != strings.TrimSpace(row.Status) ||
		compatibility.isDeleted != row.IsDeleted ||
		compatibility.appliedPayloadHash != row.AppliedPayloadHash {
		return fmt.Errorf(
			"%w: pack %q InstalledPack lifecycle/incarnation changed during source-revision generation preflight",
			ErrPackGenerationConflict,
			advance.Name,
		)
	}

	activation, activationExists, err := readPackActivationForUpdate(ctx, tx, advance.Name)
	if err != nil {
		return err
	}
	if !activationExists {
		installIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(ctx, tx, advance.Name)
		if err != nil {
			return err
		}
		if len(installIDs) != 0 {
			return fmt.Errorf(
				"%w: pack %q has generation %q without its activation",
				ErrPackGenerationConflict,
				advance.Name,
				installIDs[0],
			)
		}
		return fmt.Errorf(
			"%w: pack %q has no immutable generation authority for source revision %q; publish that exact current revision before advancing",
			ErrPackGenerationConflict,
			advance.Name,
			advance.FromRevision,
		)
	}

	switch row.Status {
	case installedPackStatusActive:
		if row.PackInstallID != activation.activePackInstallID {
			return fmt.Errorf(
				"%w: pack %q active InstalledPack incarnation %q differs from generation activation %q",
				ErrPackGenerationConflict,
				advance.Name,
				row.PackInstallID,
				activation.activePackInstallID,
			)
		}
		expected.packInstallID = activation.activePackInstallID
		expected.id = packGenerationID(expected.packName, expected.packInstallID)
		if !requiresTransition {
			return verifyExistingPackGenerationConnectorBindingDefinitionSet(
				ctx,
				tx,
				appliedAppPack{pack: current, payloadHash: payloadHash},
				expected,
				activation,
			)
		}
		active, err := verifyActivePackGeneration(ctx, tx, activation)
		if err != nil {
			return err
		}
		if active.appliedPayloadHash != compatibility.appliedPayloadHash {
			return fmt.Errorf(
				"%w: pack %q InstalledPack payload identity differs from its active generation",
				ErrPackGenerationConflict,
				advance.Name,
			)
		}
		if !sourceBackedPackGenerationNonRuntimeEqual(active, expected) {
			return fmt.Errorf(
				"%w: pack %q active generation immutable source preimage does not match the revision-bound current source",
				ErrPackGenerationConflict,
				advance.Name,
			)
		}
		if err := verifyGenerationMatchesSourceRevisionAdvanceFrom(active, advance); err != nil {
			return err
		}
		return verifySourceRevisionAdvanceGenerationTransitionPreconditions(
			advance,
			activation,
			active.category,
			string(target.Category()),
		)

	case installedPackStatusFailed, installedPackStatusInstalling:
		active, err := verifyActivePackGeneration(ctx, tx, activation)
		if err != nil {
			return err
		}
		expected.packInstallID = activation.activePackInstallID
		expected.id = packGenerationID(expected.packName, expected.packInstallID)
		if !sourceBackedPackGenerationNonRuntimeEqual(active, expected) {
			return fmt.Errorf(
				"%w: pack %q active generation immutable source preimage does not match the retry replay base",
				ErrPackGenerationConflict,
				advance.Name,
			)
		}
		if err := verifyGenerationMatchesSourceRevisionAdvanceFrom(active, advance); err != nil {
			return err
		}
		if requiresTransition {
			if err := verifySourceRevisionAdvanceGenerationTransitionPreconditions(
				advance,
				activation,
				active.category,
				string(target.Category()),
			); err != nil {
				return err
			}
		}
		if row.PackInstallID == activation.activePackInstallID {
			return fmt.Errorf(
				"%w: pack %q retry row reuses the active immutable generation identity",
				ErrPackGenerationConflict,
				advance.Name,
			)
		}
		if existing, found, err := readPackGenerationForUpdate(
			ctx, tx, row.PackInstallID,
		); err != nil {
			return err
		} else if found {
			return fmt.Errorf(
				"%w: pack %q retry incarnation already has generation %q",
				ErrPackGenerationConflict,
				advance.Name,
				existing.id,
			)
		}
		return nil

	default:
		return fmt.Errorf(
			"%w: pack %q status %q cannot pass source-revision generation preflight",
			ErrPackGenerationConflict,
			advance.Name,
			row.Status,
		)
	}
}

func verifySourceRevisionAdvanceGenerationTransitionPreconditions(
	advance normalizedApplyAllSourceRevisionAdvance,
	activation packActivation,
	activeCategory string,
	targetCategory string,
) error {
	if activeCategory != targetCategory {
		return fmt.Errorf(
			"%w: pack %q source-revision advance changes category from %q to %q",
			ErrPackGenerationConflict,
			advance.Name,
			activeCategory,
			targetCategory,
		)
	}
	if activation.activationEpoch == ^uint64(0) {
		return fmt.Errorf(
			"%w: pack %q activation epoch is exhausted",
			ErrPackGenerationConflict,
			advance.Name,
		)
	}
	return nil
}

// publishFirstSourceBackedPackGeneration is the stopped-upgrade adoption path
// for a pre-existing legacy active row that predates generation authority. New
// installs and exact repairs publish authority in their active transaction.
// On a commit error this adoption transaction reconciles by rereading the exact
// authority tuple; it never compensates the already-active compatibility row.
func publishFirstSourceBackedPackGeneration(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
) error {
	if ctx == nil || db == nil {
		return fmt.Errorf(
			"%w: context and database are required",
			ErrPackGenerationPrecondition,
		)
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin first source-backed generation publication: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()
	if err := commitFirstSourceBackedPackGeneration(ctx, tx, applied); err != nil {
		return err
	}
	if err := tx.Commit(); err == nil {
		committed = true
		return nil
	} else {
		commitErr := err
		reconcileCtx, cancel := context.WithTimeout(
			context.WithoutCancel(ctx),
			packGenerationCommitReconciliationTimeout,
		)
		verifyErr := verifyPublishedSourceBackedPackGeneration(reconcileCtx, db, applied)
		cancel()
		if verifyErr == nil {
			committed = true
			return nil
		}
		return fmt.Errorf(
			"publish first source-backed generation commit is indeterminate: %w",
			errors.Join(commitErr, verifyErr),
		)
	}
}

func verifyPublishedSourceBackedPackGeneration(
	ctx context.Context,
	db *sql.DB,
	applied appliedAppPack,
) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin published generation verification: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	if err := verifyExistingSourceBackedPackGeneration(ctx, tx, applied); err != nil {
		return err
	}
	return nil
}

// commitFirstSourceBackedPackGeneration is the transaction body for
// first-generation publication. New installs and exact repairs call it inside
// the same transaction that activates InstalledPack; stopped-upgrade adoption
// calls it from its own transaction against a pre-existing active row. The
// mutable row contributes the active PackInstallId binding and convergence
// proof, while every immutable payload fact comes from the exact verified
// in-memory pack; no repository is resolved here. Only revision-addressed
// official_repo sources are publication-eligible. local_path remains an
// operator-owned mutable development source and cannot become Catalog
// generation authority until a separate staged-replacement lifecycle exists.
//
// An exact retry is accepted only when every immutable field and the activation
// pointer already match. Source-revision replacement is owned by
// commitSourceRevisionAdvancePackGeneration; this first-publication entry point
// never changes an established activation.
func commitFirstSourceBackedPackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	applied appliedAppPack,
) error {
	generation, err := sourceBackedPackGenerationForActivation(ctx, tx, applied)
	if err != nil {
		return err
	}

	activation, exists, err := readPackActivationForUpdate(
		ctx,
		tx,
		generation.packName,
	)
	if err != nil {
		return err
	}
	if exists {
		return verifyExistingPackGenerationConnectorBindingDefinitionSet(
			ctx,
			tx,
			applied,
			generation,
			activation,
		)
	}

	existingInstallIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(
		ctx,
		tx,
		generation.packName,
	)
	if err != nil {
		return err
	}
	if len(existingInstallIDs) != 0 {
		return fmt.Errorf(
			"%w: pack %q has generation %q without its activation",
			ErrPackGenerationConflict,
			generation.packName,
			existingInstallIDs[0],
		)
	}
	definitionSet, err := sourceBackedPackConnectorBindingDefinitionSet(ctx, tx, applied, generation)
	if err != nil {
		return err
	}

	now := time.Now().UTC()
	if err := insertPackGeneration(ctx, tx, generation, now); err != nil {
		return err
	}
	if err := insertPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		generation,
		definitionSet,
		now,
	); err != nil {
		return err
	}
	if err := verifyInsertedPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		generation,
		definitionSet,
	); err != nil {
		return err
	}
	if err := insertPackActivation(ctx, tx, generation, now); err != nil {
		return err
	}
	return nil
}

func preparingStoppedSourceSetPackGeneration(
	pack packs.Pack,
	payloadHash string,
	packInstallID string,
) (sourceBackedPackGeneration, error) {
	generation, err := sourceBackedPackGenerationFromVerifiedPack(appliedAppPack{
		pack: pack, payloadHash: payloadHash,
	})
	if err != nil {
		return sourceBackedPackGeneration{}, err
	}
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: stopped source-set candidate has no installation identity",
			ErrPackGenerationPrecondition,
		)
	}
	generation.id = packGenerationID(generation.packName, packInstallID)
	generation.packInstallID = packInstallID
	generation.status = packGenerationPreparing
	generation.appliedPayloadHash = sql.NullString{}
	return generation, nil
}

func markStoppedSourceSetPackGenerationFailed(
	ctx context.Context,
	db *sql.DB,
	packName string,
	packInstallID string,
	dirty bool,
	applyErr error,
) error {
	if ctx == nil || db == nil || applyErr == nil {
		return nil
	}
	status := packGenerationFailedClean
	if dirty {
		status = packGenerationFailedDirty
	}
	message := applyErr.Error()
	if len(message) > 4000 {
		message = message[:4000]
	}
	result, err := db.ExecContext(ctx, `UPDATE _System_PackGeneration
SET Status = ?, LastError = ?, LastModifiedDate = ?
WHERE PackName = ?
  AND PackInstallId = ?
  AND Status = ?
  AND COALESCE(IsDeleted, 0) = 0`,
		status,
		message,
		time.Now().UTC(),
		strings.TrimSpace(packName),
		strings.TrimSpace(packInstallID),
		packGenerationPreparing,
	)
	if err != nil {
		return fmt.Errorf("mark stopped source-set pack %q candidate failed: %w", packName, err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("mark stopped source-set pack %q candidate failed rows affected: %w", packName, err)
	}
	if affected != 1 {
		var currentStatus string
		readErr := db.QueryRowContext(ctx, `SELECT Status
FROM _System_PackGeneration
WHERE PackName = ? AND PackInstallId = ? AND COALESCE(IsDeleted, 0) = 0
LIMIT 1`, strings.TrimSpace(packName), strings.TrimSpace(packInstallID)).Scan(&currentStatus)
		if readErr == nil && (currentStatus == packGenerationActive ||
			currentStatus == packGenerationFailedClean ||
			currentStatus == packGenerationFailedDirty) {
			return nil
		}
		return fmt.Errorf(
			"%w: stopped source-set pack %q candidate %q is no longer preparing",
			ErrPackGenerationConflict,
			packName,
			packInstallID,
		)
	}
	return nil
}

func verifyPreparingStoppedSourceSetGeneration(
	actual sourceBackedPackGeneration,
	expected sourceBackedPackGeneration,
) error {
	if actual.status != packGenerationPreparing ||
		actual.appliedPayloadHash.Valid ||
		actual.lastError.Valid ||
		actual.isDeleted != 0 {
		return fmt.Errorf(
			"%w: stopped source-set pack %q candidate %q is not a clean preparing generation",
			ErrPackGenerationConflict,
			expected.packName,
			expected.packInstallID,
		)
	}
	actual.status = packGenerationActive
	actual.appliedPayloadHash = expected.appliedPayloadHash
	if !sourceBackedPackGenerationsEqual(actual, expected) {
		return fmt.Errorf(
			"%w: stopped source-set pack %q candidate %q differs from its prepared generation",
			ErrPackGenerationConflict,
			expected.packName,
			expected.packInstallID,
		)
	}
	return nil
}

// commitStoppedSourceSetPackGeneration activates a separately prepared
// immutable generation after Runtime has been stopped. The candidate attempt
// is recorded in PackGeneration and never overwrites the active InstalledPack
// compatibility projection while it is preparing. This transaction promotes
// the candidate, switches PackActivation, retires the old generation, and is
// committed together with the compatibility projection update.
func commitStoppedSourceSetPackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	applied appliedAppPack,
	previousPackInstallID string,
	candidatePackInstallID string,
) error {
	previousPackInstallID = strings.TrimSpace(previousPackInstallID)
	candidatePackInstallID = strings.TrimSpace(candidatePackInstallID)
	if previousPackInstallID == "" || candidatePackInstallID == "" {
		return fmt.Errorf(
			"%w: stopped source-set replacement has incomplete previous/candidate installation identity",
			ErrPackGenerationPrecondition,
		)
	}
	next, err := sourceBackedPackGenerationForActivation(ctx, tx, applied)
	if err != nil {
		return err
	}
	if next.packInstallID != candidatePackInstallID {
		return fmt.Errorf(
			"%w: stopped source-set pack %q compatibility projection selected candidate %q, want %q",
			ErrPackGenerationConflict,
			next.packName,
			next.packInstallID,
			candidatePackInstallID,
		)
	}
	prepared, found, err := readPackGenerationForUpdate(ctx, tx, candidatePackInstallID)
	if err != nil {
		return err
	}
	if !found {
		return fmt.Errorf(
			"%w: stopped source-set pack %q prepared generation %q is missing",
			ErrPackGenerationConflict,
			next.packName,
			candidatePackInstallID,
		)
	}
	if err := verifyPreparingStoppedSourceSetGeneration(prepared, next); err != nil {
		return err
	}
	activation, exists, err := readPackActivationForUpdate(ctx, tx, next.packName)
	if err != nil {
		return err
	}
	if exists && strings.TrimSpace(activation.activePackInstallID) != previousPackInstallID {
		return fmt.Errorf(
			"%w: pack %q active generation changed from admitted installation %q to %q",
			ErrPackGenerationPrecondition,
			next.packName,
			previousPackInstallID,
			activation.activePackInstallID,
		)
	}
	if exists && next.packInstallID == activation.activePackInstallID {
		return fmt.Errorf(
			"%w: pack %q stopped replacement did not mint a fresh incarnation",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	definitionSet, err := sourceBackedPackConnectorBindingDefinitionSet(
		ctx, tx, applied, next,
	)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	if err := insertPackGenerationConnectorBindingDefinitionSet(
		ctx, tx, next, definitionSet, now,
	); err != nil {
		return err
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_PackGeneration
SET Status = ?, AppliedPayloadHash = DesiredPayloadHash,
    LastError = NULL, LastModifiedDate = ?
WHERE PackName = ?
  AND PackInstallId = ?
  AND Status = ?
  AND AppliedPayloadHash IS NULL
  AND COALESCE(IsDeleted, 0) = 0`,
		packGenerationActive,
		now,
		next.packName,
		next.packInstallID,
		packGenerationPreparing,
	)
	if err != nil {
		return fmt.Errorf("activate pack %q prepared generation: %w", next.packName, err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return fmt.Errorf(
			"%w: pack %q prepared generation changed before activation",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	if !exists {
		return insertPackActivation(ctx, tx, next, now)
	}
	previous, err := verifyActivePackGeneration(ctx, tx, activation)
	if err != nil {
		return fmt.Errorf(
			"%w: pack %q active generation: %v",
			ErrPackGenerationPrecondition,
			next.packName,
			err,
		)
	}
	if previous.packName != next.packName || previous.category != next.category {
		return fmt.Errorf(
			"%w: pack %q active generation category is %q, want %q",
			ErrPackGenerationPrecondition,
			next.packName,
			previous.category,
			next.category,
		)
	}
	if activation.activationEpoch == ^uint64(0) {
		return fmt.Errorf(
			"%w: pack %q activation epoch is exhausted",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	result, err = tx.ExecContext(ctx, `UPDATE _System_PackActivation
SET PreviousPackInstallId = NULL,
    ActivePackInstallId = ?,
    ActivationEpoch = ?,
    ActivatedDate = ?,
    LastModifiedDate = ?,
    IsDeleted = 0
WHERE Id = ?
  AND PackName = ?
  AND ActivePackInstallId = ?
  AND ActivationEpoch = ?
  AND COALESCE(IsDeleted, 0) = 0`,
		next.packInstallID,
		activation.activationEpoch+1,
		now,
		now,
		activation.id,
		activation.packName,
		activation.activePackInstallID,
		activation.activationEpoch,
	)
	if err != nil {
		return fmt.Errorf("activate pack %q stopped replacement generation: %w", next.packName, err)
	}
	affected, err = result.RowsAffected()
	if err != nil {
		return fmt.Errorf("activate pack %q stopped replacement rows affected: %w", next.packName, err)
	}
	if affected != 1 {
		return fmt.Errorf(
			"%w: pack %q activation changed during stopped replacement",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	retired, err := tx.ExecContext(ctx, `UPDATE _System_PackGeneration
SET Status = ?, LastModifiedDate = ?
WHERE PackName = ?
  AND PackInstallId = ?
  AND Status = ?
  AND COALESCE(IsDeleted, 0) = 0`,
		packGenerationRetired,
		now,
		next.packName,
		previous.packInstallID,
		packGenerationActive,
	)
	if err != nil {
		return fmt.Errorf("retire pack %q previous generation: %w", next.packName, err)
	}
	retiredRows, err := retired.RowsAffected()
	if err != nil || retiredRows != 1 {
		return fmt.Errorf(
			"%w: pack %q previous generation changed before retirement",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	return nil
}

// commitSourceRevisionAdvancePackGeneration publishes the fresh immutable
// generation created by one authorized ApplyAll source-revision edge and moves
// the pack's activation pointer in the same transaction. Existing generations
// are never updated or deleted.
//
// The InstalledPack row has already been changed to the fresh active
// incarnation when this function runs. The existing PackActivation therefore
// remains the sole old-generation authority until the new generation insert and
// pointer move commit together. This transaction body performs only the
// FromRevision -> ToRevision transition; already-committed retry and ambiguous
// commit classification belong to the outer ApplyAll execution owner.
func commitSourceRevisionAdvancePackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	advance normalizedApplyAllSourceRevisionAdvance,
	applied appliedAppPack,
) error {
	next, err := sourceBackedPackGenerationForActivation(ctx, tx, applied)
	if err != nil {
		return err
	}
	if next.packName != advance.Name {
		return fmt.Errorf(
			"%w: generation pack %q does not match source-revision advance %q",
			ErrPackGenerationPrecondition,
			next.packName,
			advance.Name,
		)
	}
	if next.sourceKind.String != advance.Kind ||
		next.sourceURI.String != advance.URI ||
		next.sourceRevision.String != advance.ToRevision {
		return fmt.Errorf(
			"%w: generation pack %q is not authorized target source revision %q",
			ErrPackGenerationPrecondition,
			next.packName,
			advance.ToRevision,
		)
	}

	activation, exists, err := readPackActivationForUpdate(ctx, tx, next.packName)
	if err != nil {
		return err
	}
	if !exists {
		existingInstallIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(
			ctx, tx, next.packName,
		)
		if err != nil {
			return err
		}
		if len(existingInstallIDs) != 0 {
			return fmt.Errorf(
				"%w: pack %q has generation %q without its activation",
				ErrPackGenerationConflict,
				next.packName,
				existingInstallIDs[0],
			)
		}
		return fmt.Errorf(
			"%w: pack %q cannot publish target source revision %q without an active generation proving source revision %q",
			ErrPackGenerationConflict,
			next.packName,
			advance.ToRevision,
			advance.FromRevision,
		)
	}

	if activation.activationEpoch == ^uint64(0) {
		return fmt.Errorf(
			"%w: pack %q activation epoch is exhausted",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	active, err := verifyActivePackGeneration(ctx, tx, activation)
	if err != nil {
		return err
	}
	if err := verifyGenerationMatchesSourceRevisionAdvanceFrom(active, advance); err != nil {
		return err
	}
	if active.category != next.category {
		return fmt.Errorf(
			"%w: pack %q replacement changes category from %q to %q",
			ErrPackGenerationConflict,
			next.packName,
			active.category,
			next.category,
		)
	}
	if existing, found, err := readPackGenerationForUpdate(
		ctx, tx, next.packInstallID,
	); err != nil {
		return err
	} else if found {
		return fmt.Errorf(
			"%w: fresh pack incarnation %q already has generation %q",
			ErrPackGenerationConflict,
			next.packInstallID,
			existing.id,
		)
	}
	definitionSet, err := sourceBackedPackConnectorBindingDefinitionSet(ctx, tx, applied, next)
	if err != nil {
		return err
	}

	now := time.Now().UTC()
	if err := insertPackGeneration(ctx, tx, next, now); err != nil {
		return err
	}
	if err := insertPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		next,
		definitionSet,
		now,
	); err != nil {
		return err
	}
	if err := verifyInsertedPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		next,
		definitionSet,
	); err != nil {
		return err
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_PackActivation
SET PreviousPackInstallId = NULL,
    ActivePackInstallId = ?,
    ActivationEpoch = ?,
    ActivatedDate = ?,
    LastModifiedDate = ?,
    IsDeleted = 0
WHERE Id = ?
  AND PackName = ?
  AND ActivePackInstallId = ?
  AND ActivationEpoch = ?
  AND COALESCE(IsDeleted, 0) = 0`,
		next.packInstallID,
		activation.activationEpoch+1,
		now,
		now,
		activation.id,
		activation.packName,
		activation.activePackInstallID,
		activation.activationEpoch,
	)
	if err != nil {
		return fmt.Errorf("activate pack %q replacement generation: %w", next.packName, err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("activate pack %q replacement rows affected: %w", next.packName, err)
	}
	if affected != 1 {
		return fmt.Errorf(
			"%w: pack %q activation changed during generation replacement",
			ErrPackGenerationConflict,
			next.packName,
		)
	}
	return nil
}

// verifyExistingSourceBackedPackGeneration is the read-only authority gate for
// an already-active target. Unlike the first-activation writer it never repairs
// missing authority from the compatibility projection.
func verifyExistingSourceBackedPackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	applied appliedAppPack,
) error {
	generation, err := sourceBackedPackGenerationForActivation(ctx, tx, applied)
	if err != nil {
		return err
	}
	activation, exists, err := readPackActivationForUpdate(
		ctx,
		tx,
		generation.packName,
	)
	if err != nil {
		return err
	}
	if !exists {
		return fmt.Errorf(
			"%w: active pack %q has no immutable generation activation",
			ErrPackGenerationConflict,
			generation.packName,
		)
	}
	return verifyExistingPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		applied,
		generation,
		activation,
	)
}

func sourceBackedPackGenerationForActivation(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	applied appliedAppPack,
) (sourceBackedPackGeneration, error) {
	if ctx == nil || tx == nil || applied.pack == nil {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: context, transaction, and pack are required",
			ErrPackGenerationPrecondition,
		)
	}
	generation, err := sourceBackedPackGenerationFromVerifiedPack(applied)
	if err != nil {
		return sourceBackedPackGeneration{}, err
	}
	if err := bindGenerationToActivatedCompatibilityRow(ctx, tx, &generation); err != nil {
		return sourceBackedPackGeneration{}, err
	}
	return generation, nil
}

func sourceBackedPackGenerationFromVerifiedPack(
	applied appliedAppPack,
) (sourceBackedPackGeneration, error) {
	pack := applied.pack
	if pack == nil {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack is required",
			ErrPackGenerationPrecondition,
		)
	}
	if !packs.IsPackPayloadHashV1(applied.payloadHash) {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q payload digest is not v1",
			ErrPackGenerationPrecondition,
			pack.Name(),
		)
	}
	if pack.Category() != packs.AppPack && pack.Category() != packs.ConnectorPack {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q category %q is not source-backed",
			ErrPackGenerationPrecondition,
			pack.Name(),
			pack.Category(),
		)
	}
	meta := packs.PackHashMetadataForPack(pack)
	if strings.TrimSpace(pack.Name()) != pack.Name() || pack.Name() == "" {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q has no canonical name",
			ErrPackGenerationPrecondition,
			pack.Name(),
		)
	}
	sourceKind, sourceURI, sourceRevision, publicationEligible, err := packGenerationSourceAuthority(meta)
	if err != nil {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q source authority: %v",
			ErrPackGenerationPrecondition,
			pack.Name(),
			err,
		)
	}
	if !publicationEligible {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q source kind %q is mutable and cannot publish an immutable generation",
			ErrPackGenerationPrecondition,
			pack.Name(),
			sourceKind,
		)
	}
	runtime, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: build pack %q Runtime declaration: %v",
			ErrPackGenerationPrecondition,
			pack.Name(),
			err,
		)
	}
	hostDigest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil || !packs.IsPackHostRequirementsDigestV1(hostDigest) {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: build pack %q host-requirements digest: %v",
			ErrPackGenerationPrecondition,
			pack.Name(),
			err,
		)
	}
	return sourceBackedPackGeneration{
		packName:                pack.Name(),
		category:                string(pack.Category()),
		status:                  packGenerationActive,
		packVersion:             nonBlankNullString(meta.PackVersion),
		manifestFormatVersion:   nonBlankNullString(meta.ManifestFormatVersion),
		platformCompatibility:   nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		desiredPayloadHash:      nonBlankNullString(applied.payloadHash),
		appliedPayloadHash:      nonBlankNullString(applied.payloadHash),
		payloadHashAlgorithm:    nonBlankNullString(meta.PayloadHashAlgorithm),
		sourceKind:              nonBlankNullString(sourceKind),
		sourceURI:               nonBlankNullString(sourceURI),
		sourceRevision:          sourceRevision,
		hostRequirementsDigest:  nonBlankNullString(hostDigest),
		runtimeObjectSpecDecl:   nonBlankNullString(runtime.ObjectPayloadsJSON()),
		runtimeRelatedLists:     nonBlankNullString(runtime.RelatedListRulesJSON()),
		runtimeSearchScope:      nonBlankNullString(runtime.SearchScopeOrderJSON()),
		runtimeSearchAllDefault: nonBlankNullString(runtime.SearchAllDefaultFieldsJSON()),
		runtimeFormulaBundles:   nonBlankNullString(runtime.FormulaFunctionBundlesJSON()),
		runtimeStartupShallow:   nonBlankNullString(runtime.StartupShallowDeclaredRowsJSON()),
	}, nil
}

// packGenerationSourceAuthority separates revision-addressed publication
// authority from mutable install input. An official_repo source must freeze its
// exact revision and is eligible for PackGeneration/PackActivation. local_path
// remains a mutable development source, regardless of whether its current tree
// happens to be Git-backed, and is deliberately not publication-eligible.
func packGenerationSourceAuthority(
	meta packs.PackHashMetadata,
) (string, string, sql.NullString, bool, error) {
	sourceKind := strings.TrimSpace(meta.SourceKind)
	sourceURI := strings.TrimSpace(meta.SourceURI)
	if !meta.SourceKindValid ||
		!meta.SourceURIValid ||
		sourceKind == "" ||
		sourceURI == "" ||
		sourceKind != meta.SourceKind ||
		sourceURI != meta.SourceURI {
		return "", "", sql.NullString{}, false, fmt.Errorf("source identity is incomplete or non-canonical")
	}
	canonicalKind, canonicalURI, err := canonicalPackSourceIdentity(sourceKind, sourceURI)
	if err != nil || canonicalKind != sourceKind || canonicalURI != sourceURI {
		return "", "", sql.NullString{}, false, fmt.Errorf("source identity is unsupported or non-canonical")
	}

	sourceRevision := strings.TrimSpace(meta.SourceRevision)
	hasSourceRevision := sourceRevision != ""
	if meta.SourceRevisionValid != hasSourceRevision ||
		sourceRevision != meta.SourceRevision {
		return "", "", sql.NullString{}, false, fmt.Errorf("source revision validity is inconsistent")
	}

	switch canonicalKind {
	case string(packs.PackSourceKindOfficialRepo):
		if !hasSourceRevision {
			return "", "", sql.NullString{}, false, fmt.Errorf("revision-addressed source revision is required")
		}
		return canonicalKind, canonicalURI, nonBlankNullString(sourceRevision), true, nil
	case string(packs.PackSourceKindReleaseRepo):
		if !hasSourceRevision {
			return "", "", sql.NullString{}, false, fmt.Errorf("revision-addressed source revision is required")
		}
		if err := packs.ValidateReleaseRepoRevision(sourceRevision); err != nil {
			return "", "", sql.NullString{}, false, err
		}
		return canonicalKind, canonicalURI, nonBlankNullString(sourceRevision), true, nil
	case string(packs.PackSourceKindLocalPath):
		return canonicalKind, canonicalURI, sql.NullString{}, false, nil
	default:
		return "", "", sql.NullString{}, false, fmt.Errorf("source kind %q is unsupported", canonicalKind)
	}
}

func packPublishesImmutableGeneration(pack packs.Pack) (bool, error) {
	if pack == nil {
		return false, fmt.Errorf("%w: pack is required", ErrPackGenerationPrecondition)
	}
	_, _, _, publicationEligible, err := packGenerationSourceAuthority(
		packs.PackHashMetadataForPack(pack),
	)
	if err != nil {
		return false, fmt.Errorf(
			"%w: pack %q source authority: %v",
			ErrPackGenerationPrecondition,
			pack.Name(),
			err,
		)
	}
	return publicationEligible, nil
}

// verifyMutableSourceHasNoGenerationAuthority is the negative half of the
// source-kind boundary. A mutable source must never leave Catalog authority
// behind for a serving adapter to select. Callers lock the compatibility row
// first, then activation and generation, matching publication lock order.
func verifyMutableSourceHasNoGenerationAuthority(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) error {
	packName = strings.TrimSpace(packName)
	if ctx == nil || tx == nil || packName == "" {
		return fmt.Errorf(
			"%w: mutable-source generation absence check requires context, store, and pack name",
			ErrPackGenerationPrecondition,
		)
	}
	if _, exists, err := readPackActivationForUpdate(ctx, tx, packName); err != nil {
		return err
	} else if exists {
		return fmt.Errorf(
			"%w: mutable source pack %q has PackActivation authority",
			ErrPackGenerationConflict,
			packName,
		)
	}
	installIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(ctx, tx, packName)
	if err != nil {
		return err
	}
	if len(installIDs) != 0 {
		return fmt.Errorf(
			"%w: mutable source pack %q has PackGeneration authority %q",
			ErrPackGenerationConflict,
			packName,
			installIDs[0],
		)
	}
	return nil
}

func verifyMutableSourceHasNoPublishedGenerationAuthority(
	ctx context.Context,
	db *sql.DB,
	packName string,
) error {
	if ctx == nil || db == nil {
		return fmt.Errorf(
			"%w: mutable-source generation absence check requires context and database",
			ErrPackGenerationPrecondition,
		)
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin mutable source pack %q generation absence check: %w", packName, err)
	}
	defer func() { _ = tx.Rollback() }()
	if _, _, err := readPackGenerationCompatibilityForUpdate(ctx, tx, packName); err != nil {
		return err
	}
	return verifyMutableSourceHasNoGenerationAuthority(ctx, tx, packName)
}

func bindGenerationToActivatedCompatibilityRow(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	generation *sourceBackedPackGeneration,
) error {
	var (
		status      string
		appliedHash sql.NullString
	)
	err := tx.QueryRowContext(ctx, `SELECT
COALESCE(PackInstallId, ''), Status, AppliedPayloadHash
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1
FOR UPDATE`, generation.packName).Scan(
		&generation.packInstallID,
		&status,
		&appliedHash,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: activated compatibility row for pack %q is missing",
			ErrPackGenerationPrecondition,
			generation.packName,
		)
	}
	if err != nil {
		return fmt.Errorf("lock activated pack %q compatibility row: %w", generation.packName, err)
	}
	if strings.TrimSpace(generation.packInstallID) != generation.packInstallID ||
		generation.packInstallID == "" ||
		status != installedPackStatusActive ||
		!appliedHash.Valid ||
		appliedHash.String != generation.appliedPayloadHash.String {
		return fmt.Errorf(
			"%w: pack %q compatibility row does not match the verified activation",
			ErrPackGenerationPrecondition,
			generation.packName,
		)
	}
	generation.id = packGenerationID(generation.packName, generation.packInstallID)
	return nil
}

func readInstalledPackGenerationPreimageForUpdate(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) (sourceBackedPackGeneration, bool, error) {
	var expected sourceBackedPackGeneration
	err := tx.QueryRowContext(ctx, `SELECT
COALESCE(PackInstallId, ''), Name, Category, Status,
PackVersion, ManifestFormatVersion, PlatformCompatibility,
DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm,
SourceKind, SourceURI, SourceRevision, HostRequirementsDigest,
RuntimeObjectSpecDeclarations, RuntimeRelatedListRules,
RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields,
RuntimeFormulaFunctionBundles, RuntimeStartupShallowDeclarations,
LastError, COALESCE(IsDeleted, 0)
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1
FOR UPDATE`, packName).Scan(
		&expected.packInstallID,
		&expected.packName,
		&expected.category,
		&expected.status,
		&expected.packVersion,
		&expected.manifestFormatVersion,
		&expected.platformCompatibility,
		&expected.desiredPayloadHash,
		&expected.appliedPayloadHash,
		&expected.payloadHashAlgorithm,
		&expected.sourceKind,
		&expected.sourceURI,
		&expected.sourceRevision,
		&expected.hostRequirementsDigest,
		&expected.runtimeObjectSpecDecl,
		&expected.runtimeRelatedLists,
		&expected.runtimeSearchScope,
		&expected.runtimeSearchAllDefault,
		&expected.runtimeFormulaBundles,
		&expected.runtimeStartupShallow,
		&expected.lastError,
		&expected.isDeleted,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return expected, false, nil
	}
	if err != nil {
		return expected, false, fmt.Errorf("lock pack %q InstalledPack generation preimage: %w", packName, err)
	}
	expected.id = packGenerationID(expected.packName, expected.packInstallID)
	return expected, true, nil
}

func readPackGenerationCompatibilityForUpdate(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) (packGenerationCompatibility, bool, error) {
	var compatibility packGenerationCompatibility
	err := tx.QueryRowContext(ctx, `SELECT
COALESCE(PackInstallId, ''), Status, COALESCE(IsDeleted, 0), AppliedPayloadHash
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1
FOR UPDATE`, packName).Scan(
		&compatibility.packInstallID,
		&compatibility.status,
		&compatibility.isDeleted,
		&compatibility.appliedPayloadHash,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return compatibility, false, nil
	}
	if err != nil {
		return compatibility, false, fmt.Errorf(
			"lock pack %q generation compatibility row: %w",
			packName,
			err,
		)
	}
	return compatibility, true, nil
}

func readPackActivationForUpdate(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) (packActivation, bool, error) {
	var (
		activation packActivation
		rawEpoch   []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
Id, PackName, ActivePackInstallId, PreviousPackInstallId,
ActivationEpoch, ActivatedDate, COALESCE(IsDeleted, 0)
FROM _System_PackActivation
WHERE PackName = ?
LIMIT 1
FOR UPDATE`, packName).Scan(
		&activation.id,
		&activation.packName,
		&activation.activePackInstallID,
		&activation.previousPackInstallID,
		&rawEpoch,
		&activation.activatedDate,
		&activation.isDeleted,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return activation, false, nil
	}
	if err != nil {
		return activation, false, fmt.Errorf("lock pack %q activation: %w", packName, err)
	}
	activation.activationEpoch, err = parsePackActivationEpoch(rawEpoch)
	if err != nil {
		return activation, false, fmt.Errorf(
			"%w: pack %q activation epoch: %v",
			ErrPackGenerationConflict,
			packName,
			err,
		)
	}
	return activation, true, nil
}

// readNonTerminalPackGenerationInstallIDsForUpdate returns only generations
// that can still participate in serving or activation. Retired and failed
// attempts are history, not runtime authority.
func readNonTerminalPackGenerationInstallIDsForUpdate(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) ([]string, error) {
	rows, err := tx.QueryContext(ctx, `SELECT PackInstallId
FROM _System_PackGeneration
WHERE PackName = ?
  AND Status IN ('preparing', 'active', 'draining', 'indeterminate')
  AND COALESCE(IsDeleted, 0) = 0
ORDER BY PackInstallId
LIMIT 2
FOR UPDATE`, packName)
	if err != nil {
		return nil, fmt.Errorf("lock pack %q generation set: %w", packName, err)
	}
	defer rows.Close()
	var values []string
	for rows.Next() {
		var value string
		if err := rows.Scan(&value); err != nil {
			return nil, err
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return values, nil
}

func verifyExistingPackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	expected sourceBackedPackGeneration,
	activation packActivation,
) error {
	if activation.id != packActivationID(expected.packName) ||
		activation.packName != expected.packName ||
		activation.activePackInstallID != expected.packInstallID {
		return fmt.Errorf(
			"%w: pack %q activation does not select the expected generation",
			ErrPackGenerationConflict,
			expected.packName,
		)
	}
	actual, err := verifyActivePackGeneration(ctx, tx, activation)
	if err != nil {
		return err
	}
	if !sourceBackedPackGenerationsEqual(actual, expected) {
		return fmt.Errorf(
			"%w: pack %q active generation does not match the verified source preimage",
			ErrPackGenerationConflict,
			expected.packName,
		)
	}
	return nil
}

func verifyActivePackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	activation packActivation,
) (sourceBackedPackGeneration, error) {
	if activation.id != packActivationID(activation.packName) ||
		strings.TrimSpace(activation.packName) == "" ||
		strings.TrimSpace(activation.activePackInstallID) == "" ||
		strings.TrimSpace(activation.activePackInstallID) != activation.activePackInstallID ||
		activation.activationEpoch < firstPackActivationEpoch ||
		!activation.activatedDate.Valid ||
		activation.isDeleted != 0 {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q activation is malformed",
			ErrPackGenerationConflict,
			activation.packName,
		)
	}
	if activation.previousPackInstallID.Valid {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q reserved previous-generation pointer must be null",
			ErrPackGenerationConflict,
			activation.packName,
		)
	}

	active, found, err := readPackGenerationForUpdate(
		ctx, tx, activation.activePackInstallID,
	)
	if err != nil {
		return sourceBackedPackGeneration{}, err
	}
	if !found {
		return sourceBackedPackGeneration{}, fmt.Errorf(
			"%w: pack %q active generation %q is missing",
			ErrPackGenerationConflict,
			activation.packName,
			activation.activePackInstallID,
		)
	}
	if err := validateStoredSourceBackedGeneration(active, activation.packName); err != nil {
		return sourceBackedPackGeneration{}, err
	}
	return active, nil
}

func validateStoredSourceBackedGeneration(
	generation sourceBackedPackGeneration,
	packName string,
) error {
	sourceKind, sourceURI, sourceErr := canonicalPackSourceIdentity(
		generation.sourceKind.String,
		generation.sourceURI.String,
	)
	if generation.id != packGenerationID(packName, generation.packInstallID) ||
		generation.packName != packName ||
		strings.TrimSpace(generation.packInstallID) == "" ||
		strings.TrimSpace(generation.packInstallID) != generation.packInstallID ||
		(generation.category != string(packs.AppPack) &&
			generation.category != string(packs.ConnectorPack)) ||
		generation.status != packGenerationActive ||
		generation.isDeleted != 0 ||
		generation.lastError.Valid ||
		!generation.desiredPayloadHash.Valid ||
		!generation.appliedPayloadHash.Valid ||
		generation.desiredPayloadHash.String != generation.appliedPayloadHash.String ||
		!packs.IsPackPayloadHashV1(generation.appliedPayloadHash.String) ||
		!generation.payloadHashAlgorithm.Valid ||
		generation.payloadHashAlgorithm.String != packs.PackPayloadHashAlgorithmV1 ||
		!generation.sourceKind.Valid ||
		!generation.sourceURI.Valid ||
		!generation.sourceRevision.Valid ||
		sourceErr != nil ||
		!packs.IsRevisionAddressedPackSourceKind(sourceKind) ||
		sourceKind != generation.sourceKind.String ||
		sourceURI != generation.sourceURI.String ||
		strings.TrimSpace(generation.sourceRevision.String) == "" ||
		strings.TrimSpace(generation.sourceRevision.String) != generation.sourceRevision.String ||
		!generation.hostRequirementsDigest.Valid ||
		!packs.IsPackHostRequirementsDigestV1(generation.hostRequirementsDigest.String) ||
		!generation.runtimeObjectSpecDecl.Valid ||
		!generation.runtimeRelatedLists.Valid ||
		!generation.runtimeSearchScope.Valid ||
		!generation.runtimeSearchAllDefault.Valid ||
		!generation.runtimeFormulaBundles.Valid ||
		!generation.runtimeStartupShallow.Valid ||
		!sourceBackedPackGenerationsEqual(generation, generation) {
		return fmt.Errorf(
			"%w: pack %q stored generation %q is malformed",
			ErrPackGenerationConflict,
			packName,
			generation.packInstallID,
		)
	}
	return nil
}

func sourceBackedPackGenerationNonRuntimeEqual(
	actual sourceBackedPackGeneration,
	expected sourceBackedPackGeneration,
) bool {
	actual.runtimeObjectSpecDecl = sql.NullString{}
	actual.runtimeRelatedLists = sql.NullString{}
	actual.runtimeSearchScope = sql.NullString{}
	actual.runtimeSearchAllDefault = sql.NullString{}
	actual.runtimeFormulaBundles = sql.NullString{}
	actual.runtimeStartupShallow = sql.NullString{}
	expected.runtimeObjectSpecDecl = sql.NullString{}
	expected.runtimeRelatedLists = sql.NullString{}
	expected.runtimeSearchScope = sql.NullString{}
	expected.runtimeSearchAllDefault = sql.NullString{}
	expected.runtimeFormulaBundles = sql.NullString{}
	expected.runtimeStartupShallow = sql.NullString{}
	return actual == expected
}

func verifyGenerationMatchesSourceRevisionAdvanceFrom(
	generation sourceBackedPackGeneration,
	advance normalizedApplyAllSourceRevisionAdvance,
) error {
	if err := validateStoredSourceBackedGeneration(generation, advance.Name); err != nil {
		return err
	}
	if generation.sourceKind.String != advance.Kind ||
		generation.sourceURI.String != advance.URI ||
		generation.sourceRevision.String != advance.FromRevision {
		return fmt.Errorf(
			"%w: pack %q active generation is not authorized source revision %q",
			ErrPackGenerationConflict,
			advance.Name,
			advance.FromRevision,
		)
	}
	return nil
}

// sourceBackedPackGenerationsEqual compares immutable scalar facts exactly and
// Runtime declaration JSON semantically. TiDB JSON columns are allowed to
// normalize whitespace and object-key order, so byte equality would reject an
// unchanged generation after the first database round trip. Canonicalization
// preserves JSON numbers and fails closed on invalid or duplicate-key input.
func sourceBackedPackGenerationsEqual(
	actual sourceBackedPackGeneration,
	expected sourceBackedPackGeneration,
) bool {
	actualRuntime := [...]sql.NullString{
		actual.runtimeObjectSpecDecl,
		actual.runtimeRelatedLists,
		actual.runtimeSearchScope,
		actual.runtimeSearchAllDefault,
		actual.runtimeFormulaBundles,
		actual.runtimeStartupShallow,
	}
	expectedRuntime := [...]sql.NullString{
		expected.runtimeObjectSpecDecl,
		expected.runtimeRelatedLists,
		expected.runtimeSearchScope,
		expected.runtimeSearchAllDefault,
		expected.runtimeFormulaBundles,
		expected.runtimeStartupShallow,
	}
	actual.runtimeObjectSpecDecl = sql.NullString{}
	actual.runtimeRelatedLists = sql.NullString{}
	actual.runtimeSearchScope = sql.NullString{}
	actual.runtimeSearchAllDefault = sql.NullString{}
	actual.runtimeFormulaBundles = sql.NullString{}
	actual.runtimeStartupShallow = sql.NullString{}
	expected.runtimeObjectSpecDecl = sql.NullString{}
	expected.runtimeRelatedLists = sql.NullString{}
	expected.runtimeSearchScope = sql.NullString{}
	expected.runtimeSearchAllDefault = sql.NullString{}
	expected.runtimeFormulaBundles = sql.NullString{}
	expected.runtimeStartupShallow = sql.NullString{}
	if actual != expected {
		return false
	}
	for index := range actualRuntime {
		if !packGenerationJSONEqual(actualRuntime[index], expectedRuntime[index]) {
			return false
		}
	}
	return true
}

func packGenerationJSONEqual(actual sql.NullString, expected sql.NullString) bool {
	if actual.Valid != expected.Valid {
		return false
	}
	if !actual.Valid {
		return true
	}
	actualCanonical, err := componentbehavior.CanonicalJSONV1([]byte(actual.String))
	if err != nil {
		return false
	}
	expectedCanonical, err := componentbehavior.CanonicalJSONV1([]byte(expected.String))
	if err != nil {
		return false
	}
	return bytes.Equal(actualCanonical, expectedCanonical)
}

func readPackGenerationForUpdate(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packInstallID string,
) (sourceBackedPackGeneration, bool, error) {
	var row sourceBackedPackGeneration
	err := tx.QueryRowContext(ctx, `SELECT
Id, PackInstallId, PackName, Category, Status, PackVersion,
ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash,
AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI,
SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations,
RuntimeRelatedListRules, RuntimeSearchScopeOrder,
RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles,
RuntimeStartupShallowDeclarations, LastError, COALESCE(IsDeleted, 0)
FROM _System_PackGeneration
WHERE PackInstallId = ?
LIMIT 1
FOR UPDATE`, packInstallID).Scan(
		&row.id,
		&row.packInstallID,
		&row.packName,
		&row.category,
		&row.status,
		&row.packVersion,
		&row.manifestFormatVersion,
		&row.platformCompatibility,
		&row.desiredPayloadHash,
		&row.appliedPayloadHash,
		&row.payloadHashAlgorithm,
		&row.sourceKind,
		&row.sourceURI,
		&row.sourceRevision,
		&row.hostRequirementsDigest,
		&row.runtimeObjectSpecDecl,
		&row.runtimeRelatedLists,
		&row.runtimeSearchScope,
		&row.runtimeSearchAllDefault,
		&row.runtimeFormulaBundles,
		&row.runtimeStartupShallow,
		&row.lastError,
		&row.isDeleted,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return row, false, nil
	}
	if err != nil {
		return row, false, fmt.Errorf("lock pack generation %q: %w", packInstallID, err)
	}
	return row, true, nil
}

func insertPackGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	value sourceBackedPackGeneration,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, `INSERT INTO _System_PackGeneration (
Id, PackInstallId, PackName, Category, Status, PackVersion,
ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash,
AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI,
SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations,
RuntimeRelatedListRules, RuntimeSearchScopeOrder,
RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles,
RuntimeStartupShallowDeclarations, LastError, CreatedDate,
LastModifiedDate, IsDeleted
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON),
CAST(? AS JSON), CAST(? AS JSON), CAST(? AS JSON), CAST(? AS JSON),
CAST(? AS JSON), NULL, ?, ?, 0
)`,
		value.id,
		value.packInstallID,
		value.packName,
		value.category,
		value.status,
		nullableValue(value.packVersion),
		nullableValue(value.manifestFormatVersion),
		nullableValue(value.platformCompatibility),
		nullableValue(value.desiredPayloadHash),
		nullableValue(value.appliedPayloadHash),
		nullableValue(value.payloadHashAlgorithm),
		nullableValue(value.sourceKind),
		nullableValue(value.sourceURI),
		nullableValue(value.sourceRevision),
		nullableValue(value.hostRequirementsDigest),
		nullableValue(value.runtimeObjectSpecDecl),
		nullableValue(value.runtimeRelatedLists),
		nullableValue(value.runtimeSearchScope),
		nullableValue(value.runtimeSearchAllDefault),
		nullableValue(value.runtimeFormulaBundles),
		nullableValue(value.runtimeStartupShallow),
		now,
		now,
	)
	if err != nil {
		return fmt.Errorf("insert pack %q generation: %w", value.packName, err)
	}
	return nil
}

func insertPackActivation(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	value sourceBackedPackGeneration,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, `INSERT INTO _System_PackActivation (
Id, PackName, ActivePackInstallId, PreviousPackInstallId,
ActivationEpoch, ActivatedDate, CreatedDate, LastModifiedDate, IsDeleted
) VALUES (?, ?, ?, NULL, ?, ?, ?, ?, 0)`,
		packActivationID(value.packName),
		value.packName,
		value.packInstallID,
		firstPackActivationEpoch,
		now,
		now,
		now,
	)
	if err != nil {
		return fmt.Errorf("insert pack %q first activation: %w", value.packName, err)
	}
	return nil
}

func packGenerationID(packName, packInstallID string) string {
	return connectorbindingdefinitionauthority.PackGenerationStorageID(
		packName,
		packInstallID,
	)
}

func packActivationID(packName string) string {
	return "packact_" + hashPackAuthorityID(packName)
}

func hashPackAuthorityID(value string) string {
	sum := sha256.Sum256([]byte(value))
	return hex.EncodeToString(sum[:])
}

func nonBlankNullString(value string) sql.NullString {
	return sql.NullString{String: value, Valid: strings.TrimSpace(value) != ""}
}

func nullableValue(value sql.NullString) any {
	if !value.Valid {
		return nil
	}
	return value.String
}

func parsePackActivationEpoch(raw []byte) (uint64, error) {
	var value uint64
	if len(raw) == 0 {
		return 0, fmt.Errorf("value is empty")
	}
	for _, character := range raw {
		if character < '0' || character > '9' {
			return 0, fmt.Errorf("value is not an unsigned decimal")
		}
		digit := uint64(character - '0')
		if value > (^uint64(0)-digit)/10 {
			return 0, fmt.Errorf("value overflows uint64")
		}
		value = value*10 + digit
	}
	if value == 0 {
		return 0, fmt.Errorf("value must be positive")
	}
	return value, nil
}