package persistence

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

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

func TestIntegration_ConnectorV2IngressReceiptSchemaEnforcesImmutableAdmissionIdentity(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()

	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	byName := make(map[string]schema.TableDefinition, len(definitions))
	for _, definition := range definitions {
		byName[definition.TableName] = definition
	}

	const (
		epochTable        = "_System_ConnectorServingEpoch"
		gateTable         = "_System_ConnectorAccountServingGate"
		receiptTable      = "_System_ConnectorV2Receipt"
		admissionKeyTable = "_System_ConnectorV2ReceiptAdmissionKey"
		componentTable    = "_System_ConnectorV2ReceiptComponent"
	)
	legacyTableNames := []string{epochTable, gateTable, receiptTable, admissionKeyTable}
	canonicalTableNames := append(append([]string(nil), legacyTableNames...), componentTable)
	repository := NewSchemaRepository(db, nil)
	for _, tableName := range legacyTableNames {
		definition, ok := byName[tableName]
		if !ok {
			t.Fatalf("%s absent from canonical schema catalog", tableName)
		}
		if tableName == receiptTable {
			definition = connectorV2LegacyReceiptDefinition(t, definition)
		}
		if tableName == admissionKeyTable {
			definition = connectorV2LegacyAdmissionKeyDefinition(t, definition)
		}
		if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
			t.Fatalf("CreatePhysicalTable(%s), legacy pass: %v", tableName, err)
		}
	}
	requireConnectorV2IngressReceiptIndexMissing(t, db, receiptTable, connectorV2ReceiptAccountIndex)
	requireConnectorV2IngressReceiptIndex(
		t,
		db,
		admissionKeyTable,
		connectorV2ReceiptTransportIndex,
		true,
		connectorV2ReceiptLegacyTransportIdentityColumns,
	)
	for pass := 1; pass <= 2; pass++ {
		for _, tableName := range canonicalTableNames {
			if err := repository.CreatePhysicalTable(context.Background(), byName[tableName]); err != nil {
				t.Fatalf("CreatePhysicalTable(%s), canonical warm-upgrade pass %d: %v", tableName, pass, err)
			}
		}
	}

	for _, participant := range []struct{ table, column string }{
		{receiptTable, "Id"},
		{receiptTable, "AccountServingGateId"},
		{receiptTable, "ServingEpochId"},
		{admissionKeyTable, "Id"},
		{admissionKeyTable, "AccountServingGateId"},
		{admissionKeyTable, "ReceiptId"},
		{componentTable, "Id"},
		{componentTable, "ComponentProtocolVersion"},
		{componentTable, "ReceiptId"},
		{componentTable, "PayloadCiphertextProtocolVersion"},
	} {
		requireConnectorV2IngressReceiptVarcharShape(t, db, participant.table, participant.column)
	}
	requireConnectorV2IngressReceiptColumnShape(t, db, receiptTable, "ReceiptProtocolVersion", "varchar(32)")
	requireConnectorV2IngressReceiptColumnShape(t, db, componentTable, "PayloadDigest", "char(71)")
	for _, index := range []struct {
		table   string
		name    string
		unique  bool
		columns []string
	}{
		{receiptTable, "PRIMARY", true, []string{"Id"}},
		{receiptTable, "ux_connector_v2_receipt_admission", true, []string{"ServingScopeDigest", "ConnectorAccountId", "AdmissionKind", "AdmissionAuthorityId"}},
		{receiptTable, "idx_connector_v2_receipt_account", false, []string{"ConnectorAccountId", "Id"}},
		{receiptTable, "idx_connector_v2_receipt_epoch", false, []string{"ServingEpochId", "Id"}},
		{receiptTable, "idx_connector_v2_receipt_gate", false, []string{"AccountServingGateId", "Id"}},
		{admissionKeyTable, "PRIMARY", true, []string{"Id"}},
		{admissionKeyTable, "ux_connector_v2_receipt_key_transport", true, []string{"ConnectorAccountId", "IngressChannel", "TransportIdempotencyKey"}},
		{admissionKeyTable, "ux_connector_v2_receipt_key_receipt", true, []string{"ReceiptId"}},
		{admissionKeyTable, "idx_connector_v2_receipt_key_gate", false, []string{"AccountServingGateId", "Disposition", "Id"}},
		{componentTable, "PRIMARY", true, []string{"Id"}},
		{componentTable, "ux_connector_v2_receipt_component_ordinal", true, []string{"ReceiptId", "ComponentOrdinal"}},
	} {
		requireConnectorV2IngressReceiptIndex(t, db, index.table, index.name, index.unique, index.columns)
	}

	digest := func(value string) string { return "sha256:" + strings.Repeat(value, 64) }
	const (
		epochID   = "v2-ingress-epoch"
		gateID    = "v2-ingress-gate"
		scope     = "1111111111111111111111111111111111111111111111111111111111111111"
		accountID = "connector-account-1"
	)
	if _, err := db.Exec(
		`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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		epochID, digest("1"), scope, scope, "active",
		"org-1", "installation-1", "workload-1",
		uint64(11), uint64(12), uint64(13),
		digest("2"), digest("3"),
		"1.0", "catalog.consumer.integration-binding-view.v1", uint64(14),
		"service-generation-1", digest("4"),
		[]byte("service-owner-receipt"), digest("5"),
	); err != nil {
		t.Fatalf("insert active ServingEpoch: %v", err)
	}
	if _, err := db.Exec(
		`INSERT INTO _System_ConnectorAccountServingGate (
			Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
			ServingEpochId, ServingEpochDigest
		 ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
		gateID, scope, accountID, "v2_serving", int64(7), epochID, digest("1"),
	); err != nil {
		t.Fatalf("insert serving AccountServingGate: %v", err)
	}

	payloadCiphertext := bytes.Repeat([]byte("encrypted-v2-receipt-payload\x00"), 4_096)
	if len(payloadCiphertext) <= 64*1024 {
		t.Fatalf("payload ciphertext probe is only %d bytes", len(payloadCiphertext))
	}
	insertReceipt := `INSERT INTO _System_ConnectorV2Receipt (
		Id, ReceiptProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, AccountServingFenceToken,
		ServingEpochId, ServingEpochDigest,
		AdmissionKind, AdmissionAuthorityId,
		PayloadCiphertextProtocolVersion, PayloadCiphertext,
		PayloadDigest, PayloadSizeBytes, ReceiptComponentCount
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	receiptArgs := func(id, authorityID string, ciphertext []byte) []any {
		return []any{
			id, "nexuscrm.connector-v2-receipt.v1", scope, accountID,
			gateID, int64(7), epochID, digest("1"),
			"transport", authorityID,
			"nexuscrm.connector-v2-payload-ciphertext.v1", ciphertext,
			digest("6"), int64(12345), int64(2),
		}
	}
	if _, err := db.Exec(insertReceipt, receiptArgs("v2-receipt-1", "v2-admission-key-1", payloadCiphertext)...); err != nil {
		t.Fatalf("insert V2 Receipt: %v", err)
	}
	if _, err := db.Exec(insertReceipt, receiptArgs("v2-receipt-admission-conflict", "v2-admission-key-1", []byte("other"))...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate receipt admission error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertReceipt, receiptArgs("v2-receipt-case-distinct", "V2-Admission-Key-1", []byte("case-distinct"))...); err != nil {
		t.Fatalf("insert byte-distinct receipt admission identity: %v", err)
	}

	var storedCiphertext []byte
	var storedCiphertextBytes int
	if err := db.QueryRow(
		`SELECT PayloadCiphertext, OCTET_LENGTH(PayloadCiphertext)
		   FROM _System_ConnectorV2Receipt WHERE Id = ?`,
		"v2-receipt-1",
	).Scan(&storedCiphertext, &storedCiphertextBytes); err != nil {
		t.Fatalf("read exact receipt ciphertext: %v", err)
	}
	if !bytes.Equal(storedCiphertext, payloadCiphertext) || storedCiphertextBytes != len(payloadCiphertext) {
		t.Fatalf("stored receipt ciphertext = %d/%d bytes, want exact %d bytes", len(storedCiphertext), storedCiphertextBytes, len(payloadCiphertext))
	}

	insertComponent := `INSERT INTO _System_ConnectorV2ReceiptComponent (
		Id, ComponentProtocolVersion, ReceiptId, ComponentOrdinal,
		PayloadCiphertextProtocolVersion, PayloadCiphertext,
		PayloadDigest, PayloadSizeBytes
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
	componentCiphertext := bytes.Repeat([]byte("encrypted-v2-receipt-component\x00"), 4_096)
	if len(componentCiphertext) <= 64*1024 {
		t.Fatalf("component ciphertext probe is only %d bytes", len(componentCiphertext))
	}
	componentArgs := func(id, receiptID string, ordinal int, ciphertext []byte) []any {
		return []any{
			id,
			"nexuscrm.connector-v2-receipt-component.v1",
			receiptID,
			ordinal,
			"nexuscrm.connector-v2-payload-ciphertext.v1",
			ciphertext,
			digest("9"),
			int64(1234),
		}
	}
	if _, err := db.Exec(insertComponent, componentArgs("v2-component-1", "v2-receipt-1", 0, componentCiphertext)...); err != nil {
		t.Fatalf("insert first V2 ReceiptComponent: %v", err)
	}
	if _, err := db.Exec(insertComponent, componentArgs("v2-component-2", "v2-receipt-1", 1, []byte("second-component"))...); err != nil {
		t.Fatalf("insert second V2 ReceiptComponent: %v", err)
	}
	if _, err := db.Exec(insertComponent, componentArgs("v2-component-duplicate-ordinal", "v2-receipt-1", 1, []byte("duplicate"))...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate receipt component ordinal error = %v, want duplicate-key rejection", err)
	}
	var storedComponentCiphertext []byte
	if err := db.QueryRow(
		`SELECT PayloadCiphertext
		   FROM _System_ConnectorV2ReceiptComponent
		  WHERE ReceiptId = ? AND ComponentOrdinal = ?`,
		"v2-receipt-1",
		0,
	).Scan(&storedComponentCiphertext); err != nil {
		t.Fatalf("read exact V2 ReceiptComponent ciphertext: %v", err)
	}
	if !bytes.Equal(storedComponentCiphertext, componentCiphertext) {
		t.Fatalf("stored component ciphertext = %d bytes, want exact %d bytes", len(storedComponentCiphertext), len(componentCiphertext))
	}

	insertAdmissionKey := `INSERT INTO _System_ConnectorV2ReceiptAdmissionKey (
		Id, AdmissionKeyProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, IngressChannel, TransportIdempotencyKey,
		AuthorizedPayloadDigest, Disposition, ReceiptId,
		PredecessorRefProtocolVersion, PredecessorRef, PredecessorRefDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	keyArgs := func(id, transportKey string, receiptID any) []any {
		return []any{
			id, "nexuscrm.connector-v2-receipt-admission-key.v1", scope, accountID,
			gateID, "webhook", transportKey,
			digest("6"), "v2_receipt", receiptID,
			nil, nil, nil,
		}
	}
	if _, err := db.Exec(insertAdmissionKey, keyArgs("v2-admission-key-1", "delivery-1", "v2-receipt-1")...); err != nil {
		t.Fatalf("insert V2 ReceiptAdmissionKey: %v", err)
	}
	if _, err := db.Exec(insertAdmissionKey, keyArgs("v2-admission-key-duplicate", "delivery-1", "v2-receipt-case-distinct")...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate transport identity error = %v, want duplicate-key rejection", err)
	}
	crossScopeArgs := keyArgs("v2-admission-key-cross-scope", "delivery-1", "v2-receipt-case-distinct")
	crossScopeArgs[2] = strings.Repeat("9", 64)
	if _, err := db.Exec(insertAdmissionKey, crossScopeArgs...); !IsDuplicateKeyError(err) {
		t.Fatalf("cross-scope duplicate transport identity error = %v, want duplicate-key rejection", err)
	}
	if _, err := db.Exec(insertAdmissionKey, keyArgs("v2-admission-key-case-distinct", "Delivery-1", "v2-receipt-case-distinct")...); err != nil {
		t.Fatalf("insert byte-distinct transport identity: %v", err)
	}
	if _, err := db.Exec(insertAdmissionKey, keyArgs("v2-admission-key-receipt-conflict", "delivery-2", "v2-receipt-1")...); !IsDuplicateKeyError(err) {
		t.Fatalf("duplicate ReceiptId error = %v, want duplicate-key rejection", err)
	}

	predecessorRef := bytes.Repeat([]byte("disclosure-safe-v1-predecessor\x00"), 2_500)
	if len(predecessorRef) <= 64*1024 {
		t.Fatalf("predecessor ref probe is only %d bytes", len(predecessorRef))
	}
	tombstoneArgs := func(id, transportKey string, predecessor []byte) []any {
		return []any{
			id, "nexuscrm.connector-v2-receipt-admission-key.v1", scope, accountID,
			gateID, "webhook", transportKey,
			digest("7"), "v1_tombstone", nil,
			"nexuscrm.connector-v1-predecessor-ref.v1", predecessor, digest("8"),
		}
	}
	if _, err := db.Exec(insertAdmissionKey, tombstoneArgs("v1-tombstone-1", "legacy-delivery-1", predecessorRef)...); err != nil {
		t.Fatalf("insert first V1 acceptance tombstone: %v", err)
	}
	if _, err := db.Exec(insertAdmissionKey, tombstoneArgs("v1-tombstone-2", "legacy-delivery-2", []byte("second"))...); err != nil {
		t.Fatalf("insert second NULL-ReceiptId tombstone: %v", err)
	}
	var storedPredecessor []byte
	if err := db.QueryRow(
		`SELECT PredecessorRef FROM _System_ConnectorV2ReceiptAdmissionKey WHERE Id = ?`,
		"v1-tombstone-1",
	).Scan(&storedPredecessor); err != nil {
		t.Fatalf("read exact V1 predecessor ref: %v", err)
	}
	if !bytes.Equal(storedPredecessor, predecessorRef) {
		t.Fatalf("stored predecessor ref = %d bytes, want exact %d bytes", len(storedPredecessor), len(predecessorRef))
	}

	missingGateArgs := receiptArgs("missing-gate-receipt", "missing-gate-authority", []byte("x"))
	missingGateArgs[4] = "missing-gate"
	_, err = db.Exec(insertReceipt, missingGateArgs...)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1452, "Receipt for a missing AccountServingGate")
	missingEpochArgs := receiptArgs("missing-epoch-receipt", "missing-epoch-authority", []byte("x"))
	missingEpochArgs[6] = "missing-epoch"
	_, err = db.Exec(insertReceipt, missingEpochArgs...)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1452, "Receipt for a missing ServingEpoch")
	_, err = db.Exec(insertAdmissionKey, keyArgs("missing-receipt-key", "missing-receipt", "missing-receipt")...)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1452, "ReceiptAdmissionKey for a missing Receipt")
	_, err = db.Exec(insertComponent, componentArgs("missing-receipt-component", "missing-receipt", 0, []byte("x"))...)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1452, "ReceiptComponent for a missing Receipt")

	_, err = db.Exec(`DELETE FROM _System_ConnectorV2Receipt WHERE Id = ?`, "v2-receipt-1")
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1451, "delete referenced V2 Receipt")
	_, err = db.Exec(`DELETE FROM _System_ConnectorAccountServingGate WHERE Id = ?`, gateID)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1451, "delete referenced AccountServingGate")
	_, err = db.Exec(`DELETE FROM _System_ConnectorServingEpoch WHERE Id = ?`, epochID)
	requireConnectorV2IngressReceiptForeignKeyError(t, err, 1451, "delete referenced ServingEpoch")
}

func TestIntegration_ConnectorV2IngressReceiptSchemaRejectsConflictingWarmUpgrade(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()

	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	byName := make(map[string]schema.TableDefinition, len(definitions))
	for _, definition := range definitions {
		byName[definition.TableName] = definition
	}

	const (
		epochTable        = "_System_ConnectorServingEpoch"
		gateTable         = "_System_ConnectorAccountServingGate"
		receiptTable      = "_System_ConnectorV2Receipt"
		admissionKeyTable = "_System_ConnectorV2ReceiptAdmissionKey"
		accountID         = "connector-account-warm-upgrade"
		transportKey      = "delivery-warm-upgrade"
	)
	repository := NewSchemaRepository(db, nil)
	for _, tableName := range []string{epochTable, gateTable, receiptTable, admissionKeyTable} {
		definition, ok := byName[tableName]
		if !ok {
			t.Fatalf("%s absent from canonical schema catalog", tableName)
		}
		if tableName == admissionKeyTable {
			definition = connectorV2LegacyAdmissionKeyDefinition(t, definition)
		}
		if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
			t.Fatalf("CreatePhysicalTable(%s), legacy setup: %v", tableName, err)
		}
	}

	digest := func(value string) string { return "sha256:" + strings.Repeat(value, 64) }
	insertGate := `INSERT INTO _System_ConnectorAccountServingGate (
		Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
		ServingEpochId, ServingEpochDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?)`
	insertTombstone := `INSERT INTO _System_ConnectorV2ReceiptAdmissionKey (
		Id, AdmissionKeyProtocolVersion, ServingScopeDigest, ConnectorAccountId,
		AccountServingGateId, IngressChannel, TransportIdempotencyKey,
		AuthorizedPayloadDigest, Disposition, ReceiptId,
		PredecessorRefProtocolVersion, PredecessorRef, PredecessorRefDigest
	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	for i, scopeByte := range []string{"1", "2"} {
		scope := strings.Repeat(scopeByte, 64)
		gateID := "warm-upgrade-gate-" + scopeByte
		if _, err := db.Exec(insertGate, gateID, scope, accountID, "v1_serving", int64(0), nil, nil); err != nil {
			t.Fatalf("insert retained V1 gate %d: %v", i, err)
		}
		if _, err := db.Exec(
			insertTombstone,
			"warm-upgrade-key-"+scopeByte,
			"nexuscrm.connector-v2-receipt-admission-key.v1",
			scope,
			accountID,
			gateID,
			"webhook",
			transportKey,
			digest("3"),
			"v1_tombstone",
			nil,
			"nexuscrm.connector-v1-predecessor-ref.v1",
			[]byte("retained-v1-"+scopeByte),
			digest("4"),
		); err != nil {
			t.Fatalf("insert retained V1 admission authority %d: %v", i, err)
		}
	}

	err = repository.CreatePhysicalTable(context.Background(), byName[admissionKeyTable])
	if err == nil || !strings.Contains(err.Error(), "retained rows conflict") {
		t.Fatalf("CreatePhysicalTable(canonical warm upgrade) error = %v", err)
	}
	var retainedRows int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_ConnectorV2ReceiptAdmissionKey`).Scan(&retainedRows); err != nil {
		t.Fatalf("count retained admission authorities: %v", err)
	}
	if retainedRows != 2 {
		t.Fatalf("retained admission authority rows = %d, want 2", retainedRows)
	}
	requireConnectorV2IngressReceiptIndex(
		t,
		db,
		admissionKeyTable,
		connectorV2ReceiptTransportIndex,
		true,
		connectorV2ReceiptLegacyTransportIdentityColumns,
	)
}

func connectorV2LegacyAdmissionKeyDefinition(t *testing.T, definition schema.TableDefinition) schema.TableDefinition {
	t.Helper()
	legacy := definition
	legacy.Indices = append([]schema.IndexDefinition(nil), definition.Indices...)
	for i := range legacy.Indices {
		if legacy.Indices[i].Name != connectorV2ReceiptTransportIndex {
			continue
		}
		legacy.Indices[i].Columns = append([]string(nil), connectorV2ReceiptLegacyTransportIdentityColumns...)
		return legacy
	}
	t.Fatalf("%s is absent from %s", connectorV2ReceiptTransportIndex, definition.TableName)
	return schema.TableDefinition{}
}

func connectorV2LegacyReceiptDefinition(t *testing.T, definition schema.TableDefinition) schema.TableDefinition {
	t.Helper()
	legacy := definition
	legacy.Columns = append([]schema.ColumnDefinition(nil), definition.Columns...)
	foundReceiptProtocol := false
	for index := range legacy.Columns {
		if legacy.Columns[index].Name != "ReceiptProtocolVersion" {
			continue
		}
		legacy.Columns[index].Type = "VARCHAR(255)"
		foundReceiptProtocol = true
		break
	}
	if !foundReceiptProtocol {
		t.Fatalf("ReceiptProtocolVersion is absent from %s", definition.TableName)
	}
	legacy.Indices = make([]schema.IndexDefinition, 0, len(definition.Indices)-1)
	for _, index := range definition.Indices {
		if index.Name == connectorV2ReceiptAccountIndex {
			continue
		}
		legacy.Indices = append(legacy.Indices, index)
	}
	if len(legacy.Indices) != len(definition.Indices)-1 {
		t.Fatalf("%s is absent from %s", connectorV2ReceiptAccountIndex, definition.TableName)
	}
	return legacy
}

func requireConnectorV2IngressReceiptVarcharShape(t *testing.T, db interface {
	QueryRow(query string, args ...any) *sql.Row
}, tableName, columnName string) {
	t.Helper()
	requireConnectorV2IngressReceiptColumnShape(t, db, tableName, columnName, "varchar(255)")
}

func requireConnectorV2IngressReceiptColumnShape(t *testing.T, db interface {
	QueryRow(query string, args ...any) *sql.Row
}, tableName, columnName, wantColumnType string) {
	t.Helper()

	var columnType, collation string
	if err := db.QueryRow(
		`SELECT COLUMN_TYPE, COLLATION_NAME
		   FROM INFORMATION_SCHEMA.COLUMNS
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
		tableName,
		columnName,
	).Scan(&columnType, &collation); err != nil {
		t.Fatalf("inspect %s.%s physical shape: %v", tableName, columnName, err)
	}
	if columnType != wantColumnType || collation != "utf8mb4_bin" {
		t.Fatalf("%s.%s physical shape = %s/%s, want %s/utf8mb4_bin", tableName, columnName, columnType, collation, wantColumnType)
	}
}

func requireConnectorV2IngressReceiptIndex(
	t *testing.T,
	db interface {
		Query(query string, args ...any) (*sql.Rows, error)
	},
	tableName string,
	indexName string,
	unique bool,
	wantColumns []string,
) {
	t.Helper()

	rows, err := db.Query(
		`SELECT NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, SUB_PART
		   FROM INFORMATION_SCHEMA.STATISTICS
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?
		  ORDER BY SEQ_IN_INDEX`,
		tableName,
		indexName,
	)
	if err != nil {
		t.Fatalf("inspect %s.%s: %v", tableName, indexName, err)
	}
	defer rows.Close()

	gotColumns := make([]string, 0, len(wantColumns))
	for rows.Next() {
		var nonUnique, sequence int
		var column string
		var subPart sql.NullInt64
		if err := rows.Scan(&nonUnique, &sequence, &column, &subPart); err != nil {
			t.Fatalf("scan %s.%s: %v", tableName, indexName, err)
		}
		if (nonUnique == 0) != unique {
			t.Fatalf("%s.%s NON_UNIQUE = %d, want unique %v", tableName, indexName, nonUnique, unique)
		}
		if sequence != len(gotColumns)+1 {
			t.Fatalf("%s.%s sequence = %d, want %d", tableName, indexName, sequence, len(gotColumns)+1)
		}
		if subPart.Valid {
			t.Fatalf("%s.%s column %s is a prefix index of %d bytes", tableName, indexName, column, subPart.Int64)
		}
		gotColumns = append(gotColumns, column)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("iterate %s.%s: %v", tableName, indexName, err)
	}
	if strings.Join(gotColumns, ",") != strings.Join(wantColumns, ",") {
		t.Fatalf("%s.%s columns = %v, want %v", tableName, indexName, gotColumns, wantColumns)
	}
}

func requireConnectorV2IngressReceiptIndexMissing(t *testing.T, db interface {
	QueryRow(query string, args ...any) *sql.Row
}, tableName, indexName string) {
	t.Helper()
	var count int
	if err := db.QueryRow(
		`SELECT COUNT(*)
		   FROM INFORMATION_SCHEMA.STATISTICS
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?`,
		tableName,
		indexName,
	).Scan(&count); err != nil {
		t.Fatalf("inspect absent %s.%s index: %v", tableName, indexName, err)
	}
	if count != 0 {
		t.Fatalf("%s.%s unexpectedly exists before canonical warm upgrade", tableName, indexName)
	}
}

func requireConnectorV2IngressReceiptForeignKeyError(t *testing.T, err error, number uint16, operation string) {
	t.Helper()
	if err == nil {
		t.Fatalf("%s succeeded, want foreign-key rejection", operation)
	}
	var mysqlErr *mysql.MySQLError
	if !errors.As(err, &mysqlErr) || mysqlErr.Number != number {
		t.Fatalf("%s error = %v, want MySQL %d foreign-key rejection", operation, err, number)
	}
}