//go:build integration

package controlplane

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

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-controlplane/bootstrapimpl"
)

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

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	assertConnectedDeliverySchemaReady(t, conn.DB())
	if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err != nil {
		t.Fatalf("operator verify clean connected delivery schema: %v", err)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("idempotent connected delivery schema reapply: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedAuthorizationRequest+"` DROP INDEX `ux_connected_authorization_state_digest`"); err != nil {
		t.Fatalf("drop connected authorization unique index: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, constants.TableConnectedAuthorizationRequest+" index ux_connected_authorization_state_digest composition/uniqueness does not match")
	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("repair connected authorization unique index drift: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` DROP INDEX `idx_connected_delivery_job_claim`"); err != nil {
		t.Fatalf("drop connected delivery claim index: %v", err)
	}
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` ADD INDEX `idx_connected_delivery_job_claim` (`State`, `JobId`)"); err != nil {
		t.Fatalf("reshape connected delivery claim index: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, constants.TableConnectedDeliveryJob+" index idx_connected_delivery_job_claim composition/uniqueness does not match")
	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("repair connected delivery claim index drift: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	constraintName := connectedDeliveryForeignKeyConstraintName(t, conn.DB(), constants.TableConnectedDeliveryJob, "ConnectedPrincipalId")
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` DROP FOREIGN KEY `"+constraintName+"`"); err != nil {
		t.Fatalf("drop connected delivery foreign key: %v", err)
	}
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` ADD CONSTRAINT `fk_connected_delivery_job_principal_drift` FOREIGN KEY (`ConnectedPrincipalId`) REFERENCES `User` (`Id`) ON DELETE CASCADE"); err != nil {
		t.Fatalf("reshape connected delivery foreign key: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, constants.TableConnectedDeliveryJob+".ConnectedPrincipalId foreign key target/delete rule does not match")
	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("reapply with unsupported connected delivery foreign key drift: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, "manual remediation")
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` DROP FOREIGN KEY `fk_connected_delivery_job_principal_drift`"); err != nil {
		t.Fatalf("drop drifted connected delivery foreign key: %v", err)
	}
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` ADD CONSTRAINT `fk_connected_delivery_job_principal` FOREIGN KEY (`ConnectedPrincipalId`) REFERENCES `"+constants.TableConnectedPrincipal+"` (`Id`) ON DELETE RESTRICT"); err != nil {
		t.Fatalf("restore connected delivery foreign key: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryPayload+"` MODIFY COLUMN `LastModifiedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)"); err != nil {
		t.Fatalf("add forbidden connected delivery automatic update: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, constants.TableConnectedDeliveryPayload+".LastModifiedDate has forbidden ON UPDATE behavior")
	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryPayload+"` MODIFY COLUMN `LastModifiedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)"); err != nil {
		t.Fatalf("remove forbidden connected delivery automatic update: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryPayload+"` DROP COLUMN `LastModifiedDate`"); err != nil {
		t.Fatalf("drift connected delivery payload temporal column: %v", err)
	}
	if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err == nil || !strings.Contains(err.Error(), constants.TableConnectedDeliveryPayload+".LastModifiedDate is missing") {
		t.Fatalf("VerifyConnectedDeliverySchema() after payload drift = %v, want missing LastModifiedDate", err)
	}
	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("repair connected delivery payload drift: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())

	for _, tableName := range []string{
		constants.TableConnectedDeliveryJob,
		constants.TableConnectedDeliveryPayload,
		constants.TableConnectedPrincipalToken,
		constants.TableConnectedAuthorizationRequest,
		constants.TableConnectedPrincipal,
	} {
		if _, err := conn.DB().ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("prepare supported prior state by dropping %s: %v", tableName, err)
		}
	}
	if err := bootstrapimpl.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("pod readiness must tolerate dormant connected delivery schema on completed-marker warm upgrade: %v", err)
	}
	assertConnectedDeliverySchemaDrift(t, conn, constants.TableConnectedPrincipal+" table is missing")
	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("upgrade connected delivery schema from supported prior state: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())
	if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err != nil {
		t.Fatalf("operator verify upgraded connected delivery schema: %v", err)
	}
}

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

	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new runner for supported-prior fixture: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh cache for supported-prior fixture: %v", err)
	}
	if err := ApplyRuntimeRequiredPacks(ctx, runner); err != nil {
		t.Fatalf("apply runtime-required pack baseline for supported-prior fixture: %v", err)
	}
	if err := Verify(ctx, conn); err != nil {
		t.Fatalf("controlplane.Verify() before supported-prior fixture: %v", err)
	}

	removeConnectedDeliverySchemaForSupportedPriorState(t, ctx, conn.DB())
	assertConnectedDeliveryPriorStateAbsent(t, conn.DB())

	strictResult, strictErr := schemaProvisioner.RunAssertions(conn, true)
	if strictErr == nil {
		t.Fatal("strict RunAssertions() on supported prior state = nil, want the historical pod assertion blocker")
	}
	connectedViolations := 0
	for _, violation := range strictResult.Violations {
		if isConnectedDeliveryIntegrationTable(violation.Object) {
			connectedViolations++
		}
	}
	if connectedViolations != len(connectedDeliveryIntegrationTableNames()) {
		t.Fatalf("strict RunAssertions() connected-delivery visibility violations = %d, want %d; error: %v", connectedViolations, len(connectedDeliveryIntegrationTableNames()), strictErr)
	}

	if err := Verify(ctx, conn); err != nil {
		t.Fatalf("controlplane.Verify() must allow supported prior connected-delivery state: %v", err)
	}
	if err := VerifyDeep(ctx, conn); err == nil || !strings.Contains(err.Error(), constants.TableConnectedPrincipal+" table is missing") {
		t.Fatalf("controlplane.VerifyDeep() on supported prior state = %v, want connected-delivery drift", err)
	}

	if err := schemaProvisioner.InitializeSchema(conn); err != nil {
		t.Fatalf("upgrade connected-delivery schema from supported prior state: %v", err)
	}
	if err := ReconcileManagedSystemObjectVisibility(conn.DB(), managedTableDefinitions()); err != nil {
		t.Fatalf("reconcile upgraded connected-delivery object visibility: %v", err)
	}
	assertConnectedDeliverySchemaReady(t, conn.DB())
	if err := Verify(ctx, conn); err != nil {
		t.Fatalf("controlplane.Verify() after supported prior upgrade: %v", err)
	}
	if err := VerifyDeep(ctx, conn); err != nil {
		t.Fatalf("controlplane.VerifyDeep() after supported prior upgrade: %v", err)
	}
}

func TestConnectedDeliverySchemaRemediationClassification(t *testing.T) {
	t.Run("missing primary key converges and stays idempotent", func(t *testing.T) {
		withConnectedDeliveryIntegrationDatabase(t, func(ctx context.Context, conn *database.TiDBConnection) {
			recreateConnectedDeliveryPayloadWithPrimary(t, ctx, conn.DB(), "JobId")
			assertConnectedDeliveryPrimary(t, conn.DB(), constants.TableConnectedDeliveryPayload, "JobId")
			if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryPayload+"` DROP PRIMARY KEY"); err != nil {
				t.Fatalf("drop connected delivery payload primary key: %v", err)
			}
			err := bootstrapimpl.VerifyConnectedDeliverySchema(conn)
			if err == nil || !strings.Contains(err.Error(), constants.TableConnectedDeliveryPayload+" primary key is missing") || !strings.Contains(err.Error(), "run 'cmd/setup apply'") {
				t.Fatalf("VerifyConnectedDeliverySchema() after missing primary key = %v, want apply-repairable guidance", err)
			}
			if strings.Contains(err.Error(), "manual remediation") {
				t.Fatalf("VerifyConnectedDeliverySchema() after missing primary key = %v, do not want manual remediation", err)
			}
			if err := schemaProvisioner.InitializeSchema(conn); err != nil {
				t.Fatalf("repair missing connected delivery payload primary key: %v", err)
			}
			if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err != nil {
				t.Fatalf("verify repaired connected delivery payload primary key: %v", err)
			}
			assertConnectedDeliveryPrimary(t, conn.DB(), constants.TableConnectedDeliveryPayload, "JobId")
			if err := schemaProvisioner.InitializeSchema(conn); err != nil {
				t.Fatalf("idempotent reapply after primary repair: %v", err)
			}
			if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err != nil {
				t.Fatalf("verify idempotent connected delivery payload primary repair: %v", err)
			}
			assertConnectedDeliveryPrimary(t, conn.DB(), constants.TableConnectedDeliveryPayload, "JobId")
		})
	})

	t.Run("malformed primary key remains manual and unreplaced", func(t *testing.T) {
		withConnectedDeliveryIntegrationDatabase(t, func(ctx context.Context, conn *database.TiDBConnection) {
			recreateConnectedDeliveryPayloadWithPrimary(t, ctx, conn.DB(), "PayloadSha256")
			want := constants.TableConnectedDeliveryPayload + " primary key composition/uniqueness does not match"
			err := bootstrapimpl.VerifyConnectedDeliverySchema(conn)
			if err == nil || !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "manual remediation") {
				t.Fatalf("VerifyConnectedDeliverySchema() after malformed primary key = %v, want manual remediation", err)
			}
			if strings.Contains(err.Error(), "run 'cmd/setup apply' for repairable") {
				t.Fatalf("VerifyConnectedDeliverySchema() after malformed primary key = %v, do not want apply-repairable guidance", err)
			}
			err = schemaProvisioner.InitializeSchema(conn)
			if err == nil || !strings.Contains(err.Error(), "primary key drift") {
				t.Fatalf("InitializeSchema() after malformed primary key = %v, want fail-closed drift", err)
			}
			assertConnectedDeliveryPrimary(t, conn.DB(), constants.TableConnectedDeliveryPayload, "PayloadSha256")
			err = bootstrapimpl.VerifyConnectedDeliverySchema(conn)
			if err == nil || !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "manual remediation") {
				t.Fatalf("VerifyConnectedDeliverySchema() after malformed primary reapply = %v, want drift to remain manual", err)
			}
		})
	})

	t.Run("uniqueness downgrade remains manual after apply", func(t *testing.T) {
		withConnectedDeliveryIntegrationDatabase(t, func(ctx context.Context, conn *database.TiDBConnection) {
			if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` DROP INDEX `idx_connected_delivery_job_claim`"); err != nil {
				t.Fatalf("drop connected delivery claim index: %v", err)
			}
			if _, err := conn.DB().ExecContext(ctx, "ALTER TABLE `"+constants.TableConnectedDeliveryJob+"` ADD UNIQUE INDEX `idx_connected_delivery_job_claim` (`State`, `LeaseExpiresAt`, `JobId`)"); err != nil {
				t.Fatalf("make connected delivery claim index unique: %v", err)
			}
			want := constants.TableConnectedDeliveryJob + " index idx_connected_delivery_job_claim is unique but the managed catalog expects non-unique"
			err := bootstrapimpl.VerifyConnectedDeliverySchema(conn)
			if err == nil || !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "manual remediation") {
				t.Fatalf("VerifyConnectedDeliverySchema() after uniqueness downgrade drift = %v, want manual remediation", err)
			}
			if strings.Contains(err.Error(), "run 'cmd/setup apply' for repairable") {
				t.Fatalf("VerifyConnectedDeliverySchema() after uniqueness downgrade drift = %v, do not want apply-repairable guidance", err)
			}
			if err := schemaProvisioner.InitializeSchema(conn); err != nil {
				t.Fatalf("reapply with unsupported connected delivery uniqueness downgrade: %v", err)
			}
			err = bootstrapimpl.VerifyConnectedDeliverySchema(conn)
			if err == nil || !strings.Contains(err.Error(), want) || !strings.Contains(err.Error(), "manual remediation") {
				t.Fatalf("VerifyConnectedDeliverySchema() after uniqueness downgrade reapply = %v, want drift to remain manual", err)
			}
			assertConnectedDeliveryIndex(t, conn.DB(), constants.TableConnectedDeliveryJob, "idx_connected_delivery_job_claim", 0, "State,LeaseExpiresAt,JobId")
		})
	})
}

func withConnectedDeliveryIntegrationDatabase(t *testing.T, test func(context.Context, *database.TiDBConnection)) {
	t.Helper()
	ctx := context.Background()
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	defer cleanup()
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	defer conn.Close()
	test(ctx, conn)
}

func removeConnectedDeliverySchemaForSupportedPriorState(t *testing.T, ctx context.Context, db *sql.DB) {
	t.Helper()
	for _, tableName := range []string{
		constants.TableConnectedDeliveryJob,
		constants.TableConnectedDeliveryPayload,
		constants.TableConnectedPrincipalToken,
		constants.TableConnectedAuthorizationRequest,
		constants.TableConnectedPrincipal,
	} {
		if _, err := db.ExecContext(ctx, "DROP TABLE `"+tableName+"`"); err != nil {
			t.Fatalf("drop supported-prior table %s: %v", tableName, err)
		}
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(connectedDeliveryIntegrationTableNames())), ",")
	args := make([]any, len(connectedDeliveryIntegrationTableNames()))
	for i, tableName := range connectedDeliveryIntegrationTableNames() {
		args[i] = tableName
	}
	if _, err := db.ExecContext(ctx,
		"DELETE f FROM `_System_Field` f JOIN `_System_Object` o ON o.`Id` = f.`ObjectId` WHERE o.`ApiName` IN ("+placeholders+")",
		args...,
	); err != nil {
		t.Fatalf("delete supported-prior connected-delivery fields: %v", err)
	}
	if _, err := db.ExecContext(ctx,
		"DELETE FROM `_System_Object` WHERE `ApiName` IN ("+placeholders+")",
		args...,
	); err != nil {
		t.Fatalf("delete supported-prior connected-delivery objects: %v", err)
	}
}

func assertConnectedDeliveryPriorStateAbsent(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, tableName := range connectedDeliveryIntegrationTableNames() {
		var physicalCount, metadataCount 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 supported-prior physical table %s: %v", tableName, err)
		}
		if err := db.QueryRow("SELECT COUNT(*) FROM `_System_Object` WHERE `ApiName` = ?", tableName).Scan(&metadataCount); err != nil {
			t.Fatalf("query supported-prior object row %s: %v", tableName, err)
		}
		if physicalCount != 0 || metadataCount != 0 {
			t.Fatalf("supported-prior state for %s = physical %d metadata %d, want 0/0", tableName, physicalCount, metadataCount)
		}
	}
}

func connectedDeliveryIntegrationTableNames() []string {
	return []string{
		constants.TableConnectedPrincipal,
		constants.TableConnectedAuthorizationRequest,
		constants.TableConnectedPrincipalToken,
		constants.TableConnectedDeliveryJob,
		constants.TableConnectedDeliveryPayload,
	}
}

func isConnectedDeliveryIntegrationTable(tableName string) bool {
	for _, connectedTableName := range connectedDeliveryIntegrationTableNames() {
		if strings.EqualFold(tableName, connectedTableName) {
			return true
		}
	}
	return false
}

func recreateConnectedDeliveryPayloadWithPrimary(t *testing.T, ctx context.Context, db *sql.DB, primaryColumn string) {
	t.Helper()
	if _, err := db.ExecContext(ctx, "DROP TABLE `"+constants.TableConnectedDeliveryPayload+"`"); err != nil {
		t.Fatalf("drop connected delivery payload table: %v", err)
	}
	if _, err := db.ExecContext(ctx, "CREATE TABLE `"+constants.TableConnectedDeliveryPayload+"` ("+
		"`JobId` VARCHAR(255) NOT NULL, `PayloadBase64` MEDIUMTEXT NOT NULL, "+
		"`PayloadSha256` CHAR(64) NOT NULL, `EncodedSize` BIGINT NOT NULL, "+
		"`CreatedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), "+
		"`LastModifiedDate` DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), "+
		"PRIMARY KEY (`"+primaryColumn+"`) NONCLUSTERED, "+
		"KEY `idx_connected_delivery_payload_created` (`CreatedDate`, `JobId`)) "+
		"ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"); err != nil {
		t.Fatalf("recreate connected delivery payload with PRIMARY(%s): %v", primaryColumn, err)
	}
}

func assertConnectedDeliveryPrimary(t *testing.T, db *sql.DB, tableName, columns string) {
	t.Helper()
	assertConnectedDeliveryIndex(t, db, tableName, "PRIMARY", 0, columns)
}

func assertConnectedDeliveryIndex(t *testing.T, db *sql.DB, tableName, indexName string, nonUnique int, columns string) {
	t.Helper()
	var actualNonUnique 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`, tableName, indexName).Scan(&actualNonUnique, &actualColumns); err != nil {
		t.Fatalf("query %s index %s: %v", tableName, indexName, err)
	}
	if actualNonUnique != nonUnique || actualColumns != columns {
		t.Fatalf("%s index %s = non_unique %d columns %q, want %d/%q", tableName, indexName, actualNonUnique, actualColumns, nonUnique, columns)
	}
}

func assertConnectedDeliverySchemaDrift(t *testing.T, conn *database.TiDBConnection, want string) {
	t.Helper()
	if err := bootstrapimpl.VerifyConnectedDeliverySchema(conn); err == nil || !strings.Contains(err.Error(), want) {
		t.Fatalf("VerifyConnectedDeliverySchema() after drift = %v, want %q", err, want)
	}
}

func connectedDeliveryForeignKeyConstraintName(t *testing.T, db *sql.DB, tableName, columnName string) string {
	t.Helper()
	var constraintName string
	if err := db.QueryRow(`SELECT CONSTRAINT_NAME FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
		WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ? AND REFERENCED_TABLE_NAME IS NOT NULL`, tableName, columnName).Scan(&constraintName); err != nil {
		t.Fatalf("query %s.%s foreign key constraint: %v", tableName, columnName, err)
	}
	return constraintName
}

func assertConnectedDeliverySchemaReady(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions := make(map[string]schema.TableDefinition)
	for _, def := range bootstrapimpl.GetManagedTableDefinitions() {
		definitions[def.TableName] = def
	}
	for _, tableName := range []string{
		constants.TableConnectedPrincipal,
		constants.TableConnectedAuthorizationRequest,
		constants.TableConnectedPrincipalToken,
		constants.TableConnectedDeliveryJob,
		constants.TableConnectedDeliveryPayload,
	} {
		def, ok := definitions[tableName]
		if !ok {
			t.Fatalf("managed definition for %s is missing", tableName)
		}
		for _, col := range def.Columns {
			var count int
			if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`, tableName, col.Name).Scan(&count); err != nil {
				t.Fatalf("query %s.%s: %v", tableName, col.Name, err)
			}
			if count != 1 {
				t.Fatalf("%s.%s count = %d, want 1", tableName, col.Name, count)
			}
		}
		for _, idx := range def.Indices {
			var nonUnique int
			var columns 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`, tableName, idx.Name).Scan(&nonUnique, &columns); err != nil {
				t.Fatalf("query %s index %s: %v", tableName, idx.Name, err)
			}
			wantNonUnique := 1
			if idx.Unique {
				wantNonUnique = 0
			}
			if nonUnique != wantNonUnique || columns != strings.Join(idx.Columns, ",") {
				t.Fatalf("%s index %s = non_unique %d columns %q, want %d/%q", tableName, idx.Name, nonUnique, columns, wantNonUnique, strings.Join(idx.Columns, ","))
			}
		}
		for _, fk := range def.ForeignKeys {
			referencedTable, referencedColumn, err := parseConnectedDeliveryReference(fk.References)
			if err != nil {
				t.Fatalf("parse %s.%s foreign key: %v", tableName, fk.Column, err)
			}
			var deleteRule string
			if err := db.QueryRow(`
				SELECT rc.DELETE_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 = ?
				  AND k.REFERENCED_COLUMN_NAME = ?`, tableName, fk.Column, referencedTable, referencedColumn).Scan(&deleteRule); err != nil {
				t.Fatalf("query %s.%s foreign key: %v", tableName, fk.Column, err)
			}
			if !strings.EqualFold(deleteRule, fk.OnDelete) {
				t.Fatalf("%s.%s delete rule = %q, want %q", tableName, fk.Column, deleteRule, fk.OnDelete)
			}
		}
	}
}

func parseConnectedDeliveryReference(reference string) (string, string, error) {
	open := strings.LastIndex(reference, "(")
	if open <= 0 || !strings.HasSuffix(reference, ")") {
		return "", "", fmt.Errorf("invalid reference %q", reference)
	}
	return reference[:open], reference[open+1 : len(reference)-1], nil
}