//go:build integration

package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"testing"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/pingcap-inc/nexuscrm-controlplane/bootstrapimpl"
)

// connectorServingAuthorityTableNames is ordered parent-first for catalog checks.
var connectorServingAuthorityTableNames = []string{
	"_System_ConnectorServingEpoch",
	"_System_ConnectorServingEpochBinding",
	"_System_ConnectorAccountServingGate",
	"_System_ConnectorServingEpochReproducibility",
	"_System_ConnectorServingEpochBindingOwner",
	"_System_ConnectorServingEpochCredentialGeneration",
	"_System_ConnectorV2Receipt",
	"_System_ConnectorV2ReceiptComponent",
	"_System_ConnectorV2ReceiptAdmissionKey",
}

var connectorServingEpochReproducibilityTableNames = []string{
	"_System_ConnectorServingEpochReproducibility",
	"_System_ConnectorServingEpochBindingOwner",
	"_System_ConnectorServingEpochCredentialGeneration",
}

var connectorV2IngressReceiptAuthorityTableNames = []string{
	"_System_ConnectorV2Receipt",
	"_System_ConnectorV2ReceiptComponent",
	"_System_ConnectorV2ReceiptAdmissionKey",
}

// These current-schema tables depend on serving-authority parents. Tests that
// recreate an older serving-authority schema must remove the children first.
var connectorServingAuthorityDependentTableNames = []string{
	"_System_ConnectorV2ExecutionAttempt",
	"_System_ConnectorV2Execution",
}

func TestConnectorServingAuthoritySchemaLifecycle(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	const sentinelID = "serving-authority-upgrade-sentinel"
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorIngressControl
		(Id, Namespace, PackInstallId, Channel, Mode, RetryAfterSeconds)
		VALUES (?, ?, '', '*', 'paused', 120)`, sentinelID, "serving-authority-upgrade"); err != nil {
		t.Fatalf("insert unrelated pre-upgrade control row: %v", err)
	}

	removeConnectorServingAuthorityPriorState(t, ctx, db)
	assertConnectorServingAuthorityAbsent(t, db)
	assertConnectorServingAuthoritySentinel(t, db, sentinelID)

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("converge Connector serving-authority schema from prior state: %v", err)
	}
	assertConnectorServingAuthorityReady(t, db)
	assertConnectorServingAuthoritySentinel(t, db, sentinelID)

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("idempotent Connector serving-authority schema reapply: %v", err)
	}
	assertConnectorServingAuthorityReady(t, db)
	assertConnectorServingAuthoritySentinel(t, db, sentinelID)
}

func TestConnectorServingEpochReproducibilitySchemaStoppedUpgrade(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	const (
		epochID   = "legacy-serving-epoch-before-reproducibility"
		bindingID = "legacy-serving-epoch-binding-before-reproducibility"
	)
	digest := func(value string) string { return "sha256:" + strings.Repeat(value, 64) }
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorServingEpoch (
		Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
		OrgId, AppInstallationId, AuthorizedWorkloadId,
		CatalogRevision, SchemaGeneration, EngineGeneration,
		SourceSnapshotDigest, BindingViewDigest,
		ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
		L2ConfigurationGenerationRef, L2ConfigurationDigest,
		L2VerificationReceipt, L2VerificationReceiptDigest
	) VALUES (?, ?, ?, NULL, 'prepared', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		epochID,
		digest("1"),
		strings.Repeat("2", 64),
		"legacy-org",
		"legacy-app-installation",
		"legacy-authorized-workload",
		uint64(11),
		uint64(12),
		uint64(13),
		digest("3"),
		digest("4"),
		"1.0",
		"catalog.consumer.integration-binding-view.v1",
		uint64(14),
		"legacy-l2-generation",
		digest("5"),
		[]byte("legacy-l2-receipt\x00with-exact-bytes"),
		digest("6"),
	); err != nil {
		t.Fatalf("insert pre-extension legacy ServingEpoch: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorServingEpochBinding (
		Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
		Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
		BusinessKey, DefinitionVersion, CredentialRefs,
		RuntimeCommandKind, RuntimePayloadFormatVersion,
		RuntimeDefinitionRef, RuntimeDefinitionVersion
	) VALUES (?, ?, 0, ?, ?, ?, ?, ?, ?, 7, ?, ?, 3, ?, ?)`,
		bindingID,
		epochID,
		strings.Repeat("7", 64),
		"CONNECTOR_INBOUND_ROUTE",
		"legacy-owner-pack",
		"legacy-pack-installation",
		"nexus-pack-sha256-v1:"+strings.Repeat("8", 64),
		"legacy-business-key",
		[]byte("credential-b\x00credential-a"),
		"integration.command.execute",
		"legacy-runtime-definition",
		"legacy-runtime-definition-version",
	); err != nil {
		t.Fatalf("insert pre-extension legacy ServingEpoch binding: %v", err)
	}

	wantEpoch, wantBinding := connectorServingAuthorityLegacyFingerprints(t, db, epochID, bindingID)
	removeConnectorServingEpochReproducibilityPriorState(t, ctx, db)
	assertConnectorServingAuthorityTablesAbsent(t, db, connectorServingEpochReproducibilityTableNames)
	assertConnectorServingAuthorityLegacyFingerprints(t, db, epochID, bindingID, wantEpoch, wantBinding)

	readinessErr := bootstrapimpl.VerifySchemaReadiness(conn)
	if readinessErr == nil {
		t.Fatal("VerifySchemaReadiness without ServingEpoch reproducibility tables = nil, want fail closed")
	}
	for _, tableName := range connectorServingEpochReproducibilityTableNames {
		if !strings.Contains(readinessErr.Error(), tableName) {
			t.Fatalf("VerifySchemaReadiness without reproducibility tables = %v, want %s", readinessErr, tableName)
		}
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("stopped upgrade of ServingEpoch reproducibility schema: %v", err)
	}
	assertConnectorServingAuthorityTablesReady(t, db, connectorServingEpochReproducibilityTableNames)
	assertConnectorServingEpochReproducibilityProofRowsEmpty(t, db)
	assertConnectorServingAuthorityLegacyFingerprints(t, db, epochID, bindingID, wantEpoch, wantBinding)
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("VerifySchemaReadiness after stopped reproducibility upgrade: %v", err)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("idempotent ServingEpoch reproducibility schema reapply: %v", err)
	}
	assertConnectorServingAuthorityTablesReady(t, db, connectorServingEpochReproducibilityTableNames)
	assertConnectorServingEpochReproducibilityProofRowsEmpty(t, db)
	assertConnectorServingAuthorityLegacyFingerprints(t, db, epochID, bindingID, wantEpoch, wantBinding)
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("VerifySchemaReadiness after idempotent reproducibility reapply: %v", err)
	}
}

func TestConnectorV2IngressReceiptSchemaStoppedUpgrade(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	const (
		gateID       = "legacy-serving-gate-before-v2-receipts"
		accountID    = "legacy-connector-account"
		v1ReceiptID  = "legacy-v1-ingress-receipt"
		v1ReceiptKey = "legacy-v1-transport-key"
	)
	servingScopeDigest := strings.Repeat("a", 64)
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorAccountServingGate (
		Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
		ServingEpochId, ServingEpochDigest
	) VALUES (?, ?, ?, 'v1_serving', 19, NULL, NULL)`, gateID, servingScopeDigest, accountID); err != nil {
		t.Fatalf("insert pre-upgrade account serving gate: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorIngressReceipt (
		Id, Namespace, InstalledPackId, PackVersion,
		SecretName, SecretAlgorithm, VerifierCapabilityKind, ParserCapabilityKind,
		RequestMethod, RequestTargetHash, RequestTargetCiphertext,
		ReceiptKey, BodyHash, BodySizeBytes, BodyCiphertext,
		AllowedHeadersCiphertext, Status, AttemptCount
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'enqueued', 3)`,
		v1ReceiptID,
		"legacy-v1-namespace",
		"legacy-v1-pack-installation",
		"1.0.0",
		"legacy-v1-secret",
		"hmac-sha256",
		"integration.webhook.verify",
		"integration.webhook.parse",
		"POST",
		strings.Repeat("b", 64),
		"ciphertext:target:\x00exact",
		v1ReceiptKey,
		strings.Repeat("c", 64),
		uint64(17),
		"ciphertext:body:\x00exact",
		"ciphertext:headers:\x00exact",
	); err != nil {
		t.Fatalf("insert pre-upgrade V1 ingress receipt: %v", err)
	}

	wantGate, wantV1Receipt := connectorV2IngressLegacyFingerprints(t, db, gateID, v1ReceiptID)
	dropConnectorV2IngressReceiptPhysicalTables(t, ctx, db)
	assertConnectorV2IngressReceiptPhysicalTablesAbsentWithMetadataRetained(t, db)
	assertConnectorV2IngressLegacyFingerprints(t, db, gateID, v1ReceiptID, wantGate, wantV1Receipt)

	// Characterize the real resume hazard: a late bootstrap failure skips
	// phase0 column reconciliation and phase1 batch table creation. Managed
	// metadata from the completed earlier phases remains present, while the two
	// mandatory physical tables are absent. The resume-safe reconciliation must
	// recreate them before the later phase is allowed to complete.
	result, err := db.ExecContext(ctx, `UPDATE _System_BootstrapState
		SET Phase = 'phase6_field_labels', Status = 'failed', LastError = 'V2 ingress receipt resume proof'
		WHERE StateKey = 'schema_initialize'`)
	if err != nil {
		t.Fatalf("prepare late failed bootstrap resume state: %v", err)
	}
	if affected, err := result.RowsAffected(); err != nil || affected != 1 {
		t.Fatalf("prepare bootstrap resume state affected = %d, err = %v, want 1", affected, err)
	}

	readinessErr := bootstrapimpl.VerifySchemaReadiness(conn)
	if readinessErr == nil {
		t.Fatal("VerifySchemaReadiness without V2 ingress receipt authority tables = nil, want fail closed")
	}
	for _, tableName := range connectorV2IngressReceiptAuthorityTableNames {
		if !strings.Contains(readinessErr.Error(), tableName) {
			t.Fatalf("VerifySchemaReadiness without V2 ingress receipt authority tables = %v, want %s", readinessErr, tableName)
		}
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("stopped upgrade of V2 ingress receipt authority schema: %v", err)
	}
	assertConnectorServingAuthorityTablesReady(t, db, connectorV2IngressReceiptAuthorityTableNames)
	assertConnectorV2IngressReceiptRowsEmpty(t, db)
	assertConnectorV2IngressLegacyFingerprints(t, db, gateID, v1ReceiptID, wantGate, wantV1Receipt)
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("VerifySchemaReadiness after stopped V2 ingress receipt upgrade: %v", err)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("idempotent V2 ingress receipt authority schema reapply: %v", err)
	}
	assertConnectorServingAuthorityTablesReady(t, db, connectorV2IngressReceiptAuthorityTableNames)
	assertConnectorV2IngressReceiptRowsEmpty(t, db)
	assertConnectorV2IngressLegacyFingerprints(t, db, gateID, v1ReceiptID, wantGate, wantV1Receipt)
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("VerifySchemaReadiness after idempotent V2 ingress receipt reapply: %v", err)
	}
}

func TestConnectorV2IngressReceiptProtocolWidthStoppedUpgrade(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	installLegacyConnectorV2ReceiptProtocolWidth(t, ctx, db)
	const protocol = "nexuscrm.connector-v2-receipt.v1"
	seedConnectorV2ReceiptProtocolMigrationGraph(t, ctx, db, protocol)
	markConnectorV2ReceiptMigrationAsLateFailedBootstrap(t, ctx, db)
	assertConnectorV2ReceiptProtocolColumnShape(t, db, "varchar(255)")

	readinessErr := bootstrapimpl.VerifySchemaReadiness(conn)
	if readinessErr == nil ||
		!strings.Contains(readinessErr.Error(), "ReceiptProtocolVersion") ||
		!strings.Contains(readinessErr.Error(), "cmd/setup apply") {
		t.Fatalf("legacy receipt protocol width readiness error = %v, want exact column and apply guidance", readinessErr)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("narrow retained Connector V2 receipt protocol width: %v", err)
	}
	assertConnectorV2ReceiptProtocolColumnShape(t, db, "varchar(32)")
	assertConnectorV2ReceiptProtocolMigrationRow(t, db, protocol)
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("VerifySchemaReadiness after receipt protocol-width migration: %v", err)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("idempotent Connector V2 receipt protocol-width reapply: %v", err)
	}
	assertConnectorV2ReceiptProtocolColumnShape(t, db, "varchar(32)")
	assertConnectorV2ReceiptProtocolMigrationRow(t, db, protocol)
}

func TestConnectorV2IngressReceiptProtocolWidthRejectsOversizeRetainedValue(t *testing.T) {
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	db := conn.DB()

	installLegacyConnectorV2ReceiptProtocolWidth(t, ctx, db)
	oversizeProtocol := strings.Repeat("x", 33)
	seedConnectorV2ReceiptProtocolMigrationGraph(t, ctx, db, oversizeProtocol)
	markConnectorV2ReceiptMigrationAsLateFailedBootstrap(t, ctx, db)

	err := schemaProvisioner.InitializeSchema(conn)
	if err == nil ||
		!strings.Contains(err.Error(), "ReceiptProtocolVersion exceeds the canonical VARCHAR(32) bound") {
		t.Fatalf("oversize retained receipt protocol migration error = %v, want exact bound rejection", err)
	}
	assertConnectorV2ReceiptMigrationBootstrapFailed(t, db, "ReceiptProtocolVersion exceeds the canonical VARCHAR(32) bound")
	assertConnectorV2ReceiptProtocolColumnShape(t, db, "varchar(255)")
	assertConnectorV2ReceiptProtocolMigrationRow(t, db, oversizeProtocol)
	if readinessErr := bootstrapimpl.VerifySchemaReadiness(conn); readinessErr == nil {
		t.Fatal("VerifySchemaReadiness after rejected receipt protocol migration = nil, want fail closed")
	}
}

func TestConnectorV2IngressReceiptTransportIdentityIndexStoppedUpgrade(t *testing.T) {
	t.Run("migrates retained authority", func(t *testing.T) {
		ctx := context.Background()
		cfg, cleanup := newTemporaryIntegrationDatabase(t)
		defer cleanup()

		conn := initializeControlPlaneMetadata(t, ctx, cfg)
		defer conn.Close()
		db := conn.DB()

		installLegacyConnectorV2ReceiptTransportIndex(t, ctx, db)
		seedConnectorV2ReceiptMigrationGate(t, ctx, db, "receipt-index-gate-a", strings.Repeat("a", 64), "receipt-index-account")
		seedConnectorV2ReceiptMigrationGate(t, ctx, db, "receipt-index-gate-b", strings.Repeat("b", 64), "receipt-index-account")
		seedConnectorV2ReceiptMigrationAdmissionKey(
			t,
			ctx,
			db,
			"retained-receipt-index-key",
			strings.Repeat("a", 64),
			"receipt-index-account",
			"receipt-index-gate-a",
			"retained-transport-key",
		)
		markConnectorV2ReceiptMigrationAsLateFailedBootstrap(t, ctx, db)

		assertConnectorV2ReceiptTransportIndexShape(t, db, []string{
			"ServingScopeDigest",
			"ConnectorAccountId",
			"IngressChannel",
			"TransportIdempotencyKey",
		})
		assertConnectorV2ReceiptCutoverIndexAbsent(t, db)
		readinessErr := bootstrapimpl.VerifySchemaReadiness(conn)
		if readinessErr == nil ||
			!strings.Contains(readinessErr.Error(), "ux_connector_v2_receipt_key_transport") ||
			!strings.Contains(readinessErr.Error(), "cmd/setup apply") {
			t.Fatalf("legacy transport index readiness error = %v, want exact index and apply guidance", readinessErr)
		}

		if err := schemaProvisioner.InitializeSchema(conn); err != nil {
			t.Fatalf("migrate retained Connector V2 transport identity index: %v", err)
		}
		assertConnectorV2ReceiptTransportIndexShape(t, db, []string{
			"ConnectorAccountId",
			"IngressChannel",
			"TransportIdempotencyKey",
		})
		assertConnectorV2ReceiptCutoverIndexAbsent(t, db)
		assertConnectorV2ReceiptMigrationAdmissionKey(
			t,
			db,
			"retained-receipt-index-key",
			strings.Repeat("a", 64),
			"receipt-index-account",
			"receipt-index-gate-a",
			"retained-transport-key",
		)
		if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
			t.Fatalf("VerifySchemaReadiness after transport index migration: %v", err)
		}

		_, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorV2ReceiptAdmissionKey (
			Id, AdmissionKeyProtocolVersion, ServingScopeDigest, ConnectorAccountId,
			AccountServingGateId, IngressChannel, TransportIdempotencyKey,
			AuthorizedPayloadDigest, Disposition, ReceiptId,
			PredecessorRefProtocolVersion, PredecessorRef, PredecessorRefDigest
		) VALUES (?, 'nexuscrm.connector-receipt-admission-key.v1', ?, ?, ?, 'webhook', ?, ?,
			'v1_tombstone', NULL, 'nexuscrm.connector-v1-predecessor-ref.v1', ?, ?)`,
			"cross-scope-duplicate-receipt-index-key",
			strings.Repeat("b", 64),
			"receipt-index-account",
			"receipt-index-gate-b",
			"retained-transport-key",
			"sha256:"+strings.Repeat("d", 64),
			[]byte("cross-scope duplicate must be rejected"),
			"sha256:"+strings.Repeat("e", 64),
		)
		var mysqlErr *mysql.MySQLError
		if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 {
			t.Fatalf("cross-scope duplicate insert error = %v, want MySQL 1062", err)
		}

		if err := schemaProvisioner.InitializeSchema(conn); err != nil {
			t.Fatalf("idempotent Connector V2 transport index reapply: %v", err)
		}
		assertConnectorV2ReceiptTransportIndexShape(t, db, []string{
			"ConnectorAccountId",
			"IngressChannel",
			"TransportIdempotencyKey",
		})
		assertConnectorV2ReceiptCutoverIndexAbsent(t, db)
	})

	t.Run("rejects retained duplicate authority without partial cutover", func(t *testing.T) {
		ctx := context.Background()
		cfg, cleanup := newTemporaryIntegrationDatabase(t)
		defer cleanup()

		conn := initializeControlPlaneMetadata(t, ctx, cfg)
		defer conn.Close()
		db := conn.DB()

		installLegacyConnectorV2ReceiptTransportIndex(t, ctx, db)
		seedConnectorV2ReceiptMigrationGate(t, ctx, db, "duplicate-index-gate-a", strings.Repeat("c", 64), "duplicate-index-account")
		seedConnectorV2ReceiptMigrationGate(t, ctx, db, "duplicate-index-gate-b", strings.Repeat("d", 64), "duplicate-index-account")
		seedConnectorV2ReceiptMigrationAdmissionKey(
			t,
			ctx,
			db,
			"duplicate-receipt-index-key-a",
			strings.Repeat("c", 64),
			"duplicate-index-account",
			"duplicate-index-gate-a",
			"duplicate-transport-key",
		)
		seedConnectorV2ReceiptMigrationAdmissionKey(
			t,
			ctx,
			db,
			"duplicate-receipt-index-key-b",
			strings.Repeat("d", 64),
			"duplicate-index-account",
			"duplicate-index-gate-b",
			"duplicate-transport-key",
		)
		markConnectorV2ReceiptMigrationAsLateFailedBootstrap(t, ctx, db)

		err := schemaProvisioner.InitializeSchema(conn)
		if err == nil ||
			!strings.Contains(err.Error(), "retained rows conflict") ||
			!strings.Contains(err.Error(), "ConnectorAccountId, IngressChannel, TransportIdempotencyKey") {
			t.Fatalf("retained duplicate migration error = %v, want exact authority conflict", err)
		}
		assertConnectorV2ReceiptTransportIndexShape(t, db, []string{
			"ServingScopeDigest",
			"ConnectorAccountId",
			"IngressChannel",
			"TransportIdempotencyKey",
		})
		assertConnectorV2ReceiptCutoverIndexAbsent(t, db)
		var rowCount int
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorV2ReceiptAdmissionKey
			WHERE ConnectorAccountId = 'duplicate-index-account'
			  AND IngressChannel = 'webhook'
			  AND TransportIdempotencyKey = 'duplicate-transport-key'`).Scan(&rowCount); err != nil {
			t.Fatalf("count retained duplicate authority rows: %v", err)
		}
		if rowCount != 2 {
			t.Fatalf("retained duplicate authority rows = %d, want unchanged 2", rowCount)
		}
		if readinessErr := bootstrapimpl.VerifySchemaReadiness(conn); readinessErr == nil {
			t.Fatal("VerifySchemaReadiness after rejected duplicate migration = nil, want fail closed")
		}
	})
}

func installLegacyConnectorV2ReceiptProtocolWidth(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorV2Receipt
		MODIFY COLUMN ReceiptProtocolVersion VARCHAR(255)
		CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL`); err != nil {
		t.Fatalf("install legacy Connector V2 receipt protocol width: %v", err)
	}
}

func seedConnectorV2ReceiptProtocolMigrationGraph(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	protocol string,
) {
	t.Helper()
	const (
		epochID   = "receipt-protocol-width-epoch"
		gateID    = "receipt-protocol-width-gate"
		accountID = "receipt-protocol-width-account"
		receiptID = "receipt-protocol-width-receipt"
		keyID     = "receipt-protocol-width-key"
	)
	scope := strings.Repeat("a", 64)
	epochDigest := "sha256:" + strings.Repeat("b", 64)
	payloadDigest := "sha256:" + strings.Repeat("c", 64)
	payload := []byte("retained encrypted receipt payload\x00exact")

	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorServingEpoch (
		Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
		OrgId, AppInstallationId, AuthorizedWorkloadId,
		CatalogRevision, SchemaGeneration, EngineGeneration,
		SourceSnapshotDigest, BindingViewDigest,
		ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
		L2ConfigurationGenerationRef, L2ConfigurationDigest,
		L2VerificationReceipt, L2VerificationReceiptDigest
	) VALUES (?, ?, ?, ?, 'active',
		'org-receipt-width', 'app-install-receipt-width', 'workload-receipt-width',
		1, 1, 1, ?, ?, 'receipt-v1', 'receipt-capability-v1', 1,
		'service-configuration-receipt-width', ?, ?, ?)`,
		epochID,
		epochDigest,
		scope,
		scope,
		"sha256:"+strings.Repeat("1", 64),
		"sha256:"+strings.Repeat("2", 64),
		"sha256:"+strings.Repeat("3", 64),
		[]byte("service configuration owner receipt"),
		"sha256:"+strings.Repeat("4", 64),
	); err != nil {
		t.Fatalf("insert receipt protocol-width ServingEpoch: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorAccountServingGate (
		Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
		ServingEpochId, ServingEpochDigest
	) VALUES (?, ?, ?, 'v2_halted', 41, ?, ?)`, gateID, scope, accountID, epochID, epochDigest); err != nil {
		t.Fatalf("insert receipt protocol-width account gate: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorV2Receipt (
		Id, ReceiptProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, AccountServingFenceToken,
		ServingEpochId, ServingEpochDigest, AdmissionKind,
		AdmissionAuthorityId, PayloadCiphertextProtocolVersion,
		PayloadCiphertext, PayloadDigest, PayloadSizeBytes, ReceiptComponentCount
	) VALUES (?, ?, ?, ?, ?, 41, ?, ?, 'transport', ?,
		'nexuscrm.connector-v2-payload-ciphertext.v1', ?, ?, ?, 0)`,
		receiptID,
		protocol,
		scope,
		accountID,
		gateID,
		epochID,
		epochDigest,
		keyID,
		payload,
		payloadDigest,
		len(payload),
	); err != nil {
		t.Fatalf("insert retained receipt protocol-width row: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorV2ReceiptAdmissionKey (
		Id, AdmissionKeyProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, IngressChannel, TransportIdempotencyKey,
		AuthorizedPayloadDigest, Disposition, ReceiptId,
		PredecessorRefProtocolVersion, PredecessorRef, PredecessorRefDigest
	) VALUES (?, 'nexuscrm.connector-receipt-admission-key.v1', ?, ?, ?, 'webhook',
		'receipt-protocol-width-transport', ?, 'v2_receipt', ?, NULL, NULL, NULL)`,
		keyID,
		scope,
		accountID,
		gateID,
		payloadDigest,
		receiptID,
	); err != nil {
		t.Fatalf("insert receipt protocol-width admission key: %v", err)
	}
}

func assertConnectorV2ReceiptProtocolColumnShape(t *testing.T, db *sql.DB, wantType string) {
	t.Helper()
	var columnType, collation, nullable string
	if err := db.QueryRow(`SELECT COLUMN_TYPE, COLLATION_NAME, IS_NULLABLE
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = '_System_ConnectorV2Receipt'
		  AND COLUMN_NAME = 'ReceiptProtocolVersion'`).Scan(&columnType, &collation, &nullable); err != nil {
		t.Fatalf("inspect Connector V2 receipt protocol column: %v", err)
	}
	if strings.ToLower(columnType) != wantType || collation != "utf8mb4_bin" || nullable != "NO" {
		t.Fatalf(
			"ReceiptProtocolVersion shape = %s/%s/nullable=%s, want %s/utf8mb4_bin/NO",
			columnType,
			collation,
			nullable,
			wantType,
		)
	}
}

func assertConnectorV2ReceiptProtocolMigrationRow(t *testing.T, db *sql.DB, wantProtocol string) {
	t.Helper()
	var protocol, payloadHex, payloadDigest string
	var componentCount int64
	if err := db.QueryRow(`SELECT ReceiptProtocolVersion, HEX(PayloadCiphertext),
		PayloadDigest, ReceiptComponentCount
		FROM _System_ConnectorV2Receipt
		WHERE Id = 'receipt-protocol-width-receipt'`).Scan(
		&protocol,
		&payloadHex,
		&payloadDigest,
		&componentCount,
	); err != nil {
		t.Fatalf("read retained receipt protocol-width row: %v", err)
	}
	wantPayloadHex := strings.ToUpper(fmt.Sprintf("%x", []byte("retained encrypted receipt payload\x00exact")))
	if protocol != wantProtocol || payloadHex != wantPayloadHex ||
		payloadDigest != "sha256:"+strings.Repeat("c", 64) || componentCount != 0 {
		t.Fatalf(
			"retained receipt row = protocol:%q payload:%q digest:%q count:%d, want exact retained values",
			protocol,
			payloadHex,
			payloadDigest,
			componentCount,
		)
	}
}

func assertConnectorV2ReceiptMigrationBootstrapFailed(t *testing.T, db *sql.DB, wantError string) {
	t.Helper()
	var phase, status, lastError string
	if err := db.QueryRow(`SELECT Phase, Status, LastError
		FROM _System_BootstrapState
		WHERE StateKey = 'schema_initialize'`).Scan(&phase, &status, &lastError); err != nil {
		t.Fatalf("read Connector V2 receipt migration bootstrap failure: %v", err)
	}
	if status != "failed" || phase == "completed" || !strings.Contains(lastError, wantError) {
		t.Fatalf(
			"Connector V2 receipt migration bootstrap state = phase:%q status:%q error:%q, want non-completed failed state containing %q",
			phase,
			status,
			lastError,
			wantError,
		)
	}
}

func installLegacyConnectorV2ReceiptTransportIndex(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorV2ReceiptAdmissionKey
		DROP INDEX ux_connector_v2_receipt_key_transport`); err != nil {
		t.Fatalf("drop canonical Connector V2 receipt transport index: %v", err)
	}
	if _, err := db.ExecContext(ctx, `ALTER TABLE _System_ConnectorV2ReceiptAdmissionKey
		ADD UNIQUE INDEX ux_connector_v2_receipt_key_transport
		(ServingScopeDigest, ConnectorAccountId, IngressChannel, TransportIdempotencyKey)`); err != nil {
		t.Fatalf("install legacy Connector V2 receipt transport index: %v", err)
	}
}

func seedConnectorV2ReceiptMigrationGate(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	gateID, servingScopeDigest, connectorAccountID string,
) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorAccountServingGate (
		Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
		ServingEpochId, ServingEpochDigest
	) VALUES (?, ?, ?, 'halted', 41, NULL, NULL)`, gateID, servingScopeDigest, connectorAccountID); err != nil {
		t.Fatalf("insert Connector V2 receipt migration gate %s: %v", gateID, err)
	}
}

func seedConnectorV2ReceiptMigrationAdmissionKey(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	keyID, servingScopeDigest, connectorAccountID, gateID, transportKey string,
) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorV2ReceiptAdmissionKey (
		Id, AdmissionKeyProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, IngressChannel, TransportIdempotencyKey,
		AuthorizedPayloadDigest, Disposition, ReceiptId,
		PredecessorRefProtocolVersion, PredecessorRef, PredecessorRefDigest
	) VALUES (?, 'nexuscrm.connector-receipt-admission-key.v1', ?, ?, ?, 'webhook', ?, ?,
		'v1_tombstone', NULL, 'nexuscrm.connector-v1-predecessor-ref.v1', ?, ?)`,
		keyID,
		servingScopeDigest,
		connectorAccountID,
		gateID,
		transportKey,
		"sha256:"+strings.Repeat("1", 64),
		[]byte("retained v1 admission proof for "+keyID),
		"sha256:"+strings.Repeat("2", 64),
	); err != nil {
		t.Fatalf("insert retained Connector V2 admission key %s: %v", keyID, err)
	}
}

func assertConnectorV2ReceiptMigrationAdmissionKey(
	t *testing.T,
	db *sql.DB,
	keyID, wantScope, wantAccount, wantGate, wantTransportKey string,
) {
	t.Helper()
	var scope, account, gate, channel, transportKey, disposition string
	var receiptID sql.NullString
	if err := db.QueryRow(`SELECT ServingScopeDigest, ConnectorAccountId, AccountServingGateId,
		IngressChannel, TransportIdempotencyKey, Disposition, ReceiptId
		FROM _System_ConnectorV2ReceiptAdmissionKey WHERE Id = ?`, keyID).Scan(
		&scope,
		&account,
		&gate,
		&channel,
		&transportKey,
		&disposition,
		&receiptID,
	); err != nil {
		t.Fatalf("read retained Connector V2 admission key %s: %v", keyID, err)
	}
	if scope != wantScope || account != wantAccount || gate != wantGate ||
		channel != "webhook" || transportKey != wantTransportKey ||
		disposition != "v1_tombstone" || receiptID.Valid {
		t.Fatalf(
			"retained admission key %s = scope:%q account:%q gate:%q channel:%q transport:%q disposition:%q receipt:%v, want exact retained tombstone",
			keyID,
			scope,
			account,
			gate,
			channel,
			transportKey,
			disposition,
			receiptID,
		)
	}
}

func markConnectorV2ReceiptMigrationAsLateFailedBootstrap(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	result, err := db.ExecContext(ctx, `UPDATE _System_BootstrapState
		SET Phase = 'phase6_field_labels', Status = 'failed', LastError = 'V2 receipt index resume proof'
		WHERE StateKey = 'schema_initialize'`)
	if err != nil {
		t.Fatalf("prepare V2 receipt index late failed bootstrap: %v", err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		t.Fatalf("prepare V2 receipt index bootstrap state affected = %d, err = %v, want 1", affected, err)
	}
}

func assertConnectorV2ReceiptTransportIndexShape(t *testing.T, db *sql.DB, wantColumns []string) {
	t.Helper()
	var nonUnique int
	var actualColumns string
	if err := db.QueryRow(`SELECT NON_UNIQUE, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
		FROM INFORMATION_SCHEMA.STATISTICS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = '_System_ConnectorV2ReceiptAdmissionKey'
		  AND INDEX_NAME = 'ux_connector_v2_receipt_key_transport'
		GROUP BY NON_UNIQUE`).Scan(&nonUnique, &actualColumns); err != nil {
		t.Fatalf("read Connector V2 receipt transport index: %v", err)
	}
	if nonUnique != 0 || actualColumns != strings.Join(wantColumns, ",") {
		t.Fatalf(
			"Connector V2 receipt transport index = non_unique:%d columns:%q, want 0/%q",
			nonUnique,
			actualColumns,
			strings.Join(wantColumns, ","),
		)
	}
}

func assertConnectorV2ReceiptCutoverIndexAbsent(t *testing.T, db *sql.DB) {
	t.Helper()
	var count int
	if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = '_System_ConnectorV2ReceiptAdmissionKey'
		  AND INDEX_NAME = 'ux_connector_v2_receipt_key_transport_cutover'`).Scan(&count); err != nil {
		t.Fatalf("inspect Connector V2 receipt cutover index: %v", err)
	}
	if count != 0 {
		t.Fatalf("Connector V2 receipt cutover index rows = %d, want absent", count)
	}
}

func dropConnectorV2IngressReceiptPhysicalTables(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()

	for i := len(connectorV2IngressReceiptAuthorityTableNames) - 1; i >= 0; i-- {
		tableName := connectorV2IngressReceiptAuthorityTableNames[i]
		if _, err := db.ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("drop prior-state table %s: %v", tableName, err)
		}
	}
}

func assertConnectorV2IngressReceiptPhysicalTablesAbsentWithMetadataRetained(t *testing.T, db *sql.DB) {
	t.Helper()

	definitions := make(map[string]schema.TableDefinition, len(connectorV2IngressReceiptAuthorityTableNames))
	for _, definition := range bootstrapimpl.GetManagedTableDefinitions() {
		definitions[definition.TableName] = definition
	}
	for _, tableName := range connectorV2IngressReceiptAuthorityTableNames {
		definition, ok := definitions[tableName]
		if !ok {
			t.Fatalf("managed V2 ingress receipt definition for %s is missing", tableName)
		}
		var physicalCount, registryCount, objectCount, fieldCount int
		if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
			WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, tableName).Scan(&physicalCount); err != nil {
			t.Fatalf("query absent physical table %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Table
			WHERE TableName = ? AND COALESCE(IsDeleted, 0) = 0`, tableName).Scan(&registryCount); err != nil {
			t.Fatalf("query retained registry row %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Object
			WHERE ApiName = ? AND COALESCE(IsDeleted, 0) = 0`, tableName).Scan(&objectCount); err != nil {
			t.Fatalf("query retained object row %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Field
			WHERE ObjectId = ? AND COALESCE(IsDeleted, 0) = 0`, schema.GenerateObjectID(tableName)).Scan(&fieldCount); err != nil {
			t.Fatalf("query retained field rows %s: %v", tableName, err)
		}
		if physicalCount != 0 || registryCount != 1 || objectCount != 1 || fieldCount != len(definition.Columns) {
			t.Fatalf(
				"late-failure state for %s = physical:%d registry:%d object:%d fields:%d, want 0/1/1/%d",
				tableName,
				physicalCount,
				registryCount,
				objectCount,
				fieldCount,
				len(definition.Columns),
			)
		}
	}
}

func connectorV2IngressLegacyFingerprints(t *testing.T, db *sql.DB, gateID, receiptID string) (string, string) {
	t.Helper()

	var gateFingerprint string
	if err := db.QueryRow(`SELECT SHA2(CONCAT_WS('|',
		COALESCE(HEX(Id), '<NULL>'), COALESCE(HEX(ServingScopeDigest), '<NULL>'),
		COALESCE(HEX(ConnectorAccountId), '<NULL>'), COALESCE(HEX(Mode), '<NULL>'),
		COALESCE(HEX(CAST(FenceToken AS CHAR)), '<NULL>'),
		COALESCE(HEX(ServingEpochId), '<NULL>'), COALESCE(HEX(ServingEpochDigest), '<NULL>'),
		COALESCE(HEX(CAST(CreatedDate AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(LastModifiedDate AS CHAR)), '<NULL>')
	), 256) FROM _System_ConnectorAccountServingGate WHERE Id = ?`, gateID).Scan(&gateFingerprint); err != nil {
		t.Fatalf("fingerprint legacy account serving gate: %v", err)
	}

	var receiptFingerprint string
	if err := db.QueryRow(`SELECT SHA2(CONCAT_WS('|',
		COALESCE(HEX(Id), '<NULL>'), COALESCE(HEX(Namespace), '<NULL>'),
		COALESCE(HEX(InstalledPackId), '<NULL>'), COALESCE(HEX(PackVersion), '<NULL>'),
		COALESCE(HEX(SecretName), '<NULL>'), COALESCE(HEX(SecretAlgorithm), '<NULL>'),
		COALESCE(HEX(VerifierCapabilityKind), '<NULL>'), COALESCE(HEX(ParserCapabilityKind), '<NULL>'),
		COALESCE(HEX(RequestMethod), '<NULL>'), COALESCE(HEX(RequestTargetHash), '<NULL>'),
		COALESCE(HEX(RequestTargetCiphertext), '<NULL>'), COALESCE(HEX(ReceiptKey), '<NULL>'),
		COALESCE(HEX(BodyHash), '<NULL>'), COALESCE(HEX(CAST(BodySizeBytes AS CHAR)), '<NULL>'),
		COALESCE(HEX(BodyCiphertext), '<NULL>'), COALESCE(HEX(AllowedHeadersCiphertext), '<NULL>'),
		COALESCE(HEX(Status), '<NULL>'), COALESCE(HEX(CAST(AttemptCount AS CHAR)), '<NULL>'),
		COALESCE(HEX(NextAttemptAt), '<NULL>'), COALESCE(HEX(ClaimedBy), '<NULL>'),
		COALESCE(HEX(ClaimToken), '<NULL>'), COALESCE(HEX(ClaimedAt), '<NULL>'),
		COALESCE(HEX(LeaseExpiresAt), '<NULL>'), COALESCE(HEX(EnqueuedEventIds), '<NULL>'),
		COALESCE(HEX(EnqueueMetadata), '<NULL>'), COALESCE(HEX(LastError), '<NULL>'),
		COALESCE(HEX(LastErrorAt), '<NULL>'), COALESCE(HEX(PurgeAfter), '<NULL>'),
		COALESCE(HEX(CAST(CreatedDate AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(LastModifiedDate AS CHAR)), '<NULL>')
	), 256) FROM _System_ConnectorIngressReceipt WHERE Id = ?`, receiptID).Scan(&receiptFingerprint); err != nil {
		t.Fatalf("fingerprint legacy V1 ingress receipt: %v", err)
	}
	return gateFingerprint, receiptFingerprint
}

func assertConnectorV2IngressLegacyFingerprints(
	t *testing.T,
	db *sql.DB,
	gateID, receiptID, wantGate, wantReceipt string,
) {
	t.Helper()
	gotGate, gotReceipt := connectorV2IngressLegacyFingerprints(t, db, gateID, receiptID)
	if gotGate != wantGate || gotReceipt != wantReceipt {
		t.Fatalf(
			"legacy account-gate/V1-receipt fingerprints = %s/%s, want unchanged %s/%s",
			gotGate,
			gotReceipt,
			wantGate,
			wantReceipt,
		)
	}
}

func assertConnectorV2IngressReceiptRowsEmpty(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, tableName := range connectorV2IngressReceiptAuthorityTableNames {
		var count int
		if err := db.QueryRow("SELECT COUNT(*) FROM `" + tableName + "`").Scan(&count); err != nil {
			t.Fatalf("count stopped-upgrade rows in %s: %v", tableName, err)
		}
		if count != 0 {
			t.Fatalf("stopped-upgrade rows in %s = %d, want 0", tableName, count)
		}
	}
}

func removeConnectorServingEpochReproducibilityPriorState(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()

	dropConnectorServingAuthorityDependents(t, ctx, db)
	for i := len(connectorServingEpochReproducibilityTableNames) - 1; i >= 0; i-- {
		tableName := connectorServingEpochReproducibilityTableNames[i]
		if _, err := db.ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("drop prior-state table %s: %v", tableName, err)
		}
	}
	removeConnectorServingAuthorityMetadata(t, ctx, db, connectorServingEpochReproducibilityTableNames)
}

func connectorServingAuthorityLegacyFingerprints(t *testing.T, db *sql.DB, epochID, bindingID string) (string, string) {
	t.Helper()

	var epochFingerprint string
	if err := db.QueryRow(`SELECT SHA2(CONCAT_WS('|',
		COALESCE(HEX(Id), '<NULL>'), COALESCE(HEX(EpochDigest), '<NULL>'),
		COALESCE(HEX(ServingScopeDigest), '<NULL>'), COALESCE(HEX(ActiveScopeDigest), '<NULL>'),
		COALESCE(HEX(State), '<NULL>'), COALESCE(HEX(OrgId), '<NULL>'),
		COALESCE(HEX(AppInstallationId), '<NULL>'), COALESCE(HEX(AuthorizedWorkloadId), '<NULL>'),
		COALESCE(HEX(CAST(CatalogRevision AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(SchemaGeneration AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(EngineGeneration AS CHAR)), '<NULL>'),
		COALESCE(HEX(SourceSnapshotDigest), '<NULL>'), COALESCE(HEX(BindingViewDigest), '<NULL>'),
		COALESCE(HEX(ReceiptProtocolVersion), '<NULL>'), COALESCE(HEX(ReceiptCapabilityVersion), '<NULL>'),
		COALESCE(HEX(CAST(SecurityEpoch AS CHAR)), '<NULL>'),
		COALESCE(HEX(L2ConfigurationGenerationRef), '<NULL>'),
		COALESCE(HEX(L2ConfigurationDigest), '<NULL>'),
		COALESCE(HEX(L2VerificationReceipt), '<NULL>'),
		COALESCE(HEX(L2VerificationReceiptDigest), '<NULL>'),
		COALESCE(HEX(CAST(CreatedDate AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(LastModifiedDate AS CHAR)), '<NULL>')
	), 256) FROM _System_ConnectorServingEpoch WHERE Id = ?`, epochID).Scan(&epochFingerprint); err != nil {
		t.Fatalf("fingerprint legacy ServingEpoch: %v", err)
	}

	var bindingFingerprint string
	if err := db.QueryRow(`SELECT SHA2(CONCAT_WS('|',
		COALESCE(HEX(Id), '<NULL>'), COALESCE(HEX(ServingEpochId), '<NULL>'),
		COALESCE(HEX(CAST(BindingOrdinal AS CHAR)), '<NULL>'),
		COALESCE(HEX(BindingIdentityDigest), '<NULL>'), COALESCE(HEX(Kind), '<NULL>'),
		COALESCE(HEX(OwnerPack), '<NULL>'), COALESCE(HEX(PackInstallationId), '<NULL>'),
		COALESCE(HEX(PackPayloadDigest), '<NULL>'), COALESCE(HEX(BusinessKey), '<NULL>'),
		COALESCE(HEX(CAST(DefinitionVersion AS CHAR)), '<NULL>'),
		COALESCE(HEX(CredentialRefs), '<NULL>'), COALESCE(HEX(RuntimeCommandKind), '<NULL>'),
		COALESCE(HEX(CAST(RuntimePayloadFormatVersion AS CHAR)), '<NULL>'),
		COALESCE(HEX(RuntimeDefinitionRef), '<NULL>'), COALESCE(HEX(RuntimeDefinitionVersion), '<NULL>'),
		COALESCE(HEX(CAST(CreatedDate AS CHAR)), '<NULL>'),
		COALESCE(HEX(CAST(LastModifiedDate AS CHAR)), '<NULL>')
	), 256) FROM _System_ConnectorServingEpochBinding WHERE Id = ?`, bindingID).Scan(&bindingFingerprint); err != nil {
		t.Fatalf("fingerprint legacy ServingEpoch binding: %v", err)
	}
	return epochFingerprint, bindingFingerprint
}

func assertConnectorServingAuthorityLegacyFingerprints(
	t *testing.T,
	db *sql.DB,
	epochID, bindingID, wantEpoch, wantBinding string,
) {
	t.Helper()
	gotEpoch, gotBinding := connectorServingAuthorityLegacyFingerprints(t, db, epochID, bindingID)
	if gotEpoch != wantEpoch || gotBinding != wantBinding {
		t.Fatalf(
			"legacy ServingEpoch/binding fingerprints = %s/%s, want unchanged %s/%s",
			gotEpoch,
			gotBinding,
			wantEpoch,
			wantBinding,
		)
	}
}

func assertConnectorServingEpochReproducibilityProofRowsEmpty(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, tableName := range connectorServingEpochReproducibilityTableNames {
		var count int
		if err := db.QueryRow("SELECT COUNT(*) FROM `" + tableName + "`").Scan(&count); err != nil {
			t.Fatalf("count stopped-upgrade rows in %s: %v", tableName, err)
		}
		if count != 0 {
			t.Fatalf("stopped-upgrade rows in %s = %d, want 0", tableName, count)
		}
	}
}

func removeConnectorServingAuthorityPriorState(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()

	dropConnectorServingAuthorityDependents(t, ctx, db)
	// The list is parent-first, so reverse traversal removes every dependent
	// immutable proof/admission table before its serving-authority parent.
	for i := len(connectorServingAuthorityTableNames) - 1; i >= 0; i-- {
		tableName := connectorServingAuthorityTableNames[i]
		if _, err := db.ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("drop prior-state table %s: %v", tableName, err)
		}
	}

	removeConnectorServingAuthorityMetadata(t, ctx, db, connectorServingAuthorityTableNames)
}

func dropConnectorServingAuthorityDependents(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()

	for _, tableName := range connectorServingAuthorityDependentTableNames {
		if _, err := db.ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("drop dependent prior-state table %s: %v", tableName, err)
		}
	}
}

func removeConnectorServingAuthorityMetadata(t *testing.T, ctx context.Context, db *sql.DB, tableNames []string) {
	t.Helper()

	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(tableNames)), ",")
	tableArgs := make([]any, len(tableNames))
	objectArgs := make([]any, len(tableNames))
	for i, tableName := range tableNames {
		tableArgs[i] = tableName
		objectArgs[i] = schema.GenerateObjectID(tableName)
	}
	if _, err := db.ExecContext(ctx,
		"DELETE FROM `_System_Field` WHERE `ObjectId` IN ("+placeholders+")",
		objectArgs...,
	); err != nil {
		t.Fatalf("delete prior-state Connector serving-authority fields: %v", err)
	}
	if _, err := db.ExecContext(ctx,
		"DELETE FROM `_System_Object` WHERE `ApiName` IN ("+placeholders+")",
		tableArgs...,
	); err != nil {
		t.Fatalf("delete prior-state Connector serving-authority objects: %v", err)
	}
	if _, err := db.ExecContext(ctx,
		"DELETE FROM `_System_Table` WHERE `TableName` IN ("+placeholders+")",
		tableArgs...,
	); err != nil {
		t.Fatalf("delete prior-state Connector serving-authority registry rows: %v", err)
	}
}

func assertConnectorServingAuthorityAbsent(t *testing.T, db *sql.DB) {
	t.Helper()
	assertConnectorServingAuthorityTablesAbsent(t, db, connectorServingAuthorityTableNames)
}

func assertConnectorServingAuthorityTablesAbsent(t *testing.T, db *sql.DB, tableNames []string) {
	t.Helper()

	for _, tableName := range tableNames {
		var physicalCount, registryCount, objectCount, fieldCount int
		if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
			WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, tableName).Scan(&physicalCount); err != nil {
			t.Fatalf("query prior-state physical table %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Table
			WHERE TableName = ?`, tableName).Scan(&registryCount); err != nil {
			t.Fatalf("query prior-state registry row %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Object
			WHERE ApiName = ?`, tableName).Scan(&objectCount); err != nil {
			t.Fatalf("query prior-state object row %s: %v", tableName, err)
		}
		if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Field
			WHERE ObjectId = ?`, schema.GenerateObjectID(tableName)).Scan(&fieldCount); err != nil {
			t.Fatalf("query prior-state field rows %s: %v", tableName, err)
		}
		if physicalCount != 0 || registryCount != 0 || objectCount != 0 || fieldCount != 0 {
			t.Fatalf(
				"prior state for %s = physical:%d registry:%d object:%d fields:%d, want all absent",
				tableName,
				physicalCount,
				registryCount,
				objectCount,
				fieldCount,
			)
		}
	}
}

func assertConnectorServingAuthorityReady(t *testing.T, db *sql.DB) {
	t.Helper()
	assertConnectorServingAuthorityTablesReady(t, db, connectorServingAuthorityTableNames)
}

func assertConnectorServingAuthorityTablesReady(t *testing.T, db *sql.DB, tableNames []string) {
	t.Helper()

	definitions := make(map[string]schema.TableDefinition, len(tableNames))
	for _, definition := range bootstrapimpl.GetManagedTableDefinitions() {
		definitions[definition.TableName] = definition
	}
	for _, tableName := range tableNames {
		definition, ok := definitions[tableName]
		if !ok {
			t.Fatalf("managed Connector serving-authority definition for %s is missing", tableName)
		}
		assertConnectorServingAuthorityTableRegistered(t, db, definition)
		assertConnectorServingAuthorityBinaryColumns(t, db, definition)
		assertConnectorServingAuthorityIndices(t, db, definition)
		assertConnectorServingAuthorityForeignKeys(t, db, definition)
	}
}

func assertConnectorServingAuthorityTableRegistered(t *testing.T, db *sql.DB, definition schema.TableDefinition) {
	t.Helper()

	var physicalCount, registryCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
		WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?`, definition.TableName).Scan(&physicalCount); err != nil {
		t.Fatalf("query physical table %s: %v", definition.TableName, err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Table
		WHERE TableName = ? AND TableType = 'system_core' AND Category = 'integration'
		  AND COALESCE(IsDeleted, 0) = 0`, definition.TableName).Scan(&registryCount); err != nil {
		t.Fatalf("query registry row %s: %v", definition.TableName, err)
	}
	if physicalCount != 1 || registryCount != 1 {
		t.Fatalf("%s physical/registry counts = %d/%d, want 1/1", definition.TableName, physicalCount, registryCount)
	}

	var readPermission, writePermission string
	if err := db.QueryRow(`SELECT COALESCE(RequiredReadPermission, ''), COALESCE(RequiredWritePermission, '')
		FROM _System_Object WHERE ApiName = ?`, definition.TableName).Scan(&readPermission, &writePermission); err != nil {
		t.Fatalf("query object authority %s: %v", definition.TableName, err)
	}
	if readPermission != "BLOCKED" || writePermission != "BLOCKED" {
		t.Fatalf("%s live permissions = %q/%q, want BLOCKED/BLOCKED", definition.TableName, readPermission, writePermission)
	}

	var fieldCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_Field WHERE ObjectId = ?`,
		schema.GenerateObjectID(definition.TableName),
	).Scan(&fieldCount); err != nil {
		t.Fatalf("query live field metadata %s: %v", definition.TableName, err)
	}
	if fieldCount != len(definition.Columns) {
		t.Fatalf("%s live field count = %d, want %d", definition.TableName, fieldCount, len(definition.Columns))
	}
}

func assertConnectorServingAuthorityBinaryColumns(t *testing.T, db *sql.DB, definition schema.TableDefinition) {
	t.Helper()

	for _, column := range definition.Columns {
		if column.CollationName == "" {
			continue
		}
		var columnType, nullable, collation string
		if err := db.QueryRow(`SELECT COLUMN_TYPE, IS_NULLABLE, COALESCE(COLLATION_NAME, '')
			FROM INFORMATION_SCHEMA.COLUMNS
			WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
			definition.TableName,
			column.Name,
		).Scan(&columnType, &nullable, &collation); err != nil {
			t.Fatalf("query physical column %s.%s: %v", definition.TableName, column.Name, err)
		}
		wantNullable := "NO"
		if column.Nullable {
			wantNullable = "YES"
		}
		if !strings.EqualFold(columnType, column.Type) || nullable != wantNullable || collation != column.CollationName {
			t.Fatalf(
				"%s.%s physical shape = %q/%q/%q, want %q/%q/%q",
				definition.TableName,
				column.Name,
				columnType,
				nullable,
				collation,
				column.Type,
				wantNullable,
				column.CollationName,
			)
		}
	}
}

func assertConnectorServingAuthorityIndices(t *testing.T, db *sql.DB, definition schema.TableDefinition) {
	t.Helper()

	assertIndex := func(indexName string, columns []string, unique bool) {
		t.Helper()
		var nonUnique int
		var actualColumns string
		if err := db.QueryRow(`SELECT NON_UNIQUE, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
			FROM INFORMATION_SCHEMA.STATISTICS
			WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?
			GROUP BY NON_UNIQUE`, definition.TableName, indexName).Scan(&nonUnique, &actualColumns); err != nil {
			t.Fatalf("query %s index %s: %v", definition.TableName, indexName, err)
		}
		wantNonUnique := 1
		if unique {
			wantNonUnique = 0
		}
		if nonUnique != wantNonUnique || actualColumns != strings.Join(columns, ",") {
			t.Fatalf(
				"%s index %s = non_unique:%d columns:%q, want %d/%q",
				definition.TableName,
				indexName,
				nonUnique,
				actualColumns,
				wantNonUnique,
				strings.Join(columns, ","),
			)
		}
	}

	var columnPrimary []string
	for _, column := range definition.Columns {
		if column.PrimaryKey {
			columnPrimary = append(columnPrimary, column.Name)
		}
	}
	hasExplicitPrimary := false
	for _, index := range definition.Indices {
		indexName := index.Name
		if index.Primary {
			hasExplicitPrimary = true
			indexName = "PRIMARY"
		}
		assertIndex(indexName, index.Columns, index.Unique || index.Primary)
	}
	if len(columnPrimary) > 0 && !hasExplicitPrimary {
		assertIndex("PRIMARY", columnPrimary, true)
	}
}

func assertConnectorServingAuthorityForeignKeys(t *testing.T, db *sql.DB, definition schema.TableDefinition) {
	t.Helper()

	for _, foreignKey := range definition.ForeignKeys {
		open := strings.LastIndex(foreignKey.References, "(")
		if open <= 0 || !strings.HasSuffix(foreignKey.References, ")") {
			t.Fatalf("invalid managed foreign-key reference %q", foreignKey.References)
		}
		wantTable := foreignKey.References[:open]
		wantColumn := foreignKey.References[open+1 : len(foreignKey.References)-1]
		var referencedTable, referencedColumn, deleteRule, updateRule string
		if err := db.QueryRow(`SELECT k.REFERENCED_TABLE_NAME, k.REFERENCED_COLUMN_NAME,
			       rc.DELETE_RULE, rc.UPDATE_RULE
			FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE k
			JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc
			  ON rc.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
			 AND rc.TABLE_NAME = k.TABLE_NAME
			 AND rc.CONSTRAINT_NAME = k.CONSTRAINT_NAME
			WHERE k.TABLE_SCHEMA = DATABASE()
			  AND k.TABLE_NAME = ?
			  AND k.COLUMN_NAME = ?
			  AND k.REFERENCED_TABLE_NAME IS NOT NULL`, definition.TableName, foreignKey.Column).Scan(
			&referencedTable,
			&referencedColumn,
			&deleteRule,
			&updateRule,
		); err != nil {
			t.Fatalf("query %s.%s foreign key: %v", definition.TableName, foreignKey.Column, err)
		}
		if referencedTable != wantTable || referencedColumn != wantColumn ||
			!connectorServingAuthorityRestrictiveRule(deleteRule) ||
			!connectorServingAuthorityRestrictiveRule(updateRule) {
			t.Fatalf(
				"%s.%s foreign key = %s(%s) delete:%q update:%q, want %s(%s) with restrictive actions",
				definition.TableName,
				foreignKey.Column,
				referencedTable,
				referencedColumn,
				deleteRule,
				updateRule,
				wantTable,
				wantColumn,
			)
		}
	}
}

func connectorServingAuthorityRestrictiveRule(rule string) bool {
	return strings.EqualFold(rule, "RESTRICT") || strings.EqualFold(rule, "NO ACTION")
}

func assertConnectorServingAuthoritySentinel(t *testing.T, db *sql.DB, sentinelID string) {
	t.Helper()

	var namespace, mode string
	var fenceEpoch int64
	if err := db.QueryRow(`SELECT Namespace, Mode, FenceEpoch
		FROM _System_ConnectorIngressControl WHERE Id = ?`, sentinelID).Scan(&namespace, &mode, &fenceEpoch); err != nil {
		t.Fatalf("read unrelated schema-upgrade sentinel: %v", err)
	}
	if namespace != "serving-authority-upgrade" || mode != "paused" || fenceEpoch != 0 {
		t.Fatalf(
			"unrelated schema-upgrade sentinel = namespace:%q mode:%q fence:%d, want serving-authority-upgrade/paused/0",
			namespace,
			mode,
			fenceEpoch,
		)
	}
}