//go:build integration

package catalogapp

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"crypto/tls"
	"database/sql"
	"encoding/base64"
	"errors"
	"fmt"
	"os"
	"regexp"
	"slices"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	database "github.com/nexuscrm/db"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appidentitykeydeclarationv1"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundlekeyv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/layoutbundleslotv1"
	pagelayoutresolutionv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/pagelayoutresolutionv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/bootstrapimpl"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/connectorbindingdefinitionauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/pingcap-inc/nexuscrm-controlplane/releaseartifact"
)

func TestCatalogAppTiDB(t *testing.T) {
	db, cleanup := openCatalogTiDB(t)
	defer cleanup()
	authority, err := New(db, Options{})
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.EnsureSchema(context.Background()); err != nil {
		t.Fatal(err)
	}
	registerTestPublisher(t, authority)
	t.Run("LifecycleRecoveryAndRemoval", func(t *testing.T) {
		testCatalogAppTiDBLifecycleRecoveryAndRemoval(t, db, authority)
	})
	t.Run("SameReleaseServingRefresh", func(t *testing.T) {
		testCatalogAppTiDBSameReleaseServingRefresh(t, db, authority)
	})
	t.Run("WriterLineageIsOrgWideAcrossInstallations", func(t *testing.T) {
		testCatalogAppTiDBWriterLineageAcrossInstallations(t, authority)
	})
	t.Run("IdentityReleaseNamespaceAndGrantConflicts", func(t *testing.T) {
		testCatalogAppTiDBIdentityReleaseNamespaceAndGrantConflicts(t, db, authority)
	})
	t.Run("ConcurrentCASAndConversionRollback", func(t *testing.T) {
		testCatalogAppTiDBConcurrentCASAndConversionRollback(t, db, authority)
	})
	t.Run("FrontendOnlyBindingCAS", func(t *testing.T) {
		testCatalogAppTiDBFrontendOnlyBindingCAS(t, db, authority)
	})
	t.Run("VerifiedArtifactImportUsesReleaseAuthority", func(t *testing.T) {
		testCatalogAppTiDBVerifiedArtifactImport(t, db, authority)
	})
	t.Run("SealedIdentityKeyDeclarationsUseReleaseEvidence", func(t *testing.T) {
		testCatalogAppTiDBSealedIdentityKeyDeclarations(t, authority)
	})
	t.Run("ServingArtifactCommitFencesRetries", func(t *testing.T) {
		testCatalogAppTiDBServingArtifactCommit(t, authority)
	})
	t.Run("WorkloadOnlyHandlerStaysOutsideServingSelection", func(t *testing.T) {
		testCatalogAppTiDBWorkloadOnlyHandlerSelection(t, db, authority)
	})
	t.Run("LegacyProjectionAcknowledgementBackfillsServingCommitment", func(t *testing.T) {
		testCatalogAppTiDBLegacyProjectionBackfill(t, db, authority)
	})
	t.Run("ExternalPublishedArtifactUsesReleaseAuthority", func(t *testing.T) {
		testCatalogAppTiDBExternalPublishedArtifact(t, db)
	})
	t.Run("AppLayoutBundleSlotRegistry", func(t *testing.T) {
		testCatalogAppTiDBLayoutBundleSlotRegistry(t, db, authority)
	})
	t.Run("OpaqueIdentityBytewiseStorage", func(t *testing.T) {
		testCatalogAppTiDBOpaqueIdentityBytewiseStorage(t, db)
	})
}

func testCatalogAppTiDBWriterLineageAcrossInstallations(t *testing.T, authority *Authority) {
	t.Helper()
	const orgID = "org-writer-lineage"
	releaseA := testRegisteredReleaseVersionForAppNamespace(
		t, "io.example.writer-a", "io.example.writer-a", "1.0.0", "Subject",
	)
	releaseB := testRegisteredReleaseVersionForAppNamespace(
		t, "io.example.writer-b", "io.example.writer-b", "1.0.0", "Subject",
	)
	registerTestReleases(t, authority, releaseA)
	registerTestReleases(t, authority, releaseB)
	writer := newMemoryRuntimeProjectionWriter()
	activate := func(release RegisteredRelease, idempotency string, catalogRevision, securityEpoch uint64) Operation {
		t.Helper()
		operation, err := authority.BeginOperation(
			t.Context(),
			testOperationRequest(release, orgID, idempotency, catalogRevision, securityEpoch),
		)
		if err != nil {
			t.Fatalf("begin %s: %v", idempotency, err)
		}
		stageTestBinding(t, authority, operation, release, 0)
		if _, err := authority.CommitServingArtifacts(t.Context(), operation.OperationID); err != nil {
			t.Fatalf("commit %s serving artifacts: %v", idempotency, err)
		}
		if err := authority.PublishRuntimeSecurityProjection(t.Context(), operation.OperationID, writer); err != nil {
			t.Fatalf("publish %s security projection: %v", idempotency, err)
		}
		completed, err := authority.CompleteOperation(t.Context(), operation.OperationID, nil)
		if err != nil {
			t.Fatalf("complete %s: %v", idempotency, err)
		}
		return completed
	}

	first := activate(releaseA, "writer-a", 0, 0)
	activate(releaseB, "writer-b", 1, 1)
	installation, err := authority.ReadInstallation(t.Context(), first.AppInstallationID)
	if err != nil {
		t.Fatal(err)
	}
	lineage, err := authority.ReadWriterLineage(t.Context(), orgID)
	if err != nil {
		t.Fatal(err)
	}
	if installation.CatalogRevision != 1 || lineage.OrgID != orgID ||
		lineage.RuntimeAudience != "runtime" || lineage.CatalogRevision != 2 ||
		lineage.SecurityEpoch != 2 || lineage.ActiveOperationID != "" {
		t.Fatalf("installation history = %d, org writer lineage = %#v", installation.CatalogRevision, lineage)
	}
}

func testCatalogAppTiDBSameReleaseServingRefresh(t *testing.T, db *sql.DB, authority *Authority) {
	release1 := testRegisteredReleaseVersionForApp(t, "io.example.refresh", "1.0.0", "Subject")
	release2 := testRegisteredReleaseVersionForApp(t, "io.example.refresh", "2.0.0", "Summary")
	registerTestReleases(t, authority, release1, release2)
	writer := newMemoryRuntimeProjectionWriter()

	activate := func(request BeginOperationRequest, release RegisteredRelease, ownershipEpoch uint64) Operation {
		t.Helper()
		operation, err := authority.BeginOperation(context.Background(), request)
		if err != nil {
			t.Fatal(err)
		}
		stageTestBinding(t, authority, operation, release, ownershipEpoch)
		if _, err := authority.CommitServingArtifacts(context.Background(), operation.OperationID); err != nil {
			t.Fatal(err)
		}
		if err := authority.PublishRuntimeSecurityProjection(context.Background(), operation.OperationID, writer); err != nil {
			t.Fatal(err)
		}
		completed, err := authority.CompleteOperation(context.Background(), operation.OperationID, nil)
		if err != nil {
			t.Fatal(err)
		}
		return completed
	}

	prepare := activate(testOperationRequest(release1, "org-refresh", "prepare", 0, 0), release1, 0)
	wrongTarget := testOperationRequest(release2, "org-refresh", "wrong-target-refresh", 1, 1)
	wrongTarget.Kind = OperationUpgrade
	wrongTarget.AppInstallationID = prepare.AppInstallationID
	wrongTarget.RefreshActiveRelease = true
	wrongTarget.ContinuationIntentDigest = testDigest("wrong-target-refresh")
	if _, err := authority.BeginOperation(context.Background(), wrongTarget); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("refresh to a different release = %v, want invalid transition", err)
	}

	upgrade := testOperationRequest(release2, "org-refresh", "upgrade", 1, 1)
	upgrade.Kind = OperationUpgrade
	upgrade.AppInstallationID = prepare.AppInstallationID
	active2 := activate(upgrade, release2, 1)

	implicit := testOperationRequest(release2, "org-refresh", "implicit-refresh", 2, 2)
	implicit.Kind = OperationUpgrade
	implicit.AppInstallationID = prepare.AppInstallationID
	if _, err := authority.BeginOperation(context.Background(), implicit); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("implicit same-release upgrade = %v, want invalid transition", err)
	}

	refresh := testOperationRequest(release2, "org-refresh", "refresh", 2, 2)
	refresh.Kind = OperationUpgrade
	refresh.AppInstallationID = prepare.AppInstallationID
	refresh.RefreshActiveRelease = true
	refresh.ContinuationIntentDigest = testDigest("refresh-serving-selection")
	active3 := activate(refresh, release2, 2)
	if active3.TargetCatalogRevision != 3 || active3.TargetSecurityEpoch != 3 {
		t.Fatalf("refresh lineage = catalog %d security %d, want 3/3", active3.TargetCatalogRevision, active3.TargetSecurityEpoch)
	}

	var activeRelease, lastKnownGood string
	var catalogRevision, securityEpoch []byte
	if err := db.QueryRow(`SELECT i.ActiveReleaseDigest, i.LastKnownGoodReleaseDigest,
i.CatalogRevision, s.LastSecurityEpoch
FROM _System_CatalogAppInstallation AS i
JOIN _System_CatalogOrgState AS s ON s.OrgId = i.OrgId
WHERE i.AppInstallationId = ?`, active2.AppInstallationID).Scan(
		&activeRelease, &lastKnownGood, &catalogRevision, &securityEpoch,
	); err != nil {
		t.Fatal(err)
	}
	if activeRelease != release2.AppReleaseDigest || lastKnownGood != release1.AppReleaseDigest ||
		string(catalogRevision) != "3" || string(securityEpoch) != "3" {
		t.Fatalf("same-release refresh changed release lineage: active=%s last-good=%s catalog=%s security=%s",
			activeRelease, lastKnownGood, catalogRevision, securityEpoch)
	}
}

func testCatalogAppTiDBOpaqueIdentityBytewiseStorage(
	t *testing.T,
	db *sql.DB,
) {
	t.Helper()
	const (
		operationID    = "00000000-0000-0000-0000-0000000000b1"
		installationID = "00000000-0000-0000-0000-0000000000b2"
	)
	opaqueValues := []string{"A", "B", "a", "ä"}
	for index, workload := range opaqueValues {
		grantID := fmt.Sprintf("%064x", index+1)
		if _, err := db.Exec(`INSERT INTO _System_CatalogConsumerGrant
(ConsumerGrantId, OrgId, AppInstallationId, OperationId,
 AuthorizedWorkloadId, CapabilityId, CreatedDate)
VALUES (?, 'org-bytewise', ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
			grantID, installationID, operationID, workload,
			consumerv1.AppMetadataCapabilityName,
		); err != nil {
			t.Fatalf("insert bytewise consumer grant %q: %v", workload, err)
		}
	}
	rows, err := db.Query(`SELECT AuthorizedWorkloadId
FROM _System_CatalogConsumerGrant
WHERE OperationId = ?
ORDER BY BINARY AuthorizedWorkloadId`, operationID)
	if err != nil {
		t.Fatal(err)
	}
	var storedWorkloads []string
	for rows.Next() {
		var workload string
		if err := rows.Scan(&workload); err != nil {
			_ = rows.Close()
			t.Fatal(err)
		}
		storedWorkloads = append(storedWorkloads, workload)
	}
	if err := rows.Close(); err != nil {
		t.Fatal(err)
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(storedWorkloads, opaqueValues) {
		t.Fatalf(
			"consumer workload byte order = %q, want %q",
			storedWorkloads, opaqueValues,
		)
	}

	for index, ownerPack := range opaqueValues {
		bindingID := fmt.Sprintf("%064x", index+101)
		if _, err := db.Exec(`INSERT INTO _System_CatalogIntegrationBinding
(IntegrationBindingId, OrgId, AppInstallationId, OperationId,
 AuthorizedWorkloadId, Kind, OwnerPack, PackInstallId, PackPayloadDigest,
 BusinessKey, DefinitionVersion, CredentialRefs, CreatedDate)
VALUES (?, 'org-bytewise', ?, ?, 'workload-bytewise', ?, ?, 'pack-install',
        ?, 'business-key', 1, CAST('[]' AS JSON), CURRENT_TIMESTAMP(6))`,
			bindingID, installationID, operationID,
			consumerv1.BindingConnectorPushContract, ownerPack,
			"nexus-pack-sha256-v1:"+strings.Repeat("a", 64),
		); err != nil {
			t.Fatalf("insert bytewise integration owner %q: %v", ownerPack, err)
		}
	}
	rows, err = db.Query(`SELECT OwnerPack
FROM _System_CatalogIntegrationBinding
WHERE OperationId = ?
ORDER BY BINARY AuthorizedWorkloadId, BINARY Kind, BINARY OwnerPack,
         BINARY PackInstallId, BINARY BusinessKey`, operationID)
	if err != nil {
		t.Fatal(err)
	}
	var storedOwners []string
	for rows.Next() {
		var owner string
		if err := rows.Scan(&owner); err != nil {
			_ = rows.Close()
			t.Fatal(err)
		}
		storedOwners = append(storedOwners, owner)
	}
	if err := rows.Close(); err != nil {
		t.Fatal(err)
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(storedOwners, opaqueValues) {
		t.Fatalf(
			"integration owner byte order = %q, want %q",
			storedOwners, opaqueValues,
		)
	}
}

const testCatalogAppLayoutBundleSlotKey = layoutbundlekeyv1.LocalComponentKey("key256:7143e970b024190d65ab67bb767208c64435a235e7a48c699d145673e308468b")

func testCatalogAppTiDBLayoutBundleSlotRegistry(t *testing.T, db *sql.DB, authority *Authority) {
	t.Run("LineageReuseAndOmission", func(t *testing.T) {
		appID := "io.example.layout-slot-ledger"
		targetA := testCatalogAppLayoutBundleTargetID(appID, "LayoutA")
		targetB := testCatalogAppLayoutBundleTargetID(appID, "LayoutB")
		v1, declarationV1 := testCatalogAppLayoutBundleSlotRelease(t, appID, "1.0.0", 1, "LayoutA", nil)
		v2, declarationV2 := testCatalogAppLayoutBundleSlotRelease(t, appID, "2.0.0", 2, "LayoutB", &targetA)
		historicalV1, historicalDeclarationV1 := testCatalogAppLayoutBundleSlotRelease(t, appID, "2.1.0", 1, "LayoutA", nil)
		omission := testCatalogAppLayoutBundleOmissionRelease(t, appID, "2.2.0")
		laterHistoricalV1, laterHistoricalDeclarationV1 := testCatalogAppLayoutBundleSlotRelease(t, appID, "2.3.0", 1, "LayoutA", nil)
		v3, declarationV3 := testCatalogAppLayoutBundleSlotRelease(t, appID, "3.0.0", 3, "LayoutA", &targetB)

		if err := authority.ReserveAppID(context.Background(), "io.example", appID, 1); err != nil {
			t.Fatal(err)
		}
		registerOneRelease(t, authority, v1)
		assertCatalogAppLayoutBundleSlotBinding(t, authority, v1, declarationV1)
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 1, declarationV1.RevisionDigest, targetA)

		registerOneRelease(t, authority, v2)
		assertCatalogAppLayoutBundleSlotBinding(t, authority, v2, declarationV2)
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 2, declarationV2.RevisionDigest, targetB)

		registerOneRelease(t, authority, historicalV1)
		assertCatalogAppLayoutBundleSlotBinding(t, authority, historicalV1, historicalDeclarationV1)
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 2, declarationV2.RevisionDigest, targetB)

		beforeOmission := catalogAppLayoutBundleSlotStateBytes(t, db, appID)
		registerOneRelease(t, authority, omission)
		omittedBindings, err := authority.ReadReleaseLayoutBundleSlots(context.Background(), appID, omission.AppReleaseDigest)
		if err != nil {
			t.Fatal(err)
		}
		if len(omittedBindings) != 0 {
			t.Fatalf("omitting AppRelease slot bindings = %#v, want empty", omittedBindings)
		}
		afterOmission := catalogAppLayoutBundleSlotStateBytes(t, db, appID)
		if !bytes.Equal(afterOmission, beforeOmission) {
			t.Fatalf("omitting AppRelease changed global slot reservation or ledger:\nbefore=%s\nafter=%s", beforeOmission, afterOmission)
		}

		registerOneRelease(t, authority, laterHistoricalV1)
		assertCatalogAppLayoutBundleSlotBinding(t, authority, laterHistoricalV1, laterHistoricalDeclarationV1)
		if afterReuse := catalogAppLayoutBundleSlotStateBytes(t, db, appID); !bytes.Equal(afterReuse, beforeOmission) {
			t.Fatalf("historical reuse after omission changed global slot reservation or ledger:\nbefore=%s\nafter=%s", beforeOmission, afterReuse)
		}
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 2, declarationV2.RevisionDigest, targetB)

		registrationV3 := testVerifiedReleaseRegistration(t, v3, time.Now().UTC())
		registeredV3, err := authority.registerVerifiedRelease(context.Background(), registrationV3)
		if err != nil {
			t.Fatal(err)
		}
		if registeredV3.AppReleaseDigest != v3.AppReleaseDigest {
			t.Fatalf("registered v3 digest = %s, want %s", registeredV3.AppReleaseDigest, v3.AppReleaseDigest)
		}
		assertCatalogAppLayoutBundleSlotBinding(t, authority, v3, declarationV3)
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 3, declarationV3.RevisionDigest, targetA)
		beforeRetry := catalogAppLayoutBundleSlotStateBytes(t, db, appID)
		retryV3, err := authority.registerVerifiedRelease(context.Background(), registrationV3)
		if err != nil || retryV3.AppReleaseDigest != v3.AppReleaseDigest {
			t.Fatalf("exact v3 registration retry = %#v, %v", retryV3, err)
		}
		if afterRetry := catalogAppLayoutBundleSlotStateBytes(t, db, appID); !bytes.Equal(afterRetry, beforeRetry) {
			t.Fatalf("exact registration retry changed global slot reservation or ledger:\nbefore=%s\nafter=%s", beforeRetry, afterRetry)
		}
	})

	t.Run("ConcurrentDivergentSuccessorCAS", func(t *testing.T) {
		appID := "io.example.layout-slot-race"
		baseTarget := testCatalogAppLayoutBundleTargetID(appID, "Base")
		v1, declarationV1 := testCatalogAppLayoutBundleSlotRelease(t, appID, "1.0.0", 1, "Base", nil)
		candidateA, declarationA := testCatalogAppLayoutBundleSlotRelease(t, appID, "2.0.0-a", 2, "CandidateA", &baseTarget)
		candidateB, declarationB := testCatalogAppLayoutBundleSlotRelease(t, appID, "2.0.0-b", 2, "CandidateB", &baseTarget)
		if err := authority.ReserveAppID(context.Background(), "io.example", appID, 1); err != nil {
			t.Fatal(err)
		}
		registerOneRelease(t, authority, v1)
		assertCatalogAppLayoutBundleSlotBinding(t, authority, v1, declarationV1)

		type registrationResult struct {
			candidate   RegisteredRelease
			declaration layoutbundleslotv1.Declaration
			registered  RegisteredRelease
			err         error
		}
		start := make(chan struct{})
		results := make(chan registrationResult, 2)
		now := time.Now().UTC()
		for _, input := range []struct {
			release     RegisteredRelease
			declaration layoutbundleslotv1.Declaration
		}{
			{release: candidateA, declaration: declarationA},
			{release: candidateB, declaration: declarationB},
		} {
			input := input
			registration := testVerifiedReleaseRegistration(t, input.release, now)
			go func() {
				<-start
				registered, err := authority.registerVerifiedRelease(context.Background(), registration)
				results <- registrationResult{candidate: input.release, declaration: input.declaration, registered: registered, err: err}
			}()
		}
		close(start)
		first, second := <-results, <-results
		var winner, loser registrationResult
		successes, conflicts := 0, 0
		for _, result := range []registrationResult{first, second} {
			switch {
			case result.err == nil:
				successes++
				winner = result
			case errors.Is(result.err, ErrConflict):
				conflicts++
				loser = result
			default:
				t.Fatalf("concurrent successor registration for %s = %v, want success or conflict", result.candidate.AppReleaseDigest, result.err)
			}
		}
		if successes != 1 || conflicts != 1 {
			t.Fatalf("concurrent divergent successor outcomes: successes=%d conflicts=%d first=%v second=%v", successes, conflicts, first.err, second.err)
		}
		if winner.registered.AppReleaseDigest != winner.candidate.AppReleaseDigest {
			t.Fatalf("winning registration digest = %s, want %s", winner.registered.AppReleaseDigest, winner.candidate.AppReleaseDigest)
		}
		assertCatalogAppLayoutBundleSlotBinding(t, authority, winner.candidate, winner.declaration)
		assertCatalogAppLayoutBundleSlotHead(t, authority, appID, 2, winner.declaration.RevisionDigest, winner.declaration.TargetComponentID)
		assertCatalogAppReleaseRowsAbsent(t, db, loser.candidate)
		if _, err := authority.ReadRelease(context.Background(), appID, loser.candidate.AppReleaseDigest); !errors.Is(err, ErrNotFound) {
			t.Fatalf("losing AppRelease read = %v, want not found", err)
		}
	})
}

func testCatalogAppTiDBLegacyProjectionBackfill(t *testing.T, db *sql.DB, authority *Authority) {
	release := testRegisteredReleaseVersionForApp(t, "io.example.legacy-projection", "1.0.0", "Subject")
	registerTestReleases(t, authority, release)
	operation, err := authority.BeginOperation(context.Background(), testOperationRequest(release, "org-legacy-projection", "prepare-legacy", 0, 0))
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, operation, release, 0)
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	defer rollback(tx)
	result, err := tx.Exec(`UPDATE _System_CatalogAppOperation
SET Step = 'PROJECTION_ACKNOWLEDGED', AcknowledgedSecurityEpoch = ?, AcknowledgedProjectionDigest = ?
WHERE OperationId = ? AND Step = 'CANDIDATE_COMMITTED'`, epochText(operation.TargetSecurityEpoch),
		operation.ExpectedProjectionDigest, operation.OperationID)
	if err != nil {
		t.Fatal(err)
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		t.Fatalf("create legacy acknowledged operation: affected=%d err=%v", affected, err)
	}
	if err := appendNextJournal(context.Background(), tx, operation.OperationID, "PROJECTION_ACKNOWLEDGED",
		string(operation.State), operation.ExpectedProjectionDigest); err != nil {
		t.Fatal(err)
	}
	// Simulate a stopped upgrade resuming an operation admitted by the
	// pre-consumer-view release. That release could not have persisted either
	// empty-set commitment and does not replay Begin before completion.
	for _, table := range []string{
		"_System_CatalogConsumerGrantSet",
		"_System_CatalogIntegrationBindingSet",
	} {
		result, err := tx.Exec(
			"DELETE FROM "+table+" WHERE OperationId = ?",
			operation.OperationID,
		)
		if err != nil {
			t.Fatal(err)
		}
		if affected, err := result.RowsAffected(); err != nil || affected != 1 {
			t.Fatalf(
				"remove current %s commitment: affected=%d err=%v",
				table, affected, err,
			)
		}
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	completed, err := authority.CompleteOperation(context.Background(), operation.OperationID, nil)
	if err != nil || completed.Step != "ACTIVATED" {
		t.Fatalf("complete legacy acknowledged operation = %#v, %v", completed, err)
	}
	if _, err := authority.ReadServingArtifactCommitment(context.Background(), operation.OperationID); err != nil {
		t.Fatalf("read backfilled serving commitment: %v", err)
	}
	var commitmentCount int
	if err := db.QueryRow(`SELECT
(SELECT COUNT(*) FROM _System_CatalogConsumerGrantSet WHERE OperationId = ?) +
(SELECT COUNT(*) FROM _System_CatalogIntegrationBindingSet WHERE OperationId = ?)`,
		operation.OperationID, operation.OperationID,
	).Scan(&commitmentCount); err != nil {
		t.Fatal(err)
	}
	if commitmentCount != 2 {
		t.Fatalf(
			"stopped-upgrade empty commitments = %d, want 2",
			commitmentCount,
		)
	}
}

func testCatalogAppTiDBServingArtifactCommit(t *testing.T, authority *Authority) {
	release := testRegisteredReleaseVersionForApp(t, "io.example.artifact-fence", "1.0.0", "Subject")
	registerTestReleases(t, authority, release)
	operation, err := authority.BeginOperation(context.Background(), testOperationRequest(release, "org-artifact-fence", "prepare-fenced", 0, 0))
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, operation, release, 0)
	commitment, err := authority.CommitServingArtifacts(context.Background(), operation.OperationID)
	if err != nil {
		t.Fatal(err)
	}
	if retry, err := authority.CommitServingArtifacts(context.Background(), operation.OperationID); err != nil || retry != commitment {
		t.Fatalf("same serving artifact retry: %v", err)
	}
	if got, err := authority.ReadServingArtifactCommitment(context.Background(), operation.OperationID); err != nil || got != commitment {
		t.Fatalf("committed serving selection = %q, %v", got, err)
	}
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	wrongBinding := HandlerBindingRequest{OperationID: operation.OperationID,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
		ServiceReleaseDigest:       testDigest("service-" + release.SemanticVersion),
		CompatibilityReceiptDigest: testDigest("different-receipt")}
	if _, err := authority.StageHandlerBinding(context.Background(), wrongBinding); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("binding after serving selection commit = %v, want invalid transition", err)
	}
	writer := newMemoryRuntimeProjectionWriter()
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), operation.OperationID, writer); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), operation.OperationID, nil); err != nil {
		t.Fatal(err)
	}
}

func testCatalogAppTiDBWorkloadOnlyHandlerSelection(
	t *testing.T,
	db *sql.DB,
	authority *Authority,
) {
	base := testRegisteredReleaseVersionForApp(
		t, "io.example.partial-selection", "1.0.0", "Subject",
	)
	manifest := base.Package.Manifest
	manifest.RequiredEngineCapabilities = append(
		manifest.RequiredEngineCapabilities,
		"runtime.identity.session-lifecycle.v1",
	)
	slices.Sort(manifest.RequiredEngineCapabilities)
	manifest.ProvidedCapabilities = append(
		manifest.ProvidedCapabilities,
		appauthority.ProvidedCapability{
			CapabilityID: "io.example.partial-selection.workload.v1",
			Kind:         appauthority.CapabilityKindServiceHandler,
			HandlerRef:   "io.example.partial-selection.handler.workload.v1",
			RequiredRuntimeCapabilities: []string{
				"runtime.identity.session-lifecycle.v1",
			},
		},
	)
	slices.SortFunc(
		manifest.ProvidedCapabilities,
		func(left, right appauthority.ProvidedCapability) int {
			return strings.Compare(left.CapabilityID, right.CapabilityID)
		},
	)
	metadataDigest, err := appauthority.MetadataBundleDigest(
		base.Package.Bundle, manifest,
	)
	if err != nil {
		t.Fatal(err)
	}
	manifest.MetadataBundleDigest = metadataDigest
	verified, err := appauthority.VerifyPackage(appauthority.Package{
		Manifest: manifest,
		Bundle:   base.Package.Bundle,
		ComponentBlobs: append(
			[]appauthority.ComponentBlob(nil),
			base.Package.ComponentBlobs...,
		),
	})
	if err != nil {
		t.Fatal(err)
	}
	release := registeredRelease(verified)
	registerTestReleases(t, authority, release)

	request := testOperationRequest(
		release, "org-partial-selection", "prepare", 0, 0,
	)
	selected := release.Package.Manifest.ProvidedCapabilities[0]
	request.SelectedHandlerCapabilities = []string{selected.CapabilityID}
	if _, err := authority.BeginOperation(context.Background(), request); err == nil ||
		!strings.Contains(err.Error(), "exactly partition") {
		t.Fatalf("unproved partial selection error = %v", err)
	}
	var workloadOnly appauthority.ProvidedCapability
	for _, capability := range release.Package.Manifest.ProvidedCapabilities {
		if capability.CapabilityID != selected.CapabilityID {
			workloadOnly = capability
			break
		}
	}
	request.WorkloadHandlerAdmissions = []WorkloadHandlerAdmission{{
		CapabilityID: workloadOnly.CapabilityID, AuthorizedWorkloadID: "workload-1",
		RequiredRuntimeCapabilities: append(
			[]string(nil), workloadOnly.RequiredRuntimeCapabilities...,
		),
	}}
	operation, err := authority.BeginOperation(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := authority.StageHandlerBinding(
		context.Background(),
		HandlerBindingRequest{
			OperationID:          operation.OperationID,
			CapabilityID:         selected.CapabilityID,
			HandlerRef:           selected.HandlerRef,
			WorkloadID:           "workload-1",
			ServiceReleaseDigest: testDigest("partial-service"),
			CompatibilityReceiptDigest: testDigest(
				"partial-compatibility",
			),
		},
	); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CommitServingArtifacts(
		context.Background(), operation.OperationID,
	); err != nil {
		t.Fatal(err)
	}
	writer := newMemoryRuntimeProjectionWriter()
	if err := authority.PublishRuntimeSecurityProjection(
		context.Background(), operation.OperationID, writer,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(
		context.Background(), operation.OperationID, nil,
	); err != nil {
		t.Fatal(err)
	}
	var activeCount int
	if err := db.QueryRow(`SELECT COUNT(*)
FROM _System_CatalogHandlerBinding
WHERE AppInstallationId = ? AND State = 'ACTIVE'`,
		operation.AppInstallationID,
	).Scan(&activeCount); err != nil {
		t.Fatal(err)
	}
	if activeCount != 1 {
		t.Fatalf("active selected HandlerBindings = %d, want 1", activeCount)
	}
}

// testCatalogAppTiDBExternalPublishedArtifact lets the L2 evidence workflow
// supply exact published L1 bytes without making controlplane depend on L1 or
// teaching Catalog about a repository layout. An unset fixture is an explicit
// skip so the ordinary neutral A1c suite remains self-contained.
func testCatalogAppTiDBExternalPublishedArtifact(t *testing.T, db *sql.DB) {
	packagePath := os.Getenv("NEXUSCRM_CATALOGAPP_EXTERNAL_PACKAGE")
	attestationPath := os.Getenv("NEXUSCRM_CATALOGAPP_EXTERNAL_ATTESTATION")
	publicKeyBase64 := os.Getenv("NEXUSCRM_CATALOGAPP_EXTERNAL_PUBLIC_KEY_BASE64")
	provided := 0
	for _, value := range []string{packagePath, attestationPath, publicKeyBase64} {
		if value != "" {
			provided++
		}
	}
	if provided == 0 {
		t.Skip("no external published AppRelease fixture requested")
	}
	if provided != 3 {
		t.Fatal("external AppRelease proof requires package, attestation, and public key together")
	}
	packageRaw, err := os.ReadFile(packagePath)
	if err != nil {
		t.Fatal(err)
	}
	attestationRaw, err := os.ReadFile(attestationPath)
	if err != nil {
		t.Fatal(err)
	}
	publicKey, err := base64.StdEncoding.DecodeString(publicKeyBase64)
	if err != nil {
		t.Fatal(err)
	}
	verified, err := releaseartifact.Verify(packageRaw, attestationRaw, ed25519.PublicKey(publicKey))
	if err != nil {
		t.Fatal(err)
	}
	verificationTime := verified.Attestation.IssuedAt.Add(10 * time.Minute)
	authority, err := New(db, Options{Now: func() time.Time { return verificationTime }})
	if err != nil {
		t.Fatal(err)
	}
	epoch, err := authority.RegisterPublisher(context.Background(), PublisherTrust{
		PublisherID: verified.Attestation.PublisherID,
		Active:      true,
		Keys: []PublisherKey{{
			KeyID: verified.Attestation.KeyID, Algorithm: releaseartifact.SignatureAlgorithm, PublicKey: publicKey,
			ValidFrom: verified.Attestation.IssuedAt.Add(-time.Hour), ValidUntil: verificationTime.Add(time.Hour),
		}},
	}, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.ReserveAppID(context.Background(), verified.Attestation.PublisherID, verified.Package.Manifest.AppID, epoch); err != nil {
		t.Fatal(err)
	}
	imported, err := authority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, fixtureCompatibilityEnvelope(verified.Package))
	if err != nil {
		t.Fatal(err)
	}
	if imported.AppID != verified.Package.Manifest.AppID || imported.AppReleaseDigest != verified.Package.AppReleaseDigest {
		t.Fatalf("Catalog imported %#v, want %s %s", imported, verified.Package.Manifest.AppID, verified.Package.AppReleaseDigest)
	}
	stored, err := authority.ReadRelease(context.Background(), imported.AppID, imported.AppReleaseDigest)
	if err != nil {
		t.Fatal(err)
	}
	if stored.Package.Manifest.MetadataBundleDigest != verified.Package.Manifest.MetadataBundleDigest {
		t.Fatalf("Catalog stored metadata digest %s, want %s", stored.Package.Manifest.MetadataBundleDigest, verified.Package.Manifest.MetadataBundleDigest)
	}
	retryAuthority, err := New(db, Options{Now: func() time.Time { return verificationTime.Add(time.Minute) }})
	if err != nil {
		t.Fatal(err)
	}
	replayed, err := retryAuthority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, fixtureCompatibilityEnvelope(verified.Package))
	if err != nil || replayed.AppReleaseDigest != imported.AppReleaseDigest {
		t.Fatalf("exact external artifact replay = %#v, %v", replayed, err)
	}
}

// fixtureCompatibilityEnvelope mirrors the requirements of already-verified
// fixture bytes. It proves the import/authority composition only; independent
// compatibility rejection tests remain the evidence for host mismatch.
func fixtureCompatibilityEnvelope(pkg appauthority.VerifiedPackage) releaseartifact.HostCompatibility {
	protocols := make([]releaseartifact.SupportedProtocol, len(pkg.Manifest.RequiredProtocols))
	for i, required := range pkg.Manifest.RequiredProtocols {
		protocols[i] = releaseartifact.SupportedProtocol{ProtocolID: required.ProtocolID, Major: required.Major, Minor: required.MinimumMinor}
	}
	kinds := make([]releaseartifact.SupportedComponentKind, 0, len(pkg.Bundle.Components))
	seen := make(map[releaseartifact.SupportedComponentKind]struct{}, len(pkg.Bundle.Components))
	for _, component := range pkg.Bundle.Components {
		kind := releaseartifact.SupportedComponentKind{ComponentKind: component.ComponentKind, ComponentFormatVersion: component.ComponentFormatVersion}
		if _, duplicate := seen[kind]; duplicate {
			continue
		}
		seen[kind] = struct{}{}
		kinds = append(kinds, kind)
	}
	slices.SortFunc(kinds, func(a, b releaseartifact.SupportedComponentKind) int {
		if a.ComponentKind < b.ComponentKind {
			return -1
		}
		if a.ComponentKind > b.ComponentKind {
			return 1
		}
		return int(a.ComponentFormatVersion) - int(b.ComponentFormatVersion)
	})
	return releaseartifact.HostCompatibility{Protocols: protocols,
		EngineCapabilities: slices.Clone(pkg.Manifest.RequiredEngineCapabilities), ComponentKinds: kinds}
}

func testCatalogAppTiDBVerifiedArtifactImport(t *testing.T, db *sql.DB, authority *Authority) {
	seed := sha256.Sum256([]byte("catalogapp-integration-release-key"))
	privateKey := ed25519.NewKeyFromSeed(seed[:])
	publicKey := privateKey.Public().(ed25519.PublicKey)
	now := time.Now().UTC().Truncate(time.Microsecond)
	epoch, err := authority.RegisterPublisher(context.Background(), PublisherTrust{PublisherID: "io.import", Active: true,
		Keys: []PublisherKey{{KeyID: "release-key-1", Algorithm: releaseartifact.SignatureAlgorithm, PublicKey: publicKey,
			ValidFrom: now.Add(-time.Hour), ValidUntil: now.Add(time.Hour)}}}, 0)
	if err != nil {
		t.Fatal(err)
	}
	appID := "io.import.crm"
	if err := authority.ReserveAppID(context.Background(), "io.import", appID, epoch); err != nil {
		t.Fatal(err)
	}
	release := testRegisteredReleaseVersionForApp(t, appID, "1.0.0", "Subject")
	manifest := release.Package.Manifest
	manifest.PublisherID = "io.import"
	manifest.MetadataBundleDigest, err = appauthority.MetadataBundleDigest(release.Package.Bundle, manifest)
	if err != nil {
		t.Fatal(err)
	}
	verified, err := appauthority.VerifyPackage(appauthority.Package{Manifest: manifest, Bundle: release.Package.Bundle, ComponentBlobs: release.Package.ComponentBlobs})
	if err != nil {
		t.Fatal(err)
	}
	_, packageRaw, err := releaseartifact.BuildPackage(verified.Package)
	if err != nil {
		t.Fatal(err)
	}
	packageHash := sha256.Sum256(packageRaw)
	provenance := releaseartifact.Provenance{SchemaVersion: releaseartifact.ProvenanceSchemaVersion,
		SourceRepository: "example/crm-app", SourceRevision: "revision-1", SourcePath: "metadata",
		SourceTreeDigest: "sha256:" + fmt.Sprintf("%x", packageHash[:]), BuilderID: "catalogapp-integration/v1", ContractModuleVersion: "v0.1.8"}
	_, attestationRaw, err := releaseartifact.SignAttestation(verified, provenance, "release-key-1", now, privateKey)
	if err != nil {
		t.Fatal(err)
	}
	imported, err := authority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, releaseartifact.HostCompatibility{
		Protocols: []releaseartifact.SupportedProtocol{}, EngineCapabilities: []string{"runtime.records.read.v1"},
		ComponentKinds: []releaseartifact.SupportedComponentKind{{ComponentKind: verified.Bundle.Components[0].ComponentKind,
			ComponentFormatVersion: verified.Bundle.Components[0].ComponentFormatVersion}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if imported.AppReleaseDigest != verified.AppReleaseDigest || imported.AppID != appID {
		t.Fatalf("imported release = %#v, want %s %s", imported, appID, verified.AppReleaseDigest)
	}
	retryAuthority, err := New(db, Options{Now: func() time.Time { return now.Add(10 * time.Minute) }})
	if err != nil {
		t.Fatal(err)
	}
	replayed, err := retryAuthority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, releaseartifact.HostCompatibility{
		Protocols: []releaseartifact.SupportedProtocol{}, EngineCapabilities: []string{"runtime.records.read.v1"},
		ComponentKinds: []releaseartifact.SupportedComponentKind{{ComponentKind: verified.Bundle.Components[0].ComponentKind,
			ComponentFormatVersion: verified.Bundle.Components[0].ComponentFormatVersion}},
	})
	if err != nil || replayed.AppReleaseDigest != imported.AppReleaseDigest {
		t.Fatalf("exact artifact replay = %#v, %v", replayed, err)
	}
	stored, err := authority.ReadRelease(context.Background(), appID, verified.AppReleaseDigest)
	if err != nil || stored.AppReleaseDigest != verified.AppReleaseDigest {
		t.Fatalf("stored imported release = %#v, %v", stored, err)
	}
	sealed, err := authority.ReadSealedRelease(context.Background(), appID, verified.AppReleaseDigest)
	if err != nil || sealed.AppReleaseDigest != verified.AppReleaseDigest {
		t.Fatalf("sealed imported release = %#v, %v", sealed, err)
	}
	if _, err := authority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, releaseartifact.HostCompatibility{
		Protocols: []releaseartifact.SupportedProtocol{}, EngineCapabilities: []string{},
		ComponentKinds: []releaseartifact.SupportedComponentKind{{ComponentKind: verified.Bundle.Components[0].ComponentKind,
			ComponentFormatVersion: verified.Bundle.Components[0].ComponentFormatVersion}},
	}); err == nil || !strings.Contains(err.Error(), "unavailable") {
		t.Fatalf("incompatible host import error = %v", err)
	}
}

func testCatalogAppTiDBSealedIdentityKeyDeclarations(t *testing.T, authority *Authority) {
	t.Helper()
	seed := sha256.Sum256([]byte("catalogapp-integration-identity-key-declaration"))
	privateKey := ed25519.NewKeyFromSeed(seed[:])
	publicKey := privateKey.Public().(ed25519.PublicKey)
	now := time.Now().UTC().Truncate(time.Microsecond)
	const (
		publisherID = "io.identity.import"
		appID       = "io.identity.import.crm"
		keyID       = "identity-declaration-key-1"
	)
	epoch, err := authority.RegisterPublisher(context.Background(), PublisherTrust{
		PublisherID: publisherID,
		Active:      true,
		Keys: []PublisherKey{{
			KeyID: keyID, Algorithm: releaseartifact.SignatureAlgorithm, PublicKey: publicKey,
			ValidFrom: now.Add(-time.Hour), ValidUntil: now.Add(time.Hour),
		}},
	}, 0)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.ReserveAppID(context.Background(), publisherID, appID, epoch); err != nil {
		t.Fatal(err)
	}
	verified := testIdentityKeyDeclarationPackageForApp(t, appID, publisherID, "1.0.0", false)
	_, packageRaw, err := releaseartifact.BuildPackage(verified.Package)
	if err != nil {
		t.Fatal(err)
	}
	packageHash := sha256.Sum256(packageRaw)
	_, attestationRaw, err := releaseartifact.SignAttestation(verified, releaseartifact.Provenance{
		SchemaVersion:         releaseartifact.ProvenanceSchemaVersion,
		SourceRepository:      "example/identity-declarations",
		SourceRevision:        "identity-declarations-1",
		SourcePath:            "metadata",
		SourceTreeDigest:      "sha256:" + fmt.Sprintf("%x", packageHash[:]),
		BuilderID:             "catalogapp-integration/v1",
		ContractModuleVersion: "v0.2.39",
	}, keyID, now.Add(-time.Minute), privateKey)
	if err != nil {
		t.Fatal(err)
	}
	host := releaseartifact.HostCompatibility{
		Protocols:          []releaseartifact.SupportedProtocol{},
		EngineCapabilities: []string{},
		ComponentKinds: []releaseartifact.SupportedComponentKind{
			{ComponentKind: appidentitykeydeclarationv1.ProfileComponentKind, ComponentFormatVersion: appidentitykeydeclarationv1.ComponentFormatVersion},
			{ComponentKind: appidentitykeydeclarationv1.RoleComponentKind, ComponentFormatVersion: appidentitykeydeclarationv1.ComponentFormatVersion},
		},
	}
	if _, err := authority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, host); err != nil {
		t.Fatal(err)
	}
	declarations, err := authority.ReadSealedIdentityKeyDeclarations(context.Background(), appID, verified.AppReleaseDigest)
	if err != nil {
		t.Fatal(err)
	}
	if len(declarations.Profiles) != 1 || len(declarations.Roles) != 1 ||
		declarations.Profiles[0].AppID != appID || declarations.Profiles[0].AppReleaseDigest != verified.AppReleaseDigest ||
		declarations.Profiles[0].Key.NamespaceID != appID || declarations.Profiles[0].Key.LocalKey != "profile-a" ||
		declarations.Roles[0].Key.NamespaceID != appID || declarations.Roles[0].Key.LocalKey != "role-a" {
		t.Fatalf("sealed App identity declarations = %#v", declarations)
	}
}

func testCatalogAppTiDBLifecycleRecoveryAndRemoval(t *testing.T, db *sql.DB, authority *Authority) {
	release1 := testRegisteredReleaseVersionForApp(t, "io.example.lifecycle", "1.0.0", "Subject")
	release2 := testRegisteredReleaseVersionForApp(t, "io.example.lifecycle", "2.0.0", "Summary")
	registerTestReleases(t, authority, release1, release2)
	conflictingAttestation := testVerifiedReleaseRegistration(t, release1, time.Now().UTC())
	conflictingAttestation.artifact.Attestation.AttestationDigest = testDigest("conflicting-attestation")
	if _, err := authority.registerVerifiedRelease(context.Background(), conflictingAttestation); !errors.Is(err, ErrConflict) {
		t.Fatalf("same App release with conflicting attestation = %v, want conflict", err)
	}
	publisher, err := authority.ReadPublisher(context.Background(), release1.PublisherID)
	if err != nil {
		t.Fatal(err)
	}
	changedKey := publisher.Keys[0].PublisherKey
	changedKey.PublicKey = []byte("different-public-key")
	if _, err := authority.RegisterPublisher(context.Background(), PublisherTrust{PublisherID: publisher.PublisherID,
		Active: true, Keys: []PublisherKey{changedKey}}, publisher.OwnershipEpoch); !errors.Is(err, ErrConflict) {
		t.Fatalf("same KeyID with different key material = %v, want conflict", err)
	}
	publisherAfterConflict, err := authority.ReadPublisher(context.Background(), release1.PublisherID)
	if err != nil || publisherAfterConflict.OwnershipEpoch != publisher.OwnershipEpoch ||
		!bytes.Equal(publisherAfterConflict.Keys[0].PublicKey, testImportPublicKey()) {
		t.Fatalf("key conflict was not atomic: %#v, %v", publisherAfterConflict, err)
	}

	prepareRequest := testOperationRequest(release1, "org-lifecycle", "prepare-1", 0, 0)
	prepare, err := authority.BeginOperation(context.Background(), prepareRequest)
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, prepare, release1, 0)
	if _, err := authority.CommitServingArtifacts(context.Background(), prepare.OperationID); err != nil {
		t.Fatal(err)
	}
	writer := &ambiguousRuntimeWriter{delegate: newMemoryRuntimeProjectionWriter(), ambiguousOnce: true}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, writer); err == nil {
		t.Fatal("ambiguous Runtime commit did not surface its transport error")
	}
	if operation, err := authority.ReadOperation(context.Background(), prepare.OperationID); err != nil || operation.Step != OperationStepServingArtifactsCommitted {
		t.Fatalf("operation advanced without Catalog acknowledgement: %#v %v", operation, err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, writer); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, writer); err != nil {
		t.Fatalf("acknowledged projection retry = %v", err)
	}
	active1, err := authority.CompleteOperation(context.Background(), prepare.OperationID, nil)
	if err != nil || active1.State != StateActive {
		t.Fatalf("complete prepare = %#v, %v", active1, err)
	}
	retry, err := authority.BeginOperation(context.Background(), prepareRequest)
	if err != nil || retry.OperationID != prepare.OperationID || retry.AppInstallationID != prepare.AppInstallationID {
		t.Fatalf("idempotent prepare retry = %#v, %v", retry, err)
	}
	capability := release1.Package.Manifest.ProvidedCapabilities[0]
	serviceOnlyRequest := HandlerBindingRequest{AppInstallationID: prepare.AppInstallationID,
		AppReleaseDigest: release1.AppReleaseDigest, CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef,
		WorkloadID: "workload-1", ServiceReleaseDigest: testDigest("service-only-2"), ExpectedOwnershipEpoch: 1,
		CompatibilityReceiptDigest: testDigest("service-only-compatibility-2")}
	serviceOnlyBinding, err := authority.StageHandlerBinding(context.Background(), serviceOnlyRequest)
	if err != nil {
		t.Fatal(err)
	}
	serviceOnlyRetry, err := authority.StageHandlerBinding(context.Background(), serviceOnlyRequest)
	if err != nil || serviceOnlyRetry.HandlerBindingID != serviceOnlyBinding.HandlerBindingID {
		t.Fatalf("service-only binding retry = %#v, %v", serviceOnlyRetry, err)
	}
	if _, err := authority.ActivateStagedHandlerBinding(context.Background(), serviceOnlyBinding.HandlerBindingID, 0); !errors.Is(err, ErrConflict) {
		t.Fatalf("stale service-only activation = %v, want conflict", err)
	}
	if _, err := authority.ActivateStagedHandlerBinding(context.Background(), serviceOnlyBinding.HandlerBindingID, 1); err != nil {
		t.Fatal(err)
	}
	if retry, err := authority.ActivateStagedHandlerBinding(context.Background(), serviceOnlyBinding.HandlerBindingID, 1); err != nil || retry.State != BindingActive {
		t.Fatalf("service-only activation retry = %#v, %v", retry, err)
	}
	var serviceActive, serviceDraining int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogHandlerBinding
WHERE AppInstallationId = ? AND State = 'ACTIVE'`, prepare.AppInstallationID).Scan(&serviceActive); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogHandlerBinding
WHERE AppInstallationId = ? AND State = 'DRAINING'`, prepare.AppInstallationID).Scan(&serviceDraining); err != nil {
		t.Fatal(err)
	}
	if serviceActive != 1 || serviceDraining != 1 {
		t.Fatalf("service-only ownership flip: active=%d draining=%d", serviceActive, serviceDraining)
	}

	upgradeRequest := testOperationRequest(release2, "org-lifecycle", "upgrade-1", 1, 1)
	upgradeRequest.Kind = OperationUpgrade
	upgradeRequest.AppInstallationID = prepare.AppInstallationID
	upgrade, err := authority.BeginOperation(context.Background(), upgradeRequest)
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, upgrade, release2, 2)
	if _, err := authority.CommitServingArtifacts(context.Background(), upgrade.OperationID); err != nil {
		t.Fatal(err)
	}
	failure := testDigest("transient-upgrade-failure")
	failingAfterPublication := &failOperationAfterRuntimeWriter{delegate: writer, authority: authority,
		operationID: upgrade.OperationID, failureDigest: failure}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), upgrade.OperationID, failingAfterPublication); !errors.Is(err, ErrConflict) {
		t.Fatalf("publication acknowledgement racing operation failure = %v, want conflict", err)
	}
	failedOperation, err := authority.ReadOperation(context.Background(), upgrade.OperationID)
	if err != nil || failedOperation.State != StateFailed || failedOperation.Step != OperationStepServingArtifactsCommitted {
		t.Fatalf("failure after Runtime commit advanced Catalog acknowledgement: %#v, %v", failedOperation, err)
	}
	var activeRelease, candidateRelease string
	var activeWorkloads, candidateWorkloads []byte
	if err := db.QueryRow(`SELECT ActiveReleaseDigest, CandidateReleaseDigest, ActiveAuthorizedWorkloads, CandidateAuthorizedWorkloads
FROM _System_CatalogAppInstallation WHERE AppInstallationId = ?`, prepare.AppInstallationID).Scan(
		&activeRelease, &candidateRelease, &activeWorkloads, &candidateWorkloads); err != nil {
		t.Fatal(err)
	}
	if activeRelease != release1.AppReleaseDigest || candidateRelease != release2.AppReleaseDigest ||
		string(activeWorkloads) != `["workload-1"]` || string(candidateWorkloads) != `["workload-1"]` {
		t.Fatalf("failed upgrade damaged active/candidate separation: active=%s candidate=%s workloads=%s/%s",
			activeRelease, candidateRelease, activeWorkloads, candidateWorkloads)
	}
	var namespaceActive, namespaceCandidate sql.NullString
	var namespaceState string
	if err := db.QueryRow(`SELECT ActiveReleaseDigest, CandidateReleaseDigest, State
FROM _System_CatalogNamespaceReservation WHERE OrgId = ? AND NamespaceId = ?`,
		"org-lifecycle", "io.example.issues").Scan(&namespaceActive, &namespaceCandidate, &namespaceState); err != nil {
		t.Fatal(err)
	}
	if namespaceActive.String != release1.AppReleaseDigest || namespaceCandidate.String != release2.AppReleaseDigest || namespaceState != "ACTIVE" {
		t.Fatalf("failed upgrade damaged active namespace authority: active=%#v candidate=%#v state=%s",
			namespaceActive, namespaceCandidate, namespaceState)
	}
	if _, err := authority.ResumeOperation(context.Background(), upgrade.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), upgrade.OperationID, writer); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), upgrade.OperationID, nil); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT ActiveReleaseDigest, CandidateReleaseDigest, State
FROM _System_CatalogNamespaceReservation WHERE OrgId = ? AND NamespaceId = ?`,
		"org-lifecycle", "io.example.issues").Scan(&namespaceActive, &namespaceCandidate, &namespaceState); err != nil {
		t.Fatal(err)
	}
	if namespaceActive.String != release2.AppReleaseDigest || namespaceCandidate.Valid || namespaceState != "ACTIVE" {
		t.Fatalf("activated namespace authority: active=%#v candidate=%#v state=%s",
			namespaceActive, namespaceCandidate, namespaceState)
	}
	var activeBindings, drainingBindings int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogHandlerBinding WHERE AppInstallationId = ? AND State = 'ACTIVE'`, prepare.AppInstallationID).Scan(&activeBindings); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogHandlerBinding WHERE AppInstallationId = ? AND State = 'DRAINING'`, prepare.AppInstallationID).Scan(&drainingBindings); err != nil {
		t.Fatal(err)
	}
	if activeBindings != 1 || drainingBindings != 2 {
		t.Fatalf("binding states: active=%d draining=%d", activeBindings, drainingBindings)
	}
	rollbackRequest := testOperationRequest(release1, "org-lifecycle", "rollback-1", 2, 2)
	rollbackRequest.Kind = OperationRollback
	rollbackRequest.AppInstallationID = prepare.AppInstallationID
	rollbackOperation, err := authority.BeginOperation(context.Background(), rollbackRequest)
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, rollbackOperation, release1, 3)
	if _, err := authority.CommitServingArtifacts(context.Background(), rollbackOperation.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), rollbackOperation.OperationID, writer); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), rollbackOperation.OperationID, nil); err != nil {
		t.Fatal(err)
	}

	removeRequest := BeginOperationRequest{Kind: OperationRemove, OrgID: "org-lifecycle", AppInstallationID: prepare.AppInstallationID,
		AppID: release1.AppID, IdempotencyKey: "remove-1", RuntimeAudience: "runtime", ExpectedCatalogRevision: 3, ExpectedSecurityEpoch: 3}
	remove, err := authority.BeginOperation(context.Background(), removeRequest)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), remove.OperationID, writer); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), remove.OperationID, nil); !errors.Is(err, ErrNotReady) {
		t.Fatalf("remove without retention receipts = %v, want not ready", err)
	}
	removalReceipt := &RemovalGateReceipt{
		DependencyProofDigest: testDigest("dependencies"), RollbackRetentionProofDigest: testDigest("rollback"),
		WorkRetentionProofDigest: testDigest("work"), AuditRetentionProofDigest: testDigest("audit"),
	}
	if _, err := authority.CompleteOperation(context.Background(), remove.OperationID, removalReceipt); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), remove.OperationID, removalReceipt); err != nil {
		t.Fatalf("exact remove retry = %v", err)
	}
	conflictingReceipt := *removalReceipt
	conflictingReceipt.AuditRetentionProofDigest = testDigest("different-audit")
	if _, err := authority.CompleteOperation(context.Background(), remove.OperationID, &conflictingReceipt); !errors.Is(err, ErrConflict) {
		t.Fatalf("conflicting remove retry = %v, want conflict", err)
	}
	var state string
	var liveKey sql.NullString
	var retainedRelease string
	if err := db.QueryRow(`SELECT State, LiveAppKey, ActiveReleaseDigest FROM _System_CatalogAppInstallation WHERE AppInstallationId = ?`, prepare.AppInstallationID).Scan(&state, &liveKey, &retainedRelease); err != nil {
		t.Fatal(err)
	}
	if state != string(StateRemoved) || liveKey.Valid || retainedRelease != release1.AppReleaseDigest {
		t.Fatalf("removed tombstone = state %s live %#v release %s", state, liveKey, retainedRelease)
	}
}

func testCatalogAppTiDBIdentityReleaseNamespaceAndGrantConflicts(t *testing.T, db *sql.DB, authority *Authority) {
	release := testReleaseWithHandlerConsumerDependenciesForApp(
		t, "io.example.conflicts",
	)
	registerTestReleases(t, authority, release)
	const (
		integrationPackName      = "pack-integration"
		integrationPackInstallID = "pack-install-integration-1"
		previousPackInstallID    = "pack-install-integration-0"
		integrationPackDigest    = "nexus-pack-sha256-v1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	)
	integrationGenerationID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		integrationPackName,
		integrationPackInstallID,
	)
	previousGenerationID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		integrationPackName,
		previousPackInstallID,
	)
	ensureCatalogAppTestPackAuthorityTables(t, db)
	if _, err := db.Exec(`INSERT INTO _System_PackGeneration
	(Id, PackName, PackInstallId, Category, Status, DesiredPayloadHash, AppliedPayloadHash,
	IsDeleted)
	VALUES
	(?, ?, ?, ?, 'active', ?, ?, 0),
	(?, ?, ?, ?, 'active', ?, ?, 0)`,
		previousGenerationID,
		integrationPackName,
		previousPackInstallID,
		string(packs.ConnectorPack),
		integrationPackDigest,
		integrationPackDigest,
		integrationGenerationID,
		integrationPackName,
		integrationPackInstallID,
		string(packs.ConnectorPack),
		integrationPackDigest,
		integrationPackDigest,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_PackActivation
	(Id, PackName, ActivePackInstallId, PreviousPackInstallId, ActivationEpoch,
	ActivatedDate, IsDeleted)
	VALUES (?, ?, ?, NULL, 2, CURRENT_TIMESTAMP(6), 0)`,
		"catalog-app-test-activation",
		integrationPackName,
		integrationPackInstallID,
	); err != nil {
		t.Fatal(err)
	}

	conflictingSemver := testRegisteredReleaseVersionForApp(t, release.AppID, release.SemanticVersion, "DifferentField")
	if _, err := authority.registerVerifiedRelease(context.Background(), testVerifiedReleaseRegistration(t, conflictingSemver, time.Now().UTC())); !errors.Is(err, ErrConflict) {
		t.Fatalf("same AppID/semver with different digest = %v, want conflict", err)
	}

	now := time.Now().UTC()
	secondPublisher := PublisherTrust{PublisherID: "io.other", Active: true, Keys: []PublisherKey{{KeyID: "key-other",
		Algorithm: "test", PublicKey: []byte("other-public"), ValidFrom: now.Add(-time.Hour), ValidUntil: now.Add(time.Hour)}}}
	if _, err := authority.RegisterPublisher(context.Background(), secondPublisher, 0); err != nil {
		t.Fatal(err)
	}
	if err := authority.ReserveAppID(context.Background(), secondPublisher.PublisherID, release.AppID, 1); !errors.Is(err, ErrConflict) {
		t.Fatalf("AppID collision = %v, want conflict", err)
	}
	transferAppID := "io.example.transfer"
	if err := authority.ReserveAppID(context.Background(), release.PublisherID, transferAppID, 1); err != nil {
		t.Fatal(err)
	}
	transfer := AppIDTransfer{AppID: transferAppID, CurrentPublisherID: release.PublisherID,
		NewPublisherID: secondPublisher.PublisherID, ExpectedOwnershipEpoch: 1,
		OperatorApproved: true, AuditDigest: testDigest("transfer")}
	if _, err := authority.TransferAppID(context.Background(), transfer); !errors.Is(err, ErrConflict) {
		t.Fatalf("unapproved AppID transfer = %v, want conflict", err)
	}
	transfer.CurrentPublisherApproved = true
	if epoch, err := authority.TransferAppID(context.Background(), transfer); err != nil || epoch != 2 {
		t.Fatalf("approved AppID transfer epoch = %d, %v", epoch, err)
	}

	invalidGrant := testOperationRequest(release, "org-conflicts", "invalid-grant", 0, 0)
	invalidGrant.GrantedCapabilities = []string{"runtime.records.write.v1"}
	if _, err := authority.BeginOperation(context.Background(), invalidGrant); err == nil || !strings.Contains(err.Error(), "exactly match") {
		t.Fatalf("capability conflict = %v", err)
	}
	invalidDependency := testOperationRequest(release, "org-conflicts", "invalid-dependency", 0, 0)
	invalidDependency.ResolvedDependencies = []ResolvedDependency{{Alias: "dependency", AppID: "io.example.dependency",
		AppReleaseDigest: testDigest("dependency")}}
	if _, err := authority.BeginOperation(context.Background(), invalidDependency); err == nil || !strings.Contains(err.Error(), "incomplete") {
		t.Fatalf("dependency conflict = %v", err)
	}
	invalidConsumerGrant := testOperationRequest(
		release, "org-conflicts", "invalid-consumer-grant", 0, 0,
	)
	invalidConsumerGrant.ConsumerGrants = []ConsumerGrant{{
		AuthorizedWorkloadID: "workload-integration",
		CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
	}}
	if _, err := authority.BeginOperation(
		context.Background(), invalidConsumerGrant,
	); err == nil || !strings.Contains(err.Error(), "not authorized") {
		t.Fatalf("consumer workload authority conflict = %v", err)
	}

	prepareRequest := testOperationRequest(release, "org-conflicts", "prepare", 0, 0)
	prepareRequest.AuthorizedWorkloads = []string{
		"workload-1", "workload-integration",
	}
	prepareRequest.ConsumerGrants = []ConsumerGrant{
		{
			AuthorizedWorkloadID: "workload-integration",
			CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
		},
		{
			AuthorizedWorkloadID: "workload-1",
			CapabilityID:         consumerv1.AppMetadataV2CapabilityName,
		},
		{
			AuthorizedWorkloadID: "workload-1",
			CapabilityID:         pagelayoutresolutionv1.CapabilityName,
		},
	}
	prepareRequest.IntegrationBindings = []IntegrationBindingSelection{{
		AuthorizedWorkloadID: "workload-integration",
		Binding: consumerv1.IntegrationBinding{
			Kind:               consumerv1.BindingConnectorInboundRoute,
			OwnerPack:          integrationPackName,
			PackInstallationID: integrationPackInstallID,
			PackPayloadDigest:  integrationPackDigest,
			BusinessKey:        "route-main",
			DefinitionVersion:  1,
			CredentialRefs:     []string{},
		},
	}}
	insertCatalogAppTestDefinitionSet(
		t,
		db,
		integrationGenerationID,
		prepareRequest.IntegrationBindings[0].Binding,
	)
	prepare, err := authority.BeginOperation(context.Background(), prepareRequest)
	if err != nil {
		t.Fatal(err)
	}
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	wrongHandlerWorkload := HandlerBindingRequest{OperationID: prepare.OperationID,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-integration",
		ServiceReleaseDigest: testDigest("service-wrong-workload"), ExpectedOwnershipEpoch: 0,
		CompatibilityReceiptDigest: testDigest("compatibility-wrong-workload")}
	if _, err := authority.StageHandlerBinding(
		context.Background(), wrongHandlerWorkload,
	); !errors.Is(err, ErrConflict) {
		t.Fatalf("handler workload without required Catalog grants = %v, want conflict", err)
	}
	initialHandlerRequest := HandlerBindingRequest{OperationID: prepare.OperationID,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
		ServiceReleaseDigest: testDigest("service-1.0.0"), ExpectedOwnershipEpoch: 0,
		CompatibilityReceiptDigest: testDigest("compatibility-1.0.0")}
	failedHandler, err := authority.StageHandlerBinding(context.Background(), initialHandlerRequest)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.FailStagedHandlerBinding(context.Background(), failedHandler.HandlerBindingID); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.StageHandlerBinding(context.Background(), initialHandlerRequest); !errors.Is(err, ErrInvalidTransition) {
		t.Fatalf("failed handler epoch retry = %v, want invalid transition", err)
	}
	initialHandlerRequest.ExpectedOwnershipEpoch = failedHandler.OwnershipEpoch
	replacementHandler, err := authority.StageHandlerBinding(context.Background(), initialHandlerRequest)
	if err != nil || replacementHandler.HandlerBindingID == failedHandler.HandlerBindingID || replacementHandler.OwnershipEpoch != 2 {
		t.Fatalf("replacement handler binding = %#v, %v", replacementHandler, err)
	}
	// Reject missing consumer selection before the operation leaves the step
	// where the caller can still complete this same candidate.
	if _, err := authority.CommitServingArtifacts(context.Background(), prepare.OperationID); !errors.Is(err, ErrNotReady) {
		t.Fatalf("consumer serving without snapshot = %v, want not ready", err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, newMemoryRuntimeProjectionWriter()); !errors.Is(err, ErrNotReady) {
		t.Fatalf("incomplete consumer published security: %v", err)
	}
	if _, err := authority.CommitOperationMetadataSelectionWithEffectiveComponents(context.Background(), prepare.OperationID, testDigest("consumer-selection"), testRequiredEffectiveMetadataComponents()); err != nil {
		t.Fatalf("same candidate cannot select missing metadata: %v", err)
	}
	if _, err := authority.CommitServingArtifacts(context.Background(), prepare.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, newMemoryRuntimeProjectionWriter()); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), prepare.OperationID, nil); err != nil {
		t.Fatal(err)
	}
	retry, err := authority.BeginOperation(context.Background(), prepareRequest)
	if err != nil || retry.OperationID != prepare.OperationID {
		t.Fatalf("consumer grant operation exact retry = %#v, %v", retry, err)
	}
	var (
		activeConsumerGrantCount int
		distinctGrantOperations  int
	)
	if err := db.QueryRow(`SELECT
COUNT(*), COUNT(DISTINCT OperationId)
FROM _System_CatalogConsumerGrant
WHERE AppInstallationId = ?`,
		prepare.AppInstallationID,
	).Scan(&activeConsumerGrantCount, &distinctGrantOperations); err != nil {
		t.Fatal(err)
	}
	if activeConsumerGrantCount != 3 || distinctGrantOperations != 1 {
		t.Fatalf(
			"atomic active consumer grant set = count %d operations %d, want 3/1",
			activeConsumerGrantCount, distinctGrantOperations,
		)
	}
	var (
		activeIntegrationBindingCount int
		bindingOperationID            string
	)
	if err := db.QueryRow(`SELECT COUNT(*), MIN(OperationId)
FROM _System_CatalogIntegrationBinding
WHERE AppInstallationId = ?`,
		prepare.AppInstallationID,
	).Scan(
		&activeIntegrationBindingCount, &bindingOperationID,
	); err != nil {
		t.Fatal(err)
	}
	if activeIntegrationBindingCount != 1 ||
		bindingOperationID != prepare.OperationID {
		t.Fatalf(
			"atomic active IntegrationBinding set = count %d operation %s, want 1/%s",
			activeIntegrationBindingCount, bindingOperationID, prepare.OperationID,
		)
	}
	otherRelease := testRegisteredReleaseForApp(t, "io.example.other-app", "1.0.0")
	if err := authority.ReserveAppID(context.Background(), otherRelease.PublisherID, otherRelease.AppID, 1); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.registerVerifiedRelease(context.Background(), testVerifiedReleaseRegistration(t, otherRelease, time.Now().UTC())); err != nil {
		t.Fatal(err)
	}
	otherPrepare := testOperationRequest(otherRelease, "org-conflicts", "namespace-conflict", 1, 1)
	if _, err := authority.BeginOperation(context.Background(), otherPrepare); !errors.Is(err, ErrConflict) {
		t.Fatalf("namespace ownership conflict = %v, want conflict", err)
	}
}

func ensureCatalogAppTestPackAuthorityTables(t *testing.T, db *sql.DB) {
	t.Helper()
	if err := bootstrapimpl.EnsureL0SchemaDDL(database.NewConnectionFromDB(db)); err != nil {
		t.Fatalf("initialize canonical L0 schema: %v", err)
	}
}

func insertCatalogAppTestDefinitionSet(
	t *testing.T,
	db *sql.DB,
	generationID string,
	bindings ...consumerv1.IntegrationBinding,
) {
	t.Helper()
	if len(bindings) == 0 {
		t.Fatal("Catalog App definition-set fixture requires bindings")
	}
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(bindings))
	for _, binding := range bindings {
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
			connectorbindingownerv1.AuthoringInput{
				Identity:            connectorbindingownerv1.IdentityFromBinding(binding),
				CanonicalDefinition: []byte(`{}`),
			},
		)
		if err != nil {
			t.Fatalf("build %s/%s definition envelope: %v", binding.Kind, binding.BusinessKey, err)
		}
		definitions = append(definitions, definition)
	}
	coordinate := connectorbindingownerv1.DefinitionSetCoordinate{
		OwnerPack:          bindings[0].OwnerPack,
		PackInstallationID: bindings[0].PackInstallationID,
		PackPayloadDigest:  bindings[0].PackPayloadDigest,
	}
	set, err := connectorbindingownerv1.NewDefinitionSet(coordinate, definitions)
	if err != nil {
		t.Fatalf("build Catalog App definition set: %v", err)
	}
	now := time.Now().UTC()
	if _, err := db.Exec(`INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
(PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest, CreatedDate,
LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?)`,
		generationID,
		set.Header.Protocol,
		strconv.FormatUint(set.Header.DefinitionCount, 10),
		set.Header.OrderedSetDigest,
		now,
		now,
	); err != nil {
		t.Fatalf("insert Catalog App definition set: %v", err)
	}
	for _, entry := range set.Entries {
		if _, err := db.Exec(`INSERT INTO _System_PackGenerationConnectorBindingDefinition
(Id, PackGenerationId, PackInstallId, DefinitionOrdinal, Kind, OwnerPack,
PackPayloadDigest, BusinessKey, DefinitionVersion, DefinitionEnvelopeDigest,
DefinitionEnvelope, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			connectorbindingdefinitionauthority.DefinitionStorageID(
				generationID,
				entry.Kind,
				entry.BusinessKey,
			),
			generationID,
			entry.PackInstallationID,
			strconv.FormatUint(entry.Ordinal, 10),
			string(entry.Kind),
			entry.OwnerPack,
			entry.PackPayloadDigest,
			entry.BusinessKey,
			strconv.FormatUint(entry.DefinitionVersion, 10),
			entry.EnvelopeDigest,
			entry.EnvelopeJSON,
			now,
			now,
		); err != nil {
			t.Fatalf("insert %s/%s Catalog App definition: %v", entry.Kind, entry.BusinessKey, err)
		}
	}
}

func testCatalogAppTiDBConcurrentCASAndConversionRollback(t *testing.T, db *sql.DB, authority *Authority) {
	release := testRegisteredReleaseVersionForApp(t, "io.example.conversion", "1.0.0", "Subject")
	registerTestReleases(t, authority, release)

	hijack := testConversionRequest(t, release, "org-conversion-hijack", "legacy-hijack")
	plainOperation, err := authority.BeginOperation(context.Background(), hijack.Operation)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := authority.ConvertLegacyPack(context.Background(), hijack); !errors.Is(err, ErrConflict) {
		t.Fatalf("conversion reused direct operation = %v, want conflict", err)
	}
	var hijackRows int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogCRMConversionJournal WHERE OrgId = ?`,
		"org-conversion-hijack").Scan(&hijackRows); err != nil {
		t.Fatal(err)
	}
	if hijackRows != 0 {
		t.Fatalf("direct operation gained %d conversion mappings", hijackRows)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogHandlerBinding WHERE AppInstallationId = ?`,
		plainOperation.AppInstallationID).Scan(&hijackRows); err != nil {
		t.Fatal(err)
	}
	if hijackRows != 0 {
		t.Fatalf("rejected conversion left %d conversion bindings on direct operation", hijackRows)
	}

	request := testConversionRequest(t, release, "org-conversion", "legacy-1")
	conversion, err := authority.ConvertLegacyPack(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	retry, err := authority.ConvertLegacyPack(context.Background(), request)
	if err != nil || retry.AppInstallationID != conversion.AppInstallationID || retry.OperationID != conversion.OperationID {
		t.Fatalf("conversion retry = %#v, %v", retry, err)
	}
	failureDigest := testDigest("conversion-operation-failure")
	if err := authority.FailOperation(context.Background(), conversion.OperationID, failureDigest); err != nil {
		t.Fatal(err)
	}
	failedConversion, err := authority.ReadConversion(context.Background(), conversion.MigrationID)
	if err != nil || failedConversion.State != string(StateFailed) || failedConversion.Step != "CANDIDATE_COMMITTED" {
		t.Fatalf("failed conversion lifecycle view = %#v, %v", failedConversion, err)
	}
	if _, err := authority.ResumeOperation(context.Background(), conversion.OperationID); err != nil {
		t.Fatal(err)
	}
	resumedConversion, err := authority.ReadConversion(context.Background(), conversion.MigrationID)
	if err != nil || resumedConversion.State != string(StatePreparing) || resumedConversion.Step != "CANDIDATE_COMMITTED" {
		t.Fatalf("resumed conversion lifecycle view = %#v, %v", resumedConversion, err)
	}
	if _, err := authority.CommitServingArtifacts(context.Background(), conversion.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), conversion.OperationID, newMemoryRuntimeProjectionWriter()); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), conversion.OperationID, nil); err != nil {
		t.Fatal(err)
	}
	completedConversion, err := authority.ReadConversion(context.Background(), conversion.MigrationID)
	if err != nil || completedConversion.State != string(StateActive) || completedConversion.Step != "ACTIVATED" {
		t.Fatalf("completed conversion lifecycle view = %#v, %v", completedConversion, err)
	}
	completedRetry, err := authority.ConvertLegacyPack(context.Background(), request)
	if err != nil || completedRetry.State != string(StateActive) || completedRetry.Step != "ACTIVATED" {
		t.Fatalf("completed conversion retry = %#v, %v", completedRetry, err)
	}
	conflict := request
	conflict.Legacy.AppliedPayloadHash = testDigest("other-incarnation")
	conflict.Legacy.DesiredPayloadHash = conflict.Legacy.AppliedPayloadHash
	conflict.Mapping.LegacyAuthorityDigest, err = LegacyAuthorityDigest(conflict.Legacy)
	if err != nil {
		t.Fatal(err)
	}
	conflict.Mapping.SignatureVerificationDigest = testDigest("other-mapping-signature")
	conflict.Mapping.MappingDigest, err = LegacyMappingDigest(conflict.Mapping)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := authority.ConvertLegacyPack(context.Background(), conflict); !errors.Is(err, ErrConflict) {
		t.Fatalf("changed legacy incarnation = %v, want conflict", err)
	}

	// A failure after the migration row is inserted but before the operation ID
	// is minted must roll back the complete Catalog transaction.
	var calls atomic.Int32
	failingAuthority, err := New(db, Options{NewID: func(now time.Time) (string, error) {
		if calls.Add(1) == 1 {
			return newUUIDv7(now)
		}
		return "", errors.New("injected operation-id failure")
	}})
	if err != nil {
		t.Fatal(err)
	}
	failingRequest := testConversionRequest(t, release, "org-rollback", "legacy-rollback")
	if _, err := failingAuthority.ConvertLegacyPack(context.Background(), failingRequest); err == nil {
		t.Fatal("fault-injected conversion succeeded")
	}
	var count int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogCRMConversionJournal WHERE OrgId = ?`, "org-rollback").Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("failed conversion left %d journal rows", count)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_CatalogAppInstallation WHERE OrgId = ?`, "org-rollback").Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("failed conversion left %d partial installations", count)
	}

	// The per-org row lock and CAS permit exactly one concurrent new operation.
	release2 := testRegisteredReleaseVersionForApp(t, "io.example.conversion", "2.0.0", "Summary")
	registerOneRelease(t, authority, release2)
	requests := []BeginOperationRequest{
		testOperationRequest(release2, "org-conversion", "race-a", 1, 1),
		testOperationRequest(release2, "org-conversion", "race-b", 1, 1),
	}
	for i := range requests {
		requests[i].Kind, requests[i].AppInstallationID = OperationUpgrade, conversion.AppInstallationID
	}
	results := make(chan error, 2)
	var wait sync.WaitGroup
	for _, candidate := range requests {
		wait.Add(1)
		go func() {
			defer wait.Done()
			_, err := authority.BeginOperation(context.Background(), candidate)
			results <- err
		}()
	}
	wait.Wait()
	close(results)
	successes, conflicts := 0, 0
	for err := range results {
		switch {
		case err == nil:
			successes++
		case errors.Is(err, ErrConflict):
			conflicts++
		default:
			t.Fatalf("concurrent BeginOperation = %v", err)
		}
	}
	if successes != 1 || conflicts != 1 {
		t.Fatalf("concurrent CAS: success=%d conflict=%d", successes, conflicts)
	}
}

func testCatalogAppTiDBFrontendOnlyBindingCAS(t *testing.T, db *sql.DB, authority *Authority) {
	release := testRegisteredReleaseWithFrontendForApp(t, "io.example.frontend", "1.0.0")
	release2 := testRegisteredReleaseWithFrontendForApp(t, "io.example.frontend", "2.0.0")
	registerTestReleases(t, authority, release, release2)
	prepareRequest := testOperationRequest(release, "org-frontend", "prepare-frontend", 0, 0)
	prepare, err := authority.BeginOperation(context.Background(), prepareRequest)
	if err != nil {
		t.Fatal(err)
	}
	stageTestBinding(t, authority, prepare, release, 0)
	initialRequest := FrontendBindingRequest{OperationID: prepare.OperationID,
		FrontendReleaseDigest: testDigest("frontend-1"), CompatibilityReceiptDigest: testDigest("frontend-compatibility-1")}
	failedBinding, err := authority.StageFrontendBinding(context.Background(), initialRequest)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.FailStagedFrontendBinding(context.Background(), failedBinding.FrontendBindingID); err != nil {
		t.Fatal(err)
	}
	if err := authority.FailOperation(context.Background(), prepare.OperationID, testDigest("frontend-validation-failure")); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.ResumeOperation(context.Background(), prepare.OperationID); err != nil {
		t.Fatal(err)
	}
	initial, err := authority.StageFrontendBinding(context.Background(), initialRequest)
	if err != nil {
		t.Fatal(err)
	}
	if initial.FrontendBindingID == failedBinding.FrontendBindingID || initial.State != BindingStaged {
		t.Fatalf("replacement frontend binding = %#v, failed binding = %#v", initial, failedBinding)
	}
	if retry, err := authority.StageFrontendBinding(context.Background(), initialRequest); err != nil || retry.FrontendBindingID != initial.FrontendBindingID {
		t.Fatalf("replacement frontend binding retry = %#v, %v", retry, err)
	}
	if _, err := authority.CommitServingArtifacts(context.Background(), prepare.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(context.Background(), prepare.OperationID, newMemoryRuntimeProjectionWriter()); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(context.Background(), prepare.OperationID, nil); err != nil {
		t.Fatal(err)
	}
	request := FrontendBindingRequest{AppInstallationID: prepare.AppInstallationID, AppReleaseDigest: release.AppReleaseDigest,
		FrontendReleaseDigest: testDigest("frontend-2"), CompatibilityReceiptDigest: testDigest("frontend-compatibility-2")}
	failedCandidate, err := authority.StageFrontendBinding(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.FailStagedFrontendBinding(context.Background(), failedCandidate.FrontendBindingID); err != nil {
		t.Fatal(err)
	}
	candidate, err := authority.StageFrontendBinding(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if candidate.FrontendBindingID == failedCandidate.FrontendBindingID {
		t.Fatalf("failed standalone frontend attempt was reused: %#v", candidate)
	}
	retry, err := authority.StageFrontendBinding(context.Background(), request)
	if err != nil || retry.FrontendBindingID != candidate.FrontendBindingID {
		t.Fatalf("frontend-only binding retry = %#v, %v", retry, err)
	}
	upgradeRequest := testOperationRequest(release2, "org-frontend", "upgrade-frontend", 1, 1)
	upgradeRequest.Kind = OperationUpgrade
	upgradeRequest.AppInstallationID = prepare.AppInstallationID
	if _, err := authority.BeginOperation(context.Background(), upgradeRequest); !errors.Is(err, ErrNotReady) {
		t.Fatalf("App upgrade with standalone frontend candidate = %v, want not ready", err)
	}
	if _, err := authority.ActivateStagedFrontendBinding(context.Background(), candidate.FrontendBindingID, "stale-binding"); !errors.Is(err, ErrConflict) {
		t.Fatalf("stale frontend activation = %v, want conflict", err)
	}
	activated, err := authority.ActivateStagedFrontendBinding(context.Background(), candidate.FrontendBindingID, initial.FrontendBindingID)
	if err != nil || activated.State != BindingActive {
		t.Fatalf("frontend-only activation = %#v, %v", activated, err)
	}
	if retry, err := authority.ActivateStagedFrontendBinding(context.Background(), candidate.FrontendBindingID, initial.FrontendBindingID); err != nil || retry.State != BindingActive {
		t.Fatalf("frontend-only activation retry = %#v, %v", retry, err)
	}
	if _, err := authority.ActivateStagedFrontendBinding(context.Background(), candidate.FrontendBindingID, "wrong-predecessor"); !errors.Is(err, ErrConflict) {
		t.Fatalf("conflicting frontend activation retry = %v, want conflict", err)
	}
	old, err := authority.ReadFrontendBinding(context.Background(), initial.FrontendBindingID)
	if err != nil || old.State != BindingDraining {
		t.Fatalf("prior frontend binding = %#v, %v", old, err)
	}

	// Rolling the frontend back to the exact original artifact and receipt is
	// a new lifecycle attempt with a new immutable audit identity.
	rollbackRequest := FrontendBindingRequest{AppInstallationID: prepare.AppInstallationID, AppReleaseDigest: release.AppReleaseDigest,
		FrontendReleaseDigest: initial.FrontendReleaseDigest, CompatibilityReceiptDigest: initial.CompatibilityReceiptDigest}
	rollbackCandidate, err := authority.StageFrontendBinding(context.Background(), rollbackRequest)
	if err != nil {
		t.Fatal(err)
	}
	if rollbackCandidate.FrontendBindingID == initial.FrontendBindingID {
		t.Fatalf("frontend rollback reused historical binding identity: %#v", rollbackCandidate)
	}
	if retry, err := authority.StageFrontendBinding(context.Background(), rollbackRequest); err != nil || retry.FrontendBindingID != rollbackCandidate.FrontendBindingID {
		t.Fatalf("frontend rollback staging retry = %#v, %v", retry, err)
	}
	if _, err := authority.ActivateStagedFrontendBinding(context.Background(), rollbackCandidate.FrontendBindingID, candidate.FrontendBindingID); err != nil {
		t.Fatalf("frontend rollback activation = %v", err)
	}

	capability := release.Package.Manifest.ProvidedCapabilities[0]
	handlerCandidate, err := authority.StageHandlerBinding(context.Background(), HandlerBindingRequest{
		AppInstallationID: prepare.AppInstallationID, AppReleaseDigest: release.AppReleaseDigest,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
		ServiceReleaseDigest: testDigest("service-standalone"), ExpectedOwnershipEpoch: 1,
		CompatibilityReceiptDigest: testDigest("service-standalone-compatibility"),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := authority.BeginOperation(context.Background(), upgradeRequest); !errors.Is(err, ErrNotReady) {
		t.Fatalf("App upgrade with standalone handler candidate = %v, want not ready", err)
	}
	if err := authority.FailStagedHandlerBinding(context.Background(), handlerCandidate.HandlerBindingID); err != nil {
		t.Fatal(err)
	}
	upgrade, err := authority.BeginOperation(context.Background(), upgradeRequest)
	if err != nil {
		t.Fatalf("App upgrade after resolving standalone candidates = %v", err)
	}
	if err := authority.FailOperation(context.Background(), upgrade.OperationID, testDigest("test-cleanup")); err != nil {
		t.Fatal(err)
	}
}

type ambiguousRuntimeWriter struct {
	delegate      RuntimeProjectionWriter
	ambiguousOnce bool
}

// memoryRuntimeProjectionWriter proves Catalog's publication and recovery
// protocol without importing one particular Runtime persistence owner into
// Catalog's package closure. The real persistence repository aliases the same
// contract-owned projection type and proves direct interface satisfaction in
// the coordinated NexusCRM consumer change.
type memoryRuntimeProjectionWriter struct {
	mu      sync.Mutex
	epochs  map[string]uint64
	digests map[string]string
}

func newMemoryRuntimeProjectionWriter() RuntimeProjectionWriter {
	return &memoryRuntimeProjectionWriter{epochs: map[string]uint64{}, digests: map[string]string{}}
}

func (w *memoryRuntimeProjectionWriter) ActivateProjection(_ context.Context, expected uint64, projection RuntimeSecurityProjection) error {
	digest, err := ProjectionDigest(projection)
	if err != nil {
		return err
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	current := w.epochs[projection.OrgID]
	if current == projection.SecurityEpoch {
		if w.digests[projection.OrgID] == digest {
			return nil
		}
		return ErrConflict
	}
	if current != expected {
		return ErrConflict
	}
	w.epochs[projection.OrgID] = projection.SecurityEpoch
	w.digests[projection.OrgID] = digest
	return nil
}

type failOperationAfterRuntimeWriter struct {
	delegate      RuntimeProjectionWriter
	authority     *Authority
	operationID   string
	failureDigest string
}

func (w *ambiguousRuntimeWriter) ActivateProjection(ctx context.Context, expected uint64, projection RuntimeSecurityProjection) error {
	if err := w.delegate.ActivateProjection(ctx, expected, projection); err != nil {
		return err
	}
	if w.ambiguousOnce {
		w.ambiguousOnce = false
		return errors.New("injected response loss after Runtime commit")
	}
	return nil
}

func (w *failOperationAfterRuntimeWriter) ActivateProjection(ctx context.Context, expected uint64, projection RuntimeSecurityProjection) error {
	if err := w.delegate.ActivateProjection(ctx, expected, projection); err != nil {
		return err
	}
	return w.authority.FailOperation(ctx, w.operationID, w.failureDigest)
}

func testCatalogAppLayoutBundleSlotRelease(t *testing.T, appID, semanticVersion string, definitionVersion uint64,
	targetBusinessKey string, previousTargetComponentID *string,
) (RegisteredRelease, layoutbundleslotv1.Declaration) {
	t.Helper()
	verified, declarations := testCatalogAppLayoutBundlePackage(t, appID, semanticVersion, definitionVersion,
		targetBusinessKey, previousTargetComponentID, true)
	if len(declarations) != 1 {
		t.Fatalf("verified layout-bundle slot declaration count = %d, want 1", len(declarations))
	}
	return registeredRelease(verified), declarations[0]
}

func testCatalogAppLayoutBundleOmissionRelease(t *testing.T, appID, semanticVersion string) RegisteredRelease {
	t.Helper()
	verified, declarations := testCatalogAppLayoutBundlePackage(t, appID, semanticVersion, 0, "Omitted", nil, false)
	if len(declarations) != 0 {
		t.Fatalf("verified omitting AppRelease declaration count = %d, want 0", len(declarations))
	}
	return registeredRelease(verified)
}

func testCatalogAppLayoutBundlePackage(t *testing.T, appID, semanticVersion string, definitionVersion uint64,
	targetBusinessKey string, previousTargetComponentID *string, includeSlot bool,
) (appauthority.VerifiedPackage, []layoutbundleslotv1.Declaration) {
	t.Helper()
	targetBehavior := []byte(`{"fields":["Name"]}`)
	targetDigest, err := appauthority.ComponentDigest("metadata.object.v1", 1, targetBehavior)
	if err != nil {
		t.Fatal(err)
	}
	target := appauthority.Component{
		ComponentKind: "metadata.object.v1", ComponentFormatVersion: 1,
		NamespaceID: appID, BusinessKey: targetBusinessKey, DefinitionVersion: "1",
		LocalDependencies: []string{}, ExternalComponentRefs: []appauthority.ExternalComponentRef{},
		OverridePolicy: appauthority.OverridePolicyReplace,
		RemovalPolicy:  appauthority.RemovalPolicyBlockIfReferencedOrOverridden,
		ContentDigest:  targetDigest,
	}
	target.ComponentID = appauthority.ComponentID(target.NamespaceID, target.ComponentKind, target.BusinessKey)
	type fixtureComponent struct {
		component appauthority.Component
		behavior  []byte
	}
	components := []fixtureComponent{{component: target, behavior: targetBehavior}}
	if includeSlot {
		behavior, err := layoutbundleslotv1.CanonicalBehaviorBytes(layoutbundleslotv1.Behavior{
			SchemaVersion: layoutbundleslotv1.BehaviorSchemaVersion, LocalComponentKey: testCatalogAppLayoutBundleSlotKey,
			TargetComponentID: target.ComponentID, PreviousTargetComponentID: previousTargetComponentID,
		})
		if err != nil {
			t.Fatal(err)
		}
		slotDigest, err := appauthority.ComponentDigest(layoutbundleslotv1.ComponentKind, layoutbundleslotv1.ComponentFormatVersion, behavior)
		if err != nil {
			t.Fatal(err)
		}
		slot := appauthority.Component{
			ComponentKind: layoutbundleslotv1.ComponentKind, ComponentFormatVersion: layoutbundleslotv1.ComponentFormatVersion,
			NamespaceID: appID, BusinessKey: string(testCatalogAppLayoutBundleSlotKey), DefinitionVersion: strconv.FormatUint(definitionVersion, 10),
			LocalDependencies: []string{target.ComponentID}, ExternalComponentRefs: []appauthority.ExternalComponentRef{},
			OverridePolicy: appauthority.OverridePolicyReplace,
			RemovalPolicy:  appauthority.RemovalPolicyBlockIfReferencedOrOverridden,
			ContentDigest:  slotDigest,
		}
		slot.ComponentID = appauthority.ComponentID(slot.NamespaceID, slot.ComponentKind, slot.BusinessKey)
		components = append(components, fixtureComponent{component: slot, behavior: behavior})
	}
	sort.Slice(components, func(left, right int) bool {
		return components[left].component.ComponentID < components[right].component.ComponentID
	})
	bundleComponents := make([]appauthority.Component, len(components))
	blobs := make([]appauthority.ComponentBlob, len(components))
	for index, component := range components {
		bundleComponents[index] = component.component
		blobs[index] = appauthority.ComponentBlob{ComponentID: component.component.ComponentID, CanonicalBehavior: component.behavior}
	}
	manifest := appauthority.AppManifest{
		SchemaVersion: appauthority.AppManifestSchemaVersion, AppID: appID, PublisherID: "io.example",
		SemanticVersion: semanticVersion, MetadataBundleDigest: testDigest("layout-bundle-slot-placeholder"),
		Dependencies: []appauthority.Dependency{}, OwnedNamespaces: []string{appID},
		RequiredProtocols: []appauthority.ProtocolRequirement{}, RequiredEngineCapabilities: []string{},
		ProvidedCapabilities:  []appauthority.ProvidedCapability{},
		FrontendCompatibility: appauthority.FrontendCompatibility{Mode: appauthority.FrontendModeNone},
	}
	bundle := appauthority.MetadataBundle{SchemaVersion: appauthority.MetadataBundleSchemaVersion, Components: bundleComponents}
	manifest.MetadataBundleDigest, err = appauthority.MetadataBundleDigest(bundle, manifest)
	if err != nil {
		t.Fatal(err)
	}
	verified, declarations, err := layoutbundleslotv1.VerifyPackageDeclarations(appauthority.Package{
		Manifest: manifest, Bundle: bundle, ComponentBlobs: blobs,
	})
	if err != nil {
		t.Fatal(err)
	}
	return verified, declarations
}

func testCatalogAppLayoutBundleTargetID(appID, targetBusinessKey string) string {
	return appauthority.ComponentID(appID, "metadata.object.v1", targetBusinessKey)
}

func assertCatalogAppLayoutBundleSlotBinding(t *testing.T, authority *Authority, release RegisteredRelease,
	declaration layoutbundleslotv1.Declaration,
) {
	t.Helper()
	bindings, err := authority.ReadReleaseLayoutBundleSlots(context.Background(), release.AppID, release.AppReleaseDigest)
	if err != nil {
		t.Fatal(err)
	}
	if len(bindings) != 1 {
		t.Fatalf("AppRelease %s layout-bundle slot bindings = %#v, want one", release.AppReleaseDigest, bindings)
	}
	binding := bindings[0]
	if binding.AppID != release.AppID || binding.AppReleaseDigest != release.AppReleaseDigest ||
		binding.LocalComponentKey != declaration.LocalComponentKey || binding.DeclarationComponentID != declaration.DeclarationComponentID ||
		binding.NamespaceID != declaration.NamespaceID || binding.DefinitionVersion != declaration.DefinitionVersion ||
		binding.RevisionDigest != declaration.RevisionDigest || binding.TargetComponentID != declaration.TargetComponentID ||
		!testCatalogAppEqualOptionalString(binding.PreviousTargetComponentID, declaration.PreviousTargetComponentID) {
		t.Fatalf("AppRelease layout-bundle slot binding = %#v, want declaration %#v", binding, declaration)
	}
}

func assertCatalogAppLayoutBundleSlotHead(t *testing.T, authority *Authority, appID string, definitionVersion uint64,
	revisionDigest, targetComponentID string,
) {
	t.Helper()
	lineage, err := authority.ReadLayoutBundleSlotLineage(context.Background(), appID, testCatalogAppLayoutBundleSlotKey)
	if err != nil {
		t.Fatal(err)
	}
	if err := layoutbundleslotv1.ValidateLineage(lineage); err != nil {
		t.Fatalf("stored layout-bundle slot lineage is invalid: %v", err)
	}
	if lineage.AppID != appID || lineage.LocalComponentKey != testCatalogAppLayoutBundleSlotKey || len(lineage.Revisions) != int(definitionVersion) {
		t.Fatalf("layout-bundle slot lineage = %#v, want app %s with %d revisions", lineage, appID, definitionVersion)
	}
	head := lineage.Revisions[len(lineage.Revisions)-1]
	if head.DefinitionVersion != definitionVersion || head.RevisionDigest != revisionDigest || head.TargetComponentID != targetComponentID {
		t.Fatalf("layout-bundle slot head = %#v, want version %d digest %s target %s", head, definitionVersion, revisionDigest, targetComponentID)
	}
}

func testCatalogAppEqualOptionalString(left, right *string) bool {
	if left == nil || right == nil {
		return left == right
	}
	return *left == *right
}

func catalogAppLayoutBundleSlotStateBytes(t *testing.T, db *sql.DB, appID string) []byte {
	t.Helper()
	var snapshot bytes.Buffer
	reservationRows, err := db.Query(`SELECT HEX(AppId), HEX(LocalComponentKey), HEX(DeclarationComponentId), HEX(NamespaceId),
HEX(CAST(HeadDefinitionVersion AS CHAR)), HEX(HeadRevisionDigest), HEX(CAST(CreatedDate AS CHAR)), HEX(CAST(LastModifiedDate AS CHAR))
FROM _System_CatalogAppLayoutBundleSlot WHERE AppId = ? ORDER BY LocalComponentKey`, appID)
	if err != nil {
		t.Fatal(err)
	}
	for reservationRows.Next() {
		var app, key, declaration, namespace, version, digest, created, modified string
		if err := reservationRows.Scan(&app, &key, &declaration, &namespace, &version, &digest, &created, &modified); err != nil {
			reservationRows.Close()
			t.Fatal(err)
		}
		fmt.Fprintf(&snapshot, "reservation|%s|%s|%s|%s|%s|%s|%s|%s\n", app, key, declaration, namespace, version, digest, created, modified)
	}
	if err := reservationRows.Err(); err != nil {
		reservationRows.Close()
		t.Fatal(err)
	}
	if err := reservationRows.Close(); err != nil {
		t.Fatal(err)
	}
	versionRows, err := db.Query(`SELECT HEX(AppId), HEX(LocalComponentKey), HEX(CAST(DefinitionVersion AS CHAR)), HEX(RevisionDigest),
HEX(TargetComponentId), HEX(PreviousTargetComponentId), HEX(CAST(CreatedDate AS CHAR))
FROM _System_CatalogAppLayoutBundleSlotVersion WHERE AppId = ? ORDER BY LocalComponentKey, DefinitionVersion`, appID)
	if err != nil {
		t.Fatal(err)
	}
	for versionRows.Next() {
		var app, key, version, digest, target, created string
		var previous sql.NullString
		if err := versionRows.Scan(&app, &key, &version, &digest, &target, &previous, &created); err != nil {
			versionRows.Close()
			t.Fatal(err)
		}
		fmt.Fprintf(&snapshot, "version|%s|%s|%s|%s|%s|%t:%s|%s\n", app, key, version, digest, target, previous.Valid, previous.String, created)
	}
	if err := versionRows.Err(); err != nil {
		versionRows.Close()
		t.Fatal(err)
	}
	if err := versionRows.Close(); err != nil {
		t.Fatal(err)
	}
	return snapshot.Bytes()
}

func assertCatalogAppReleaseRowsAbsent(t *testing.T, db *sql.DB, release RegisteredRelease) {
	t.Helper()
	for _, table := range []string{
		"_System_CatalogAppRelease",
		"_System_CatalogAppReleaseDependency",
		"_System_CatalogAppReleaseNamespace",
		"_System_CatalogAppReleaseCapability",
		"_System_CatalogAppReleaseComponent",
		"_System_CatalogAppReleaseLayoutBundleSlot",
	} {
		var count int
		query := fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE AppId = ? AND AppReleaseDigest = ?", table)
		if err := db.QueryRow(query, release.AppID, release.AppReleaseDigest).Scan(&count); err != nil {
			t.Fatal(err)
		}
		if count != 0 {
			t.Fatalf("losing AppRelease retained %d rows in %s", count, table)
		}
	}
}

func registerTestReleases(t *testing.T, authority *Authority, releases ...RegisteredRelease) {
	t.Helper()
	now := time.Now().UTC()
	if err := authority.ReserveAppID(context.Background(), releases[0].PublisherID, releases[0].AppID, 1); err != nil {
		t.Fatal(err)
	}
	for _, release := range releases {
		registerOneReleaseAt(t, authority, release, now)
	}
}

func registerTestPublisher(t *testing.T, authority *Authority) {
	t.Helper()
	if _, err := authority.RegisterPublisher(context.Background(), PublisherTrust{PublisherID: "io.example", Active: true,
		Keys: []PublisherKey{testReleaseSigningKey()}}, 0); err != nil {
		t.Fatal(err)
	}
}

func registerOneRelease(t *testing.T, authority *Authority, release RegisteredRelease) {
	t.Helper()
	registerOneReleaseAt(t, authority, release, time.Now().UTC())
}

func registerOneReleaseAt(t *testing.T, authority *Authority, release RegisteredRelease, now time.Time) {
	t.Helper()
	_, err := authority.registerVerifiedRelease(context.Background(), testVerifiedReleaseRegistration(t, release, now))
	if err != nil {
		t.Fatal(err)
	}
}

func testOperationRequest(release RegisteredRelease, orgID, idempotency string, catalogRevision, securityEpoch uint64) BeginOperationRequest {
	return BeginOperationRequest{Kind: OperationPrepare, OrgID: orgID, AppID: release.AppID,
		TargetAppReleaseDigest: release.AppReleaseDigest, IdempotencyKey: idempotency, RuntimeAudience: "runtime",
		AuthorizedWorkloads: []string{"workload-1"}, GrantedCapabilities: release.Package.Manifest.RequiredEngineCapabilities,
		HandlerWorkloadID:    "workload-1",
		ResolvedDependencies: []ResolvedDependency{}, ExpectedCatalogRevision: catalogRevision, ExpectedSecurityEpoch: securityEpoch}
}

func stageTestBinding(t *testing.T, authority *Authority, operation Operation, release RegisteredRelease, ownershipEpoch uint64) {
	t.Helper()
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	_, err := authority.StageHandlerBinding(context.Background(), HandlerBindingRequest{OperationID: operation.OperationID,
		CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef, WorkloadID: "workload-1",
		ServiceReleaseDigest: testDigest("service-" + release.SemanticVersion), ExpectedOwnershipEpoch: ownershipEpoch,
		CompatibilityReceiptDigest: testDigest("compatibility-" + release.SemanticVersion)})
	if err != nil {
		t.Fatal(err)
	}
}

func testRegisteredReleaseWithFrontendForApp(t *testing.T, appID, version string) RegisteredRelease {
	t.Helper()
	base := testRegisteredReleaseVersionForApp(t, appID, version, "Subject")
	manifest := base.Package.Manifest
	manifest.FrontendCompatibility = appauthority.FrontendCompatibility{Mode: appauthority.FrontendModeRequired,
		HostSDK:            &appauthority.ProtocolRequirement{ProtocolID: "nexuscrm.frontend.host-sdk.v1", Major: 1},
		ContributionSchema: &appauthority.ProtocolRequirement{ProtocolID: "nexuscrm.frontend.contribution.v1", Major: 1}}
	metadataDigest, err := appauthority.MetadataBundleDigest(base.Package.Bundle, manifest)
	if err != nil {
		t.Fatal(err)
	}
	manifest.MetadataBundleDigest = metadataDigest
	verified, err := appauthority.VerifyPackage(appauthority.Package{Manifest: manifest, Bundle: base.Package.Bundle,
		ComponentBlobs: base.Package.ComponentBlobs})
	if err != nil {
		t.Fatal(err)
	}
	return registeredRelease(verified)
}

func testRegisteredReleaseForApp(t *testing.T, appID, version string) RegisteredRelease {
	t.Helper()
	base := testRegisteredReleaseVersion(t, version, "Subject")
	manifest := base.Package.Manifest
	manifest.AppID = appID
	metadataDigest, err := appauthority.MetadataBundleDigest(base.Package.Bundle, manifest)
	if err != nil {
		t.Fatal(err)
	}
	manifest.MetadataBundleDigest = metadataDigest
	verified, err := appauthority.VerifyPackage(appauthority.Package{Manifest: manifest, Bundle: base.Package.Bundle,
		ComponentBlobs: base.Package.ComponentBlobs})
	if err != nil {
		t.Fatal(err)
	}
	return registeredRelease(verified)
}

func testConversionRequest(t *testing.T, release RegisteredRelease, orgID, packInstallID string) ConversionRequest {
	t.Helper()
	component := release.Package.Bundle.Components[0]
	legacy := testLegacyInstalledPack(orgID, packInstallID, component)
	legacyAuthorityDigest, err := LegacyAuthorityDigest(legacy)
	if err != nil {
		t.Fatal(err)
	}
	mapping := VerifiedLegacyMapping{LegacyAuthorityDigest: legacyAuthorityDigest, SignerID: "operator.example",
		SignatureVerificationDigest: testDigest("mapping-signature"), AppID: release.AppID,
		TargetReleaseDigest: release.AppReleaseDigest, VerifiedAt: time.Unix(1, 0).UTC(),
		Components: []LegacyComponentMapping{{LegacyComponentKind: "legacy.object.v1", LegacyBusinessKey: component.BusinessKey, TargetComponentID: component.ComponentID}}}
	mapping.MappingDigest, err = LegacyMappingDigest(mapping)
	if err != nil {
		t.Fatal(err)
	}
	capability := release.Package.Manifest.ProvidedCapabilities[0]
	return ConversionRequest{Legacy: legacy,
		Mapping: mapping, Operation: testOperationRequest(release, orgID, "convert-"+packInstallID, 0, 0),
		HandlerBindings: []HandlerBindingRequest{{CapabilityID: capability.CapabilityID, HandlerRef: capability.HandlerRef,
			WorkloadID: "workload-1", ServiceReleaseDigest: testDigest("service-conversion"),
			CompatibilityReceiptDigest: testDigest("compatibility-conversion")}}}
}

func openCatalogTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := getenv("TIDB_HOST", "127.0.0.1")
	port := getenv("TIDB_PORT", "4000")
	cfg := &mysql.Config{User: getenv("TIDB_USER", "root"), Passwd: os.Getenv("TIDB_PASSWORD"), Net: "tcp",
		Addr: host + ":" + port, Collation: "utf8mb4_unicode_ci", ParseTime: true, Loc: time.UTC, AllowNativePasswords: true}
	if host != "127.0.0.1" && host != "localhost" {
		tlsName := fmt.Sprintf("catalogapp_%d", time.Now().UnixNano())
		if err := mysql.RegisterTLSConfig(tlsName, &tls.Config{MinVersion: tls.VersionTLS12, ServerName: host}); err != nil {
			t.Fatal(err)
		}
		cfg.TLSConfig = tlsName
	}
	admin, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	if err := admin.Ping(); err != nil {
		admin.Close()
		if os.Getenv("NEXUSCRM_REQUIRE_CATALOGAPP_TIDB") == "1" {
			t.Fatalf("required TiDB is unavailable at %s:%s: %v", host, port, err)
		}
		t.Skipf("TiDB is unavailable at %s:%s: %v", host, port, err)
	}
	runID := os.Getenv("NEXUSCRM_CATALOGAPP_RUN_ID")
	if runID != "" && (!regexp.MustCompile(`^[A-Za-z0-9_]{1,40}$`).MatchString(runID)) {
		t.Fatalf("NEXUSCRM_CATALOGAPP_RUN_ID must contain only 1-40 ASCII letters, digits, or underscores")
	}
	prefix := "nexuscrm_catalogapp"
	if runID != "" {
		prefix += "_" + runID
	}
	database := fmt.Sprintf("%s_%d", prefix, time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + database + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); err != nil {
		admin.Close()
		t.Fatal(err)
	}
	testConfig := *cfg
	testConfig.DBName = database
	db, err := sql.Open("mysql", testConfig.FormatDSN())
	if err != nil {
		admin.Close()
		t.Fatal(err)
	}
	cleanup := func() {
		db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + database + "`")
		admin.Close()
	}
	return db, cleanup
}

func getenv(name, fallback string) string {
	if value := os.Getenv(name); value != "" {
		return value
	}
	return fallback
}