package catalogapp

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strings"

	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	layoutbundlekeyv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogadmissionfence"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
)

type installationRecord struct {
	ID                    string
	OrgID                 string
	AppID                 string
	State                 InstallationState
	ActiveRelease         string
	CandidateRelease      string
	LastKnownGoodRelease  string
	CatalogRevision       uint64
	ActivationEpoch       uint64
	ActiveOperationID     string
	ActiveWorkloads       []string
	CandidateWorkloads    []string
	ActiveCapabilities    []string
	CandidateCapabilities []string
	ActiveDependencies    []ResolvedDependency
	CandidateDependencies []ResolvedDependency
}

type operationInitiationKind string

const (
	initiationDirect           operationInitiationKind = "DIRECT"
	initiationLegacyConversion operationInitiationKind = "LEGACY_CONVERSION"
)

func (a *Authority) BeginOperation(ctx context.Context, request BeginOperationRequest) (Operation, error) {
	normalized, release, requestDigest, err := a.validateBeginOperation(ctx, request)
	if err != nil {
		return Operation{}, err
	}
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Operation{}, err
	}
	defer rollback(tx)
	operation, err := a.beginOperationTx(ctx, tx, normalized, release, requestDigest, initiationDirect)
	if err != nil {
		return Operation{}, err
	}
	if err := tx.Commit(); err != nil {
		return Operation{}, err
	}
	return operation, nil
}

func (a *Authority) validateBeginOperation(ctx context.Context, request BeginOperationRequest) (BeginOperationRequest, RegisteredRelease, string, error) {
	return a.validateBeginOperationForTarget(ctx, request, nil, false)
}

func (a *Authority) validateBeginOperationForTarget(
	ctx context.Context,
	request BeginOperationRequest,
	verifiedTarget *RegisteredRelease,
	stoppedSource bool,
) (BeginOperationRequest, RegisteredRelease, string, error) {
	if err := validateOrgAndOpaqueID(request.OrgID, "idempotency key", request.IdempotencyKey); err != nil {
		return request, RegisteredRelease{}, "", err
	}
	if len(request.IdempotencyKey) > 255 || strings.TrimSpace(request.RuntimeAudience) == "" || len(request.RuntimeAudience) > 255 {
		return request, RegisteredRelease{}, "", fmt.Errorf("idempotency key and Runtime audience must fit their Catalog fields")
	}
	if err := appauthority.ValidateAppID(request.AppID); err != nil {
		return request, RegisteredRelease{}, "", err
	}
	switch request.Kind {
	case OperationPrepare, OperationUpgrade, OperationRollback:
		if err := requireDigest("target AppReleaseDigest", request.TargetAppReleaseDigest); err != nil {
			return request, RegisteredRelease{}, "", err
		}
	case OperationRemove:
		if request.TargetAppReleaseDigest != "" {
			return request, RegisteredRelease{}, "", fmt.Errorf("remove cannot declare a target release")
		}
	default:
		return request, RegisteredRelease{}, "", fmt.Errorf("unknown operation kind %q", request.Kind)
	}
	if request.RefreshActiveRelease && request.Kind != OperationUpgrade {
		return request, RegisteredRelease{}, "", fmt.Errorf("active release refresh requires an upgrade operation")
	}
	if request.Kind == OperationPrepare && request.AppInstallationID != "" {
		return request, RegisteredRelease{}, "", fmt.Errorf("prepare installation ID is Catalog-minted")
	}
	if request.Kind != OperationPrepare {
		if err := appauthority.ValidateAppInstallationID(request.AppInstallationID); err != nil {
			return request, RegisteredRelease{}, "", err
		}
	}
	var release RegisteredRelease
	var err error
	if request.Kind != OperationRemove {
		if verifiedTarget == nil {
			release, err = a.ReadRelease(ctx, request.AppID, request.TargetAppReleaseDigest)
			if err != nil {
				return request, RegisteredRelease{}, "", err
			}
		} else {
			release = *verifiedTarget
			if release.AppID != request.AppID || release.AppReleaseDigest != request.TargetAppReleaseDigest {
				return request, RegisteredRelease{}, "", ErrConflict
			}
		}
		if stoppedSource {
			if release.AdmissionKind != ReleaseAdmissionStoppedSource {
				return request, RegisteredRelease{}, "", fmt.Errorf(
					"%w: stopped-source operation requires a non-promotable stopped-source AppRelease",
					ErrNotReady,
				)
			}
			if len(release.Package.Manifest.Dependencies) != 0 || len(request.ResolvedDependencies) != 0 {
				return request, RegisteredRelease{}, "", fmt.Errorf(
					"%w: stopped-source operation v1 requires an empty target dependency set",
					ErrNotReady,
				)
			}
		} else if release.AdmissionKind != ReleaseAdmissionSignedPublisher {
			return request, RegisteredRelease{}, "", fmt.Errorf(
				"%w: ordinary Catalog operation cannot target non-promotable AppRelease admission %q",
				ErrNotReady,
				release.AdmissionKind,
			)
		}
		request.AuthorizedWorkloads, err = canonicalStrings("authorized workloads", request.AuthorizedWorkloads, true)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
		request.GrantedCapabilities, err = canonicalStrings("granted capabilities", request.GrantedCapabilities, true)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
		if !slices.Equal(request.GrantedCapabilities, release.Package.Manifest.RequiredEngineCapabilities) {
			return request, RegisteredRelease{}, "", fmt.Errorf("granted capabilities must exactly match the verified App release requirements")
		}
		if (len(request.AuthorizedWorkloads) == 0) != (len(request.GrantedCapabilities) == 0) {
			return request, RegisteredRelease{}, "", fmt.Errorf("authorized workloads and Runtime capability grants must either both be empty or both be non-empty")
		}
		request.SelectedHandlerCapabilities,
			request.WorkloadHandlerAdmissions,
			err = canonicalHandlerProjectionAuthority(
			release.Package.Manifest.ProvidedCapabilities,
			request.SelectedHandlerCapabilities,
			request.WorkloadHandlerAdmissions,
			request.AuthorizedWorkloads,
			request.GrantedCapabilities,
		)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
		request.ResolvedDependencies, err = canonicalDependencies(request.ResolvedDependencies)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
		if !stoppedSource {
			if err := a.validateDependencyReleases(ctx, release, request.ResolvedDependencies); err != nil {
				return request, RegisteredRelease{}, "", err
			}
		}
		request.ConsumerGrants, err = canonicalOperationConsumerGrants(
			release.Package.Manifest.ProvidedCapabilities,
			release.Package.Manifest.RequiredEngineCapabilities,
			request.HandlerWorkloadID,
			request.WorkloadHandlerAdmissions,
			request.ConsumerGrants,
			request.AuthorizedWorkloads,
		)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
		request.IntegrationBindings, err = canonicalIntegrationBindingSelections(
			request.IntegrationBindings,
			request.AuthorizedWorkloads,
			request.ConsumerGrants,
		)
		if err != nil {
			return request, RegisteredRelease{}, "", err
		}
	} else if len(request.AuthorizedWorkloads) != 0 ||
		len(request.GrantedCapabilities) != 0 ||
		request.HandlerWorkloadID != "" ||
		len(request.SelectedHandlerCapabilities) != 0 ||
		len(request.WorkloadHandlerAdmissions) != 0 ||
		len(request.ResolvedDependencies) != 0 ||
		len(request.ConsumerGrants) != 0 ||
		len(request.IntegrationBindings) != 0 {
		return request, RegisteredRelease{}, "", fmt.Errorf("remove uses the active installation inputs and cannot declare candidate inputs")
	}
	if request.ContinuationIntentDigest != "" {
		if err := requireDigest("continuation intent", request.ContinuationIntentDigest); err != nil {
			return request, RegisteredRelease{}, "", err
		}
	}
	if request.RefreshActiveRelease && request.ContinuationIntentDigest == "" {
		return request, RegisteredRelease{}, "", fmt.Errorf("active release refresh requires a continuation intent")
	}
	requestDigest, err := operationRequestDigest(request)
	if err != nil {
		return request, RegisteredRelease{}, "", err
	}
	return request, release, requestDigest, nil
}

func operationRequestDigest(request BeginOperationRequest) (string, error) {
	encoded, err := json.Marshal(struct {
		Schema  string                `json:"schema"`
		Request BeginOperationRequest `json:"request"`
	}{"nexuscrm.catalog-app-operation-request/v1", request})
	if err != nil {
		return "", err
	}
	return digestParts(string(encoded)), nil
}

func canonicalDependencies(values []ResolvedDependency) ([]ResolvedDependency, error) {
	result := append([]ResolvedDependency(nil), values...)
	for i, dependency := range result {
		if strings.TrimSpace(dependency.Alias) == "" {
			return nil, fmt.Errorf("resolved dependency[%d] alias is required", i)
		}
		if err := appauthority.ValidateAppID(dependency.AppID); err != nil {
			return nil, fmt.Errorf("resolved dependency[%d]: %w", i, err)
		}
		if err := requireDigest("resolved dependency release", dependency.AppReleaseDigest); err != nil {
			return nil, fmt.Errorf("resolved dependency[%d]: %w", i, err)
		}
		if i > 0 && dependency.Alias <= result[i-1].Alias {
			return nil, fmt.Errorf("resolved dependencies must be strictly sorted by alias")
		}
	}
	return result, nil
}

func (a *Authority) validateDependencyReleases(ctx context.Context, release RegisteredRelease, resolved []ResolvedDependency) error {
	manifest := release.Package.Manifest
	if len(manifest.Dependencies) != len(resolved) {
		return fmt.Errorf("resolved dependency set is incomplete")
	}
	for i, requirement := range manifest.Dependencies {
		resolution := resolved[i]
		if resolution.Alias != requirement.Alias || resolution.AppID != requirement.AppID {
			return fmt.Errorf("resolved dependency %q does not match the App release", resolution.Alias)
		}
		dependency, err := a.ReadRelease(ctx, resolution.AppID, resolution.AppReleaseDigest)
		if err != nil {
			return err
		}
		if dependency.AdmissionKind != ReleaseAdmissionSignedPublisher {
			return fmt.Errorf(
				"%w: resolved dependency %q uses non-promotable AppRelease admission %q",
				ErrNotReady,
				resolution.Alias,
				dependency.AdmissionKind,
			)
		}
		minimum, err := appauthority.CompareSemanticVersions(dependency.SemanticVersion, requirement.MinimumVersionInclusive)
		if err != nil || minimum < 0 {
			return fmt.Errorf("resolved dependency %q is below its minimum version", resolution.Alias)
		}
		maximum, err := appauthority.CompareSemanticVersions(dependency.SemanticVersion, requirement.MaximumVersionExclusive)
		if err != nil || maximum >= 0 {
			return fmt.Errorf("resolved dependency %q is at or above its exclusive maximum version", resolution.Alias)
		}
		provided := make(map[string]struct{}, len(dependency.Package.Manifest.ProvidedCapabilities))
		for _, capability := range dependency.Package.Manifest.ProvidedCapabilities {
			provided[capability.CapabilityID] = struct{}{}
		}
		for _, capability := range requirement.RequiredCapabilities {
			if _, found := provided[capability]; !found {
				return fmt.Errorf("resolved dependency %q does not provide %q", resolution.Alias, capability)
			}
		}
	}
	return nil
}

func (a *Authority) beginOperationTx(ctx context.Context, tx *sql.Tx, request BeginOperationRequest, release RegisteredRelease, requestDigest string, initiation operationInitiationKind) (Operation, error) {
	if initiation != initiationDirect && initiation != initiationLegacyConversion {
		return Operation{}, fmt.Errorf("invalid operation initiation kind")
	}
	if _, err := catalogadmissionfence.LockExclusive(ctx, tx); err != nil {
		return Operation{}, fmt.Errorf(
			"%w: publication admission fence: %v",
			ErrCorrupt,
			err,
		)
	}
	return a.beginOperationAfterAdmissionFence(
		ctx,
		tx,
		request,
		release,
		requestDigest,
		initiation,
	)
}

func (a *Authority) beginOperationAfterAdmissionFence(ctx context.Context, tx *sql.Tx, request BeginOperationRequest, release RegisteredRelease, requestDigest string, initiation operationInitiationKind) (Operation, error) {
	var existingID, existingDigest, existingInitiation string
	err := tx.QueryRowContext(ctx, `SELECT OperationId, RequestDigest, InitiationKind FROM _System_CatalogAppOperation
WHERE OrgId = ? AND AppId = ? AND IdempotencyKey = ? FOR UPDATE`, request.OrgID, request.AppID, request.IdempotencyKey).Scan(
		&existingID, &existingDigest, &existingInitiation)
	if err == nil {
		if existingDigest != requestDigest || existingInitiation != string(initiation) {
			return Operation{}, ErrConflict
		}
		if err := rejectCandidateSourceOperationIDCollisionLocked(
			ctx, tx, existingID, true,
		); err != nil {
			return Operation{}, err
		}
		operation, err := a.readOperation(ctx, tx, existingID, true)
		if err != nil {
			return Operation{}, err
		}
		if err := ensureConsumerGrantSetForOperationRetryTx(
			ctx, tx, operation, request.ConsumerGrants,
		); err != nil {
			return Operation{}, err
		}
		if err := ensureIntegrationBindingSetForOperationRetryTx(
			ctx, tx, operation, request.IntegrationBindings,
		); err != nil {
			return Operation{}, err
		}
		return operation, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return Operation{}, err
	}

	operationID, err := a.newID(a.now())
	if err != nil {
		return Operation{}, err
	}
	if err := rejectCandidateSourceOperationIDCollisionLocked(
		ctx, tx, operationID, false,
	); err != nil {
		return Operation{}, err
	}

	writerState, err := catalogwriter.LockOrCreate(ctx, tx, request.OrgID, request.RuntimeAudience)
	if err != nil {
		return Operation{}, catalogWriterError(err)
	}
	currentCatalog, currentSecurity, activeOperation := writerState.CatalogRevision, writerState.SecurityEpoch, writerState.ActiveCommandID
	if currentCatalog != request.ExpectedCatalogRevision || currentSecurity != request.ExpectedSecurityEpoch || activeOperation != "" {
		return Operation{}, ErrConflict
	}
	// A source command can reserve and release this writer while App Begin is
	// waiting. Re-check the minted identity under the now-idle writer before
	// any App installation, namespace, or operation mutation.
	if err := rejectCandidateSourceOperationIDCollisionLocked(
		ctx, tx, operationID, false,
	); err != nil {
		return Operation{}, err
	}
	targetCatalog, catalogErr := catalogwriter.NextCatalogRevision(writerState)
	targetSecurity, securityErr := catalogwriter.NextSecurityEpoch(writerState)
	if catalogErr != nil || securityErr != nil {
		return Operation{}, fmt.Errorf("Catalog revision or security epoch exhausted")
	}

	installationID := request.AppInstallationID
	transitionState := transitionState(request.Kind)
	var installation installationRecord
	if request.Kind == OperationPrepare {
		installationID, err = a.newID(a.now())
		if err != nil {
			return Operation{}, err
		}
		if err := appauthority.ValidateAppInstallationID(installationID); err != nil {
			return Operation{}, err
		}
		installation = installationRecord{ID: installationID, OrgID: request.OrgID, AppID: request.AppID, State: transitionState}
	} else {
		installation, err = readInstallationForUpdate(ctx, tx, installationID)
		if err != nil {
			return Operation{}, err
		}
		if installation.OrgID != request.OrgID || installation.AppID != request.AppID || installation.State != StateActive || installation.ActiveOperationID != "" || installation.ActiveRelease == "" {
			return Operation{}, ErrInvalidTransition
		}
		if err := ensureNoStandaloneBindingCandidate(ctx, tx, installationID); err != nil {
			return Operation{}, err
		}
		switch request.Kind {
		case OperationUpgrade:
			if request.RefreshActiveRelease {
				if request.TargetAppReleaseDigest != installation.ActiveRelease {
					return Operation{}, ErrInvalidTransition
				}
			} else if request.TargetAppReleaseDigest == installation.ActiveRelease {
				return Operation{}, ErrInvalidTransition
			}
		case OperationRollback:
			if installation.LastKnownGoodRelease == "" || request.TargetAppReleaseDigest != installation.LastKnownGoodRelease || request.TargetAppReleaseDigest == installation.ActiveRelease {
				return Operation{}, ErrInvalidTransition
			}
		}
	}
	if request.Kind != OperationRemove {
		if err := validateActiveDependencies(ctx, tx, request.OrgID, request.ResolvedDependencies); err != nil {
			return Operation{}, err
		}
		if err := reserveNamespaces(ctx, tx, request.OrgID, installationID, request.AppID, request.TargetAppReleaseDigest, release.Package.Manifest.OwnedNamespaces); err != nil {
			return Operation{}, err
		}
	}

	workloadsJSON, err := marshalJSON(request.AuthorizedWorkloads)
	if err != nil {
		return Operation{}, err
	}
	capabilitiesJSON, err := marshalJSON(request.GrantedCapabilities)
	if err != nil {
		return Operation{}, err
	}
	dependenciesJSON, err := marshalJSON(request.ResolvedDependencies)
	if err != nil {
		return Operation{}, err
	}
	var selectedHandlerCapabilitiesJSON any
	if len(request.SelectedHandlerCapabilities) != 0 {
		selectedHandlerCapabilitiesJSON, err = marshalJSON(
			request.SelectedHandlerCapabilities,
		)
		if err != nil {
			return Operation{}, err
		}
	}
	var workloadHandlerAdmissionsJSON any
	if len(request.WorkloadHandlerAdmissions) != 0 {
		workloadHandlerAdmissionsJSON, err = marshalJSON(
			request.WorkloadHandlerAdmissions,
		)
		if err != nil {
			return Operation{}, err
		}
	}
	if request.Kind == OperationPrepare {
		_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogAppInstallation
(AppInstallationId, OrgId, AppId, State, ActiveReleaseDigest, CandidateReleaseDigest,
LastKnownGoodReleaseDigest, CatalogRevision, ActivationEpoch, ActiveOperationId,
ActiveAuthorizedWorkloads, CandidateAuthorizedWorkloads, ActiveGrantedCapabilities, CandidateGrantedCapabilities,
ActiveResolvedDependencies, CandidateResolvedDependencies, LiveAppKey, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, NULL, ?, NULL, ?, 0, ?, NULL, CAST(? AS JSON), NULL, CAST(? AS JSON), NULL, CAST(? AS JSON), ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
			installationID, request.OrgID, request.AppID, transitionState, request.TargetAppReleaseDigest,
			epochText(currentCatalog), operationID, workloadsJSON, capabilitiesJSON, dependenciesJSON, hashKey(request.OrgID, request.AppID))
		if isDuplicate(err) {
			return Operation{}, ErrConflict
		}
	} else {
		candidateRelease := nullableString(request.TargetAppReleaseDigest)
		candidateWorkloads, candidateCapabilities, candidateDependencies := any(nil), any(nil), any(nil)
		if request.Kind != OperationRemove {
			candidateWorkloads, candidateCapabilities, candidateDependencies = workloadsJSON, capabilitiesJSON, dependenciesJSON
		}
		result, updateErr := tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation
SET State = ?, CandidateReleaseDigest = ?, ActiveOperationId = ?,
CandidateAuthorizedWorkloads = CAST(? AS JSON), CandidateGrantedCapabilities = CAST(? AS JSON),
CandidateResolvedDependencies = CAST(? AS JSON), LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND State = 'ACTIVE' AND ActiveOperationId IS NULL`,
			transitionState, candidateRelease, operationID, candidateWorkloads, candidateCapabilities, candidateDependencies, installationID)
		if updateErr != nil {
			return Operation{}, updateErr
		}
		affected, updateErr := result.RowsAffected()
		if updateErr != nil || affected != 1 {
			return Operation{}, ErrConflict
		}
	}
	priorRelease := installation.ActiveRelease
	placeholder := digestParts(operationID, "projection-pending")
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogAppOperation
(OperationId, AppInstallationId, OrgId, AppId, Kind, InitiationKind, IdempotencyKey, RequestDigest,
SelectedHandlerCapabilities, WorkloadHandlerAdmissions, State, Step,
TargetReleaseDigest, PriorReleaseDigest, ExpectedCatalogRevision, ExpectedSecurityEpoch,
ExpectedProjectionDigest, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), CAST(? AS JSON), ?, 'CANDIDATE_COMMITTED', ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		operationID, installationID, request.OrgID, request.AppID, request.Kind, initiation, request.IdempotencyKey, requestDigest,
		selectedHandlerCapabilitiesJSON, workloadHandlerAdmissionsJSON, transitionState, nullableString(request.TargetAppReleaseDigest), nullableString(priorRelease), epochText(targetCatalog), epochText(targetSecurity), placeholder)
	if isDuplicate(err) {
		return Operation{}, ErrConflict
	}
	if err != nil {
		return Operation{}, err
	}
	if err := stageConsumerGrantsTx(
		ctx, tx, operationID, installationID, request.OrgID,
		request.ConsumerGrants,
	); err != nil {
		return Operation{}, err
	}
	if err := stageIntegrationBindingsTx(
		ctx, tx, operationID, installationID, request.OrgID,
		request.IntegrationBindings,
	); err != nil {
		return Operation{}, err
	}
	if err := catalogwriter.Reserve(ctx, tx, writerState, operationID); err != nil {
		return Operation{}, catalogWriterError(err)
	}
	projection, err := a.buildProjection(ctx, tx, operationID, installationID, request.OrgID, request.Kind, targetSecurity)
	if err != nil {
		return Operation{}, err
	}
	projectionDigest, err := ProjectionDigest(projection)
	if err != nil {
		return Operation{}, err
	}
	if _, err := tx.ExecContext(ctx, `UPDATE _System_CatalogAppOperation SET ExpectedProjectionDigest = ? WHERE OperationId = ?`, projectionDigest, operationID); err != nil {
		return Operation{}, err
	}
	if err := appendJournal(ctx, tx, operationID, 1, "CANDIDATE_COMMITTED", string(transitionState), projectionDigest); err != nil {
		return Operation{}, err
	}
	return Operation{OperationID: operationID, Kind: request.Kind, AppInstallationID: installationID,
		OrgID: request.OrgID, AppID: request.AppID, State: transitionState, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.TargetAppReleaseDigest, TargetCatalogRevision: targetCatalog,
		TargetSecurityEpoch: targetSecurity, ExpectedProjectionDigest: projectionDigest}, nil
}

// rejectCandidateSourceOperationIDCollisionLocked keeps Catalog-minted App
// operation IDs disjoint from durable candidate-source writer commands. A
// missing source table is equivalent to an empty source journal during a
// rolling schema deployment; once the table is provisioned, source Prepare
// performs the symmetric App-operation check.
func rejectCandidateSourceOperationIDCollisionLocked(
	ctx context.Context,
	tx *sql.Tx,
	operationID string,
	asCorruption bool,
) error {
	var sourceOrgID string
	err := tx.QueryRowContext(ctx, `SELECT OrgId
FROM _System_CatalogCandidateSource
WHERE WriterCommandId = ? LIMIT 1 FOR UPDATE`, operationID).Scan(&sourceOrgID)
	if errors.Is(err, sql.ErrNoRows) || isMissingTable(err) {
		return nil
	}
	if err != nil {
		return err
	}
	if asCorruption {
		return fmt.Errorf(
			"%w: App operation collides with a candidate-source journal",
			ErrCorrupt,
		)
	}
	return ErrConflict
}

// ensureNoStandaloneBindingCandidate prevents an AppRelease ownership flip
// from stranding a service-only or frontend-only candidate that was fenced to
// the currently active release. Stage*Binding and beginOperationTx both lock
// the installation row first, so the installation is the serialization owner
// for this cross-plane invariant.
func ensureNoStandaloneBindingCandidate(ctx context.Context, tx *sql.Tx, installationID string) error {
	queries := []string{
		`SELECT HandlerBindingId FROM _System_CatalogHandlerBinding
WHERE AppInstallationId = ? AND OperationId IS NULL AND State = 'STAGED' LIMIT 1 FOR UPDATE`,
		`SELECT FrontendBindingId FROM _System_CatalogFrontendBinding
WHERE AppInstallationId = ? AND OperationId IS NULL AND State = 'STAGED' LIMIT 1 FOR UPDATE`,
	}
	for _, query := range queries {
		var bindingID string
		err := tx.QueryRowContext(ctx, query, installationID).Scan(&bindingID)
		if err == nil {
			return fmt.Errorf("%w: resolve the standalone binding candidate before changing the App release", ErrNotReady)
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
	}
	return nil
}

func readInstallationForUpdate(ctx context.Context, tx *sql.Tx, installationID string) (installationRecord, error) {
	var record installationRecord
	var activeRelease, candidateRelease, lastGood, activeOperation sql.NullString
	var rawCatalog, rawActivation []byte
	var activeWorkloads, candidateWorkloads, activeCapabilities, candidateCapabilities, activeDependencies, candidateDependencies []byte
	err := tx.QueryRowContext(ctx, `SELECT AppInstallationId, OrgId, AppId, State,
ActiveReleaseDigest, CandidateReleaseDigest, LastKnownGoodReleaseDigest, CatalogRevision, ActivationEpoch, ActiveOperationId,
ActiveAuthorizedWorkloads, CandidateAuthorizedWorkloads, ActiveGrantedCapabilities, CandidateGrantedCapabilities,
ActiveResolvedDependencies, CandidateResolvedDependencies
FROM _System_CatalogAppInstallation WHERE AppInstallationId = ? FOR UPDATE`, installationID).Scan(
		&record.ID, &record.OrgID, &record.AppID, &record.State, &activeRelease, &candidateRelease, &lastGood,
		&rawCatalog, &rawActivation, &activeOperation, &activeWorkloads, &candidateWorkloads,
		&activeCapabilities, &candidateCapabilities, &activeDependencies, &candidateDependencies)
	if errors.Is(err, sql.ErrNoRows) {
		return record, ErrNotFound
	}
	if err != nil {
		return record, err
	}
	record.ActiveRelease, record.CandidateRelease, record.LastKnownGoodRelease, record.ActiveOperationID = activeRelease.String, candidateRelease.String, lastGood.String, activeOperation.String
	if record.CatalogRevision, err = parseEpochAllowZero(rawCatalog); err != nil {
		return record, err
	}
	if record.ActivationEpoch, err = parseEpochAllowZero(rawActivation); err != nil {
		return record, err
	}
	for raw, target := range map[*[]byte]any{
		&activeWorkloads: &record.ActiveWorkloads, &candidateWorkloads: &record.CandidateWorkloads,
		&activeCapabilities: &record.ActiveCapabilities, &candidateCapabilities: &record.CandidateCapabilities,
		&activeDependencies: &record.ActiveDependencies, &candidateDependencies: &record.CandidateDependencies,
	} {
		if len(*raw) != 0 {
			if err := json.Unmarshal(*raw, target); err != nil {
				return record, err
			}
		}
	}
	return record, nil
}

func validateActiveDependencies(ctx context.Context, tx *sql.Tx, orgID string, dependencies []ResolvedDependency) error {
	for _, dependency := range dependencies {
		var activeDigest string
		err := tx.QueryRowContext(ctx, `SELECT ActiveReleaseDigest FROM _System_CatalogAppInstallation
WHERE OrgId = ? AND AppId = ? AND State = 'ACTIVE' AND LiveAppKey IS NOT NULL FOR UPDATE`, orgID, dependency.AppID).Scan(&activeDigest)
		if errors.Is(err, sql.ErrNoRows) || (err == nil && activeDigest != dependency.AppReleaseDigest) {
			return fmt.Errorf("%w: dependency %q is not active at the resolved release", ErrNotReady, dependency.Alias)
		}
		if err != nil {
			return err
		}
	}
	return nil
}

func reserveNamespaces(ctx context.Context, tx *sql.Tx, orgID, installationID, appID, releaseDigest string, namespaces []string) error {
	for _, namespace := range namespaces {
		if namespace == layoutbundlekeyv1.SystemSlotOwnerNamespace {
			return fmt.Errorf(
				"%w: system namespace cannot be reserved by an App",
				ErrConflict,
			)
		}
		var existingInstallation string
		err := tx.QueryRowContext(ctx, `SELECT AppInstallationId FROM _System_CatalogNamespaceReservation
WHERE OrgId = ? AND NamespaceId = ? FOR UPDATE`, orgID, namespace).Scan(&existingInstallation)
		switch {
		case errors.Is(err, sql.ErrNoRows):
			_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogNamespaceReservation
(OrgId, NamespaceId, AppInstallationId, AppId, ActiveReleaseDigest, CandidateReleaseDigest, State, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, NULL, ?, 'STAGED', CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`, orgID, namespace, installationID, appID, releaseDigest)
		case err != nil:
			return err
		case existingInstallation != installationID:
			return fmt.Errorf("%w: namespace %q is owned by another installation", ErrConflict, namespace)
		default:
			_, err = tx.ExecContext(ctx, `UPDATE _System_CatalogNamespaceReservation
SET CandidateReleaseDigest = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ? AND NamespaceId = ? AND AppInstallationId = ?`, releaseDigest, orgID, namespace, installationID)
		}
		if isDuplicate(err) {
			return ErrConflict
		}
		if err != nil {
			return err
		}
	}
	return nil
}

func transitionState(kind OperationKind) InstallationState {
	switch kind {
	case OperationPrepare:
		return StatePreparing
	case OperationUpgrade:
		return StateUpgrading
	case OperationRollback:
		return StateRollingBack
	case OperationRemove:
		return StateRemoving
	default:
		return StateFailed
	}
}

func (a *Authority) CompleteOperation(ctx context.Context, operationID string, removal *RemovalGateReceipt) (Operation, error) {
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Operation{}, err
	}
	defer rollback(tx)
	if _, err := catalogadmissionfence.LockExclusive(ctx, tx); err != nil {
		return Operation{}, fmt.Errorf(
			"%w: publication admission fence: %v",
			ErrCorrupt,
			err,
		)
	}
	operationOrgID, err := readOperationOrgID(ctx, tx, operationID)
	if err != nil {
		return Operation{}, err
	}
	writerState, err := catalogwriter.Lock(ctx, tx, operationOrgID)
	if err != nil {
		return Operation{}, catalogWriterError(err)
	}
	operation, err := a.readOperation(ctx, tx, operationID, true)
	if err != nil {
		return Operation{}, err
	}
	if operation.OrgID != operationOrgID {
		return Operation{}, ErrConflict
	}
	if operation.Step == "ACTIVATED" {
		if operation.Kind == OperationRemove {
			if err := verifyRemovalGate(ctx, tx, operation, removal); err != nil {
				return Operation{}, err
			}
		} else if removal != nil {
			return Operation{}, ErrConflict
		}
		if err := tx.Commit(); err != nil {
			return Operation{}, err
		}
		return operation, nil
	}
	if writerState.ActiveCommandID != operationID {
		return Operation{}, ErrConflict
	}
	if operation.Step != "PROJECTION_ACKNOWLEDGED" || operation.State == StateFailed ||
		operation.AcknowledgedSecurityEpoch != operation.TargetSecurityEpoch || operation.AcknowledgedProjectionDigest != operation.ExpectedProjectionDigest {
		return Operation{}, ErrNotReady
	}
	installation, err := readInstallationForUpdate(ctx, tx, operation.AppInstallationID)
	if err != nil {
		return Operation{}, err
	}
	if installation.ActiveOperationID != operationID || installation.State != transitionState(operation.Kind) {
		return Operation{}, ErrConflict
	}
	if operation.Kind == OperationRemove {
		if err := persistRemovalGate(ctx, tx, operation, removal); err != nil {
			return Operation{}, err
		}
	} else if removal != nil {
		return Operation{}, fmt.Errorf("removal receipt is only valid for remove")
	} else if _, err := ensureServingArtifactCommitmentForActivation(ctx, tx, operation); err != nil {
		return Operation{}, fmt.Errorf("verify committed serving selection before activation: %w", err)
	}
	// A stopped upgrade can resume an operation admitted by the previous
	// release directly at PROJECTION_ACKNOWLEDGED, without replaying Begin.
	// Consumer grants and Integration bindings did not exist in that request
	// grammar, so the only truthful compatibility commitment is the empty set.
	// The ensure helpers refuse to invent a non-empty selection; activation's
	// verification below remains fail-closed for any staged row or mismatched
	// commitment.
	if err := ensureStoppedUpgradeConsumerCommitmentsTx(
		ctx, tx, operation,
	); err != nil {
		return Operation{}, err
	}
	if err := verifyConsumerMetadataSelectionTx(
		ctx, tx, operation,
	); err != nil {
		return Operation{}, err
	}
	if err := activateBindingsTx(ctx, tx, operation); err != nil {
		return Operation{}, err
	}

	newState := StateActive
	switch operation.Kind {
	case OperationPrepare:
		var result sql.Result
		result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'ACTIVE',
ActiveReleaseDigest = CandidateReleaseDigest, CandidateReleaseDigest = NULL,
LastKnownGoodReleaseDigest = ?, CatalogRevision = ?, ActivationEpoch = ActivationEpoch + 1,
ActiveOperationId = NULL, ActiveAuthorizedWorkloads = CandidateAuthorizedWorkloads, CandidateAuthorizedWorkloads = NULL,
ActiveGrantedCapabilities = CandidateGrantedCapabilities, CandidateGrantedCapabilities = NULL,
ActiveResolvedDependencies = CandidateResolvedDependencies, CandidateResolvedDependencies = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE AppInstallationId = ? AND ActiveOperationId = ?`, operation.TargetAppReleaseDigest, epochText(operation.TargetCatalogRevision), operation.AppInstallationID, operationID)
		if err == nil {
			var affected int64
			affected, err = result.RowsAffected()
			if err == nil && affected != 1 {
				return Operation{}, ErrConflict
			}
		}
	case OperationUpgrade:
		var result sql.Result
		if operation.TargetAppReleaseDigest == installation.ActiveRelease {
			// Same-release refresh replaces only the complete serving selection.
			// App release and rollback lineage remain byte-for-byte unchanged.
			result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'ACTIVE',
CandidateReleaseDigest = NULL, CatalogRevision = ?, ActivationEpoch = ActivationEpoch + 1, ActiveOperationId = NULL,
ActiveAuthorizedWorkloads = CandidateAuthorizedWorkloads, CandidateAuthorizedWorkloads = NULL,
ActiveGrantedCapabilities = CandidateGrantedCapabilities, CandidateGrantedCapabilities = NULL,
ActiveResolvedDependencies = CandidateResolvedDependencies, CandidateResolvedDependencies = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE AppInstallationId = ? AND ActiveOperationId = ?`, epochText(operation.TargetCatalogRevision), operation.AppInstallationID, operationID)
		} else {
			result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'ACTIVE',
LastKnownGoodReleaseDigest = ActiveReleaseDigest, ActiveReleaseDigest = CandidateReleaseDigest, CandidateReleaseDigest = NULL,
CatalogRevision = ?, ActivationEpoch = ActivationEpoch + 1, ActiveOperationId = NULL,
ActiveAuthorizedWorkloads = CandidateAuthorizedWorkloads, CandidateAuthorizedWorkloads = NULL,
ActiveGrantedCapabilities = CandidateGrantedCapabilities, CandidateGrantedCapabilities = NULL,
ActiveResolvedDependencies = CandidateResolvedDependencies, CandidateResolvedDependencies = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE AppInstallationId = ? AND ActiveOperationId = ?`, epochText(operation.TargetCatalogRevision), operation.AppInstallationID, operationID)
		}
		if err == nil {
			var affected int64
			affected, err = result.RowsAffected()
			if err == nil && affected != 1 {
				return Operation{}, ErrConflict
			}
		}
	case OperationRollback:
		var result sql.Result
		result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'ACTIVE',
LastKnownGoodReleaseDigest = ActiveReleaseDigest, ActiveReleaseDigest = CandidateReleaseDigest, CandidateReleaseDigest = NULL,
CatalogRevision = ?, ActivationEpoch = ActivationEpoch + 1, ActiveOperationId = NULL,
ActiveAuthorizedWorkloads = CandidateAuthorizedWorkloads, CandidateAuthorizedWorkloads = NULL,
ActiveGrantedCapabilities = CandidateGrantedCapabilities, CandidateGrantedCapabilities = NULL,
ActiveResolvedDependencies = CandidateResolvedDependencies, CandidateResolvedDependencies = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE AppInstallationId = ? AND ActiveOperationId = ?`, epochText(operation.TargetCatalogRevision), operation.AppInstallationID, operationID)
		if err == nil {
			var affected int64
			affected, err = result.RowsAffected()
			if err == nil && affected != 1 {
				return Operation{}, ErrConflict
			}
		}
	case OperationRemove:
		newState = StateRemoved
		var result sql.Result
		result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'REMOVED',
CandidateReleaseDigest = NULL, CatalogRevision = ?, ActivationEpoch = ActivationEpoch + 1, ActiveOperationId = NULL,
CandidateAuthorizedWorkloads = NULL, CandidateGrantedCapabilities = NULL, CandidateResolvedDependencies = NULL,
LiveAppKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND ActiveOperationId = ?`, epochText(operation.TargetCatalogRevision), operation.AppInstallationID, operationID)
		if err == nil {
			var affected int64
			affected, err = result.RowsAffected()
			if err == nil && affected != 1 {
				return Operation{}, ErrConflict
			}
		}
	}
	if err != nil {
		return Operation{}, err
	}
	if operation.Kind == OperationRemove {
		if _, err := tx.ExecContext(ctx, `DELETE FROM _System_CatalogNamespaceReservation WHERE AppInstallationId = ?`, operation.AppInstallationID); err != nil {
			return Operation{}, err
		}
	} else {
		if _, err := tx.ExecContext(ctx, `DELETE FROM _System_CatalogNamespaceReservation
WHERE AppInstallationId = ? AND CandidateReleaseDigest IS NULL`, operation.AppInstallationID); err != nil {
			return Operation{}, err
		}
		result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogNamespaceReservation
SET State = 'ACTIVE', ActiveReleaseDigest = CandidateReleaseDigest, CandidateReleaseDigest = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND CandidateReleaseDigest = ?`, operation.AppInstallationID, operation.TargetAppReleaseDigest)
		if err != nil {
			return Operation{}, err
		}
		activated, err := result.RowsAffected()
		if err != nil {
			return Operation{}, err
		}
		var expected int64
		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_CatalogAppReleaseNamespace
WHERE AppId = ? AND AppReleaseDigest = ?`, operation.AppID, operation.TargetAppReleaseDigest).Scan(&expected); err != nil {
			return Operation{}, err
		}
		if activated != expected {
			return Operation{}, fmt.Errorf("%w: namespace candidate set is incomplete", ErrConflict)
		}
	}
	if err := catalogwriter.Commit(ctx, tx, operation.OrgID, operationID, operation.TargetCatalogRevision, operation.TargetSecurityEpoch); err != nil {
		return Operation{}, catalogWriterError(err)
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogAppOperation SET State = ?, Step = 'ACTIVATED', LastError = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6) WHERE OperationId = ? AND Step = 'PROJECTION_ACKNOWLEDGED' AND State = ?`,
		newState, operationID, transitionState(operation.Kind))
	if err != nil {
		return Operation{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return Operation{}, ErrConflict
	}
	if err := appendNextJournal(ctx, tx, operationID, "ACTIVATED", string(newState), operation.ExpectedProjectionDigest); err != nil {
		return Operation{}, err
	}
	if err := tx.Commit(); err != nil {
		return Operation{}, err
	}
	operation.State, operation.Step = newState, "ACTIVATED"
	return operation, nil
}

func ensureStoppedUpgradeConsumerCommitmentsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) error {
	type commitment struct {
		table  string
		ensure func(context.Context, *sql.Tx, Operation) error
	}
	for _, value := range []commitment{
		{
			table: "_System_CatalogConsumerGrantSet",
			ensure: func(ctx context.Context, tx *sql.Tx, operation Operation) error {
				return ensureConsumerGrantSetForOperationRetryTx(
					ctx, tx, operation, nil,
				)
			},
		},
		{
			table: "_System_CatalogIntegrationBindingSet",
			ensure: func(ctx context.Context, tx *sql.Tx, operation Operation) error {
				return ensureIntegrationBindingSetForOperationRetryTx(
					ctx, tx, operation, nil,
				)
			},
		},
	} {
		var present int
		err := tx.QueryRowContext(
			ctx,
			"SELECT 1 FROM "+value.table+" WHERE OperationId = ? FOR UPDATE",
			operation.OperationID,
		).Scan(&present)
		if err == nil {
			continue
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return fmt.Errorf(
				"read stopped-upgrade %s commitment: %w",
				value.table, err,
			)
		}
		if err := value.ensure(ctx, tx, operation); err != nil {
			return fmt.Errorf(
				"ensure stopped-upgrade %s empty commitment: %w",
				value.table, err,
			)
		}
	}
	return nil
}

func persistRemovalGate(ctx context.Context, tx *sql.Tx, operation Operation, receipt *RemovalGateReceipt) error {
	if receipt == nil {
		return fmt.Errorf("%w: remove requires all retention gate receipts", ErrNotReady)
	}
	for label, digest := range map[string]string{
		"dependency proof": receipt.DependencyProofDigest, "rollback retention proof": receipt.RollbackRetentionProofDigest,
		"work retention proof": receipt.WorkRetentionProofDigest, "audit retention proof": receipt.AuditRetentionProofDigest,
	} {
		if err := requireDigest(label, digest); err != nil {
			return err
		}
	}
	_, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogRemovalGateReceipt
(OperationId, AppInstallationId, DependencyProofDigest, RollbackRetentionProofDigest, WorkRetentionProofDigest, AuditRetentionProofDigest, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`, operation.OperationID, operation.AppInstallationID,
		receipt.DependencyProofDigest, receipt.RollbackRetentionProofDigest, receipt.WorkRetentionProofDigest, receipt.AuditRetentionProofDigest)
	if isDuplicate(err) {
		return ErrConflict
	}
	return err
}

func verifyRemovalGate(ctx context.Context, tx *sql.Tx, operation Operation, receipt *RemovalGateReceipt) error {
	if receipt == nil {
		return fmt.Errorf("%w: remove retry requires its exact retention gate receipt", ErrConflict)
	}
	var stored RemovalGateReceipt
	err := tx.QueryRowContext(ctx, `SELECT DependencyProofDigest, RollbackRetentionProofDigest,
WorkRetentionProofDigest, AuditRetentionProofDigest FROM _System_CatalogRemovalGateReceipt
WHERE OperationId = ? FOR UPDATE`, operation.OperationID).Scan(&stored.DependencyProofDigest,
		&stored.RollbackRetentionProofDigest, &stored.WorkRetentionProofDigest, &stored.AuditRetentionProofDigest)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrConflict
	}
	if err != nil {
		return err
	}
	if stored != *receipt {
		return ErrConflict
	}
	return nil
}

func (a *Authority) FailOperation(ctx context.Context, operationID, failureDigest string) error {
	if err := requireDigest("failure evidence", failureDigest); err != nil {
		return err
	}
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer rollback(tx)
	operation, err := a.readOperation(ctx, tx, operationID, true)
	if err != nil {
		return err
	}
	if operation.State == StateFailed {
		if operation.LastErrorDigest != failureDigest {
			return ErrConflict
		}
		return tx.Commit()
	}
	if operation.Step == "ACTIVATED" {
		return ErrInvalidTransition
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogAppOperation SET State = 'FAILED', LastError = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OperationId = ? AND State = ?`, failureDigest, operationID, operation.State)
	if err != nil {
		return err
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return ErrConflict
	}
	result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = 'FAILED', LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND ActiveOperationId = ? AND State = ?`, operation.AppInstallationID, operationID, operation.State)
	if err != nil {
		return err
	}
	affected, err = result.RowsAffected()
	if err != nil || affected != 1 {
		return ErrConflict
	}
	if err := appendNextJournal(ctx, tx, operationID, operation.Step, string(StateFailed), failureDigest); err != nil {
		return err
	}
	return tx.Commit()
}

// AbandonUnpublishedUpgrade releases an upgrade that never selected metadata
// or committed serving artifacts. It restores the already-active installation
// without changing its release, Catalog revision, security epoch, or activation
// epoch. The failed operation and its input commitments remain as audit facts;
// only operation-owned staged bindings and namespace candidates are retired.
func (a *Authority) AbandonUnpublishedUpgrade(
	ctx context.Context,
	operationID string,
	evidenceDigest string,
) (Operation, error) {
	if err := requireDigest("abandonment evidence", evidenceDigest); err != nil {
		return Operation{}, err
	}
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Operation{}, err
	}
	defer rollback(tx)
	if _, err := catalogadmissionfence.LockExclusive(ctx, tx); err != nil {
		return Operation{}, fmt.Errorf(
			"%w: publication admission fence: %v",
			ErrCorrupt,
			err,
		)
	}
	operationOrgID, err := readOperationOrgID(ctx, tx, operationID)
	if err != nil {
		return Operation{}, err
	}
	writerState, err := catalogwriter.Lock(ctx, tx, operationOrgID)
	if err != nil {
		return Operation{}, catalogWriterError(err)
	}
	operation, err := a.readOperation(ctx, tx, operationID, true)
	if err != nil {
		return Operation{}, err
	}
	if operation.OrgID != operationOrgID || operation.Kind != OperationUpgrade {
		return Operation{}, ErrInvalidTransition
	}
	installation, err := readInstallationForUpdate(
		ctx, tx, operation.AppInstallationID,
	)
	if err != nil {
		return Operation{}, err
	}
	if operation.Step == OperationStepAbandoned {
		if operation.State != StateFailed ||
			operation.LastErrorDigest != evidenceDigest ||
			installation.OrgID != operation.OrgID ||
			installation.AppID != operation.AppID ||
			installation.State != StateActive ||
			installation.ActiveOperationID != "" ||
			writerState.ActiveCommandID != "" {
			return Operation{}, ErrConflict
		}
		if err := tx.Commit(); err != nil {
			return Operation{}, err
		}
		return operation, nil
	}
	if operation.Step != "CANDIDATE_COMMITTED" ||
		(operation.State != StateUpgrading && operation.State != StateFailed) ||
		installation.OrgID != operation.OrgID ||
		installation.AppID != operation.AppID ||
		installation.ActiveOperationID != operationID ||
		(installation.State != StateUpgrading && installation.State != StateFailed) ||
		installation.ActiveRelease == "" ||
		installation.CandidateRelease != operation.TargetAppReleaseDigest ||
		writerState.ActiveCommandID != operationID {
		return Operation{}, ErrInvalidTransition
	}

	journalRows, err := tx.QueryContext(ctx, `SELECT Step
FROM _System_CatalogAppOperationJournal
WHERE OperationId = ? ORDER BY SequenceNumber FOR UPDATE`, operationID)
	if err != nil {
		return Operation{}, err
	}
	journalCount := 0
	for journalRows.Next() {
		journalCount++
		var step string
		if err := journalRows.Scan(&step); err != nil {
			journalRows.Close()
			return Operation{}, err
		}
		if step != "CANDIDATE_COMMITTED" {
			journalRows.Close()
			return Operation{}, ErrInvalidTransition
		}
	}
	if err := journalRows.Err(); err != nil {
		journalRows.Close()
		return Operation{}, err
	}
	if err := journalRows.Close(); err != nil {
		return Operation{}, err
	}
	if journalCount == 0 {
		return Operation{}, ErrConflict
	}

	if _, err := tx.ExecContext(ctx, `UPDATE _System_CatalogHandlerBinding
SET State = 'FAILED', ActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OperationId = ? AND State = 'STAGED'`, operationID); err != nil {
		return Operation{}, err
	}
	if _, err := tx.ExecContext(ctx, `UPDATE _System_CatalogFrontendBinding
SET State = 'FAILED', ActiveKey = NULL, OperationKey = NULL, StagingKey = NULL,
LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OperationId = ? AND State = 'STAGED'`, operationID); err != nil {
		return Operation{}, err
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM _System_CatalogNamespaceReservation
WHERE AppInstallationId = ? AND State = 'STAGED' AND ActiveReleaseDigest IS NULL`,
		operation.AppInstallationID,
	); err != nil {
		return Operation{}, err
	}
	if _, err := tx.ExecContext(ctx, `UPDATE _System_CatalogNamespaceReservation
SET CandidateReleaseDigest = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND State = 'ACTIVE'`,
		operation.AppInstallationID,
	); err != nil {
		return Operation{}, err
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation
SET State = 'ACTIVE', CandidateReleaseDigest = NULL, ActiveOperationId = NULL,
CandidateAuthorizedWorkloads = NULL, CandidateGrantedCapabilities = NULL,
CandidateResolvedDependencies = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND ActiveOperationId = ? AND State IN ('UPGRADING','FAILED')`,
		operation.AppInstallationID, operationID,
	)
	if err != nil {
		return Operation{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return Operation{}, ErrConflict
	}
	if err := catalogwriter.Release(
		ctx, tx, operation.OrgID, operationID,
	); err != nil {
		return Operation{}, catalogWriterError(err)
	}
	result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppOperation
SET State = 'FAILED', Step = ?, LastError = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OperationId = ? AND Step = 'CANDIDATE_COMMITTED' AND State IN ('UPGRADING','FAILED')`,
		OperationStepAbandoned, evidenceDigest, operationID,
	)
	if err != nil {
		return Operation{}, err
	}
	affected, err = result.RowsAffected()
	if err != nil || affected != 1 {
		return Operation{}, ErrConflict
	}
	if err := appendNextJournal(
		ctx, tx, operationID, OperationStepAbandoned,
		string(StateFailed), evidenceDigest,
	); err != nil {
		return Operation{}, err
	}
	if err := tx.Commit(); err != nil {
		return Operation{}, err
	}
	operation.State = StateFailed
	operation.Step = OperationStepAbandoned
	operation.LastErrorDigest = evidenceDigest
	return operation, nil
}

func (a *Authority) ResumeOperation(ctx context.Context, operationID string) (Operation, error) {
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return Operation{}, err
	}
	defer rollback(tx)
	operationOrgID, err := readOperationOrgID(ctx, tx, operationID)
	if err != nil {
		return Operation{}, err
	}
	writerState, err := catalogwriter.Lock(ctx, tx, operationOrgID)
	if err != nil {
		return Operation{}, catalogWriterError(err)
	}
	operation, err := a.readOperation(ctx, tx, operationID, true)
	if err != nil {
		return Operation{}, err
	}
	if operation.OrgID != operationOrgID {
		return Operation{}, ErrConflict
	}
	if operation.State != StateFailed {
		return Operation{}, ErrInvalidTransition
	}
	var installationOperation sql.NullString
	if err := tx.QueryRowContext(ctx, `SELECT ActiveOperationId FROM _System_CatalogAppInstallation WHERE AppInstallationId = ? FOR UPDATE`, operation.AppInstallationID).Scan(&installationOperation); err != nil {
		return Operation{}, err
	}
	if writerState.ActiveCommandID != operationID || installationOperation.String != operationID {
		return Operation{}, ErrConflict
	}
	state := transitionState(operation.Kind)
	result, err := tx.ExecContext(ctx, `UPDATE _System_CatalogAppOperation SET State = ?, LastError = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OperationId = ? AND State = 'FAILED'`, state, operationID)
	if err != nil {
		return Operation{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return Operation{}, ErrConflict
	}
	result, err = tx.ExecContext(ctx, `UPDATE _System_CatalogAppInstallation SET State = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE AppInstallationId = ? AND State = 'FAILED' AND ActiveOperationId = ?`, state, operation.AppInstallationID, operationID)
	if err != nil {
		return Operation{}, err
	}
	affected, err = result.RowsAffected()
	if err != nil || affected != 1 {
		return Operation{}, ErrConflict
	}
	if err := appendNextJournal(ctx, tx, operationID, operation.Step, string(state), operation.ExpectedProjectionDigest); err != nil {
		return Operation{}, err
	}
	if err := tx.Commit(); err != nil {
		return Operation{}, err
	}
	operation.State = state
	return operation, nil
}

func catalogWriterError(err error) error {
	if errors.Is(err, catalogwriter.ErrConflict) ||
		errors.Is(err, catalogwriter.ErrLineageExhausted) {
		return ErrConflict
	}
	return err
}

func (a *Authority) ReadOperation(ctx context.Context, operationID string) (Operation, error) {
	return a.readOperation(ctx, a.db, operationID, false)
}

// ReadOperationByIdempotency reads the durable operation selected by Catalog's
// unique idempotency key. It is a read-only replay discovery seam: callers
// must still call BeginOperation to validate the complete request digest and
// atomically create an operation when this lookup finds none.
func (a *Authority) ReadOperationByIdempotency(ctx context.Context, orgID, appID, idempotencyKey string) (Operation, error) {
	if err := validateOrgAndOpaqueID(orgID, "idempotency key", idempotencyKey); err != nil {
		return Operation{}, err
	}
	if len(idempotencyKey) > 255 {
		return Operation{}, fmt.Errorf("idempotency key must fit its Catalog field")
	}
	if err := appauthority.ValidateAppID(appID); err != nil {
		return Operation{}, err
	}
	var operationID string
	err := a.db.QueryRowContext(ctx, `SELECT OperationId FROM _System_CatalogAppOperation
WHERE OrgId = ? AND AppId = ? AND IdempotencyKey = ?`, orgID, appID, idempotencyKey).Scan(&operationID)
	if errors.Is(err, sql.ErrNoRows) {
		return Operation{}, ErrNotFound
	}
	if err != nil {
		return Operation{}, err
	}
	return a.ReadOperation(ctx, operationID)
}