package persistence

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/shared/pkg/constants"
)

// openSchemaDDLTiDB opens a throwaway TiDB database for DDL-primitive
// integration tests. It skips (never fails) when TiDB is unreachable, matching
// the local-oracle discipline: these tests must run against a real TiDB (in
// bootstrap_local_site.sh / CI), because they pin dialect behavior — FK-blocked
// DROP TABLE, KEY_COLUMN_USAGE probing, DROP FOREIGN KEY — that a scripted SQL
// stub cannot reproduce.
func openSchemaDDLTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()

	host := os.Getenv("TIDB_HOST")
	if host == "" {
		host = "127.0.0.1"
	}
	port := os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4001"
	}
	user := os.Getenv("TIDB_USER")
	if user == "" {
		user = "root"
	}
	password := os.Getenv("TIDB_PASSWORD")

	adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=Local", user, password, host, port)
	adminDB, err := sql.Open("mysql", adminDSN)
	if err != nil {
		schemaDDLTestUnavailable(t, "open admin TiDB: %v", err)
	}
	if err := adminDB.Ping(); err != nil {
		_ = adminDB.Close()
		schemaDDLTestUnavailable(t, "TiDB not reachable at %s:%s: %v", host, port, err)
	}

	dbName := fmt.Sprintf("nexuscrm_schema_ddl_%d", time.Now().UnixNano())
	if _, err := adminDB.Exec("CREATE DATABASE `" + dbName + "`"); err != nil {
		_ = adminDB.Close()
		schemaDDLTestUnavailable(t, "create temp database: %v", err)
	}

	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=Local&collation=utf8mb4_unicode_ci", user, password, host, port, dbName)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
		t.Fatalf("open test TiDB connection: %v", err)
	}
	if err := db.Ping(); err != nil {
		_ = db.Close()
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
		t.Fatalf("ping test TiDB connection: %v", err)
	}

	cleanup := func() {
		_ = db.Close()
		_, _ = adminDB.Exec("DROP DATABASE IF EXISTS `" + dbName + "`")
		_ = adminDB.Close()
	}
	return db, cleanup
}

func schemaDDLTestUnavailable(t *testing.T, format string, args ...interface{}) {
	t.Helper()
	if os.Getenv("NEXUSCRM_TIDB_REQUIRED") == "1" {
		t.Fatalf("required TiDB integration environment unavailable: "+format, args...)
	}
	t.Skipf("skipping integration test: "+format, args...)
}

// createSchemaMetadataTables creates the minimal _System_* metadata tables the
// DropTable metadata transaction deletes from, so the DDL primitives run
// end-to-end against a real DB.
func createSchemaMetadataTables(t *testing.T, db *sql.DB) {
	t.Helper()
	stmts := []string{
		"CREATE TABLE `_System_Object` (Id VARCHAR(255) PRIMARY KEY, ApiName VARCHAR(255) NOT NULL) " + constants.DefaultTableOptions,
		"CREATE TABLE `_System_Field` (Id VARCHAR(64) PRIMARY KEY, ObjectId VARCHAR(255) NULL, ApiName VARCHAR(255) NULL) " + constants.DefaultTableOptions,
		// Full column set the BatchRegisterTables upsert writes, so the batch
		// idempotent-rerun test can register through the real code path. The
		// DropTable tests only use Id+TableName (a subset), so this stays
		// compatible.
		"CREATE TABLE `_System_Table` (" +
			"Id VARCHAR(128) PRIMARY KEY, TableName VARCHAR(255) NOT NULL UNIQUE, " +
			"TableType VARCHAR(64) NULL, Category VARCHAR(64) NULL, Description TEXT NULL, " +
			"IsManaged BOOL NULL, SchemaVersion VARCHAR(64) NULL, CreatedBy VARCHAR(64) NULL, " +
			"CreatedDate DATETIME NULL, IsDeleted BOOL NULL, LastModifiedDate DATETIME NULL) " +
			constants.DefaultTableOptions,
		"CREATE TABLE `ObjectPermissions` (Id VARCHAR(64) PRIMARY KEY, SobjectType VARCHAR(255) NULL) " + constants.DefaultTableOptions,
		"CREATE TABLE `FieldPermissions` (Id VARCHAR(64) PRIMARY KEY, SobjectType VARCHAR(255) NULL) " + constants.DefaultTableOptions,
	}
	for _, s := range stmts {
		if _, err := db.Exec(s); err != nil {
			t.Fatalf("create metadata table: %v (%s)", err, s)
		}
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	for _, definition := range definitions {
		if definition.TableName != constants.TableRecordType {
			continue
		}
		if err := NewSchemaRepository(db, nil).CreatePhysicalTable(
			t.Context(),
			definition,
		); err != nil {
			t.Fatalf("create %s from catalog: %v", constants.TableRecordType, err)
		}
		return
	}
	t.Fatalf("%s definition not found", constants.TableRecordType)
}

func TestIntegration_CreatePhysicalTableRestoresForeignKeyChecksOnReusedSession(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)

	var beforeConnectionID int64
	if err := db.QueryRowContext(
		t.Context(),
		"SELECT CONNECTION_ID()",
	).Scan(&beforeConnectionID); err != nil {
		t.Fatalf("read initial connection id: %v", err)
	}

	repo := NewSchemaRepository(db, nil)
	if err := repo.createPhysicalTable(
		t.Context(),
		widgetDef(),
		false,
	); err != nil {
		t.Fatalf("createPhysicalTable: %v", err)
	}

	var (
		afterConnectionID int64
		foreignKeyChecks  int
	)
	if err := db.QueryRowContext(
		t.Context(),
		"SELECT CONNECTION_ID(), @@SESSION.FOREIGN_KEY_CHECKS",
	).Scan(&afterConnectionID, &foreignKeyChecks); err != nil {
		t.Fatalf("read reused DDL session state: %v", err)
	}
	if afterConnectionID != beforeConnectionID {
		t.Fatalf(
			"test did not reuse the pinned DDL session: before = %d, after = %d",
			beforeConnectionID,
			afterConnectionID,
		)
	}
	if foreignKeyChecks != 1 {
		t.Fatalf(
			"reused DDL session FOREIGN_KEY_CHECKS = %d, want 1",
			foreignKeyChecks,
		)
	}
}

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

	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	wanted := map[string]bool{
		"_System_ManagedComponentTargetLock":        false,
		"_System_ManagedComponentOperationEvidence": false,
	}
	repo := NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repo.CreatePhysicalTable(context.Background(), definition); err != nil {
			t.Fatalf("CreatePhysicalTable(%s): %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for tableName, found := range wanted {
		if !found {
			t.Fatalf("%s not found in system table definitions", tableName)
		}
	}

	targets := []struct {
		domain []byte
		key    []byte
	}{
		{domain: []byte("workflow/status"), key: []byte{'A', 0, 'x'}},
		{domain: []byte("workflow/status"), key: []byte{'a', 0, 'x'}},
		{domain: []byte("workflow/status"), key: []byte{'A', 0, 'x', ' '}},
		{domain: []byte("workflow/transition"), key: []byte{'A', 0, 'x'}},
		{domain: []byte{'w', 0, 's'}, key: []byte{0}},
	}
	for _, target := range targets {
		if _, err := db.Exec("INSERT INTO `_System_ManagedComponentTargetLock` (`TargetDomain`, `PhysicalKey`) VALUES (?, ?)", target.domain, target.key); err != nil {
			t.Fatalf("insert binary target %x/%x: %v", target.domain, target.key, err)
		}
	}
	if _, err := db.Exec("INSERT INTO `_System_ManagedComponentTargetLock` (`TargetDomain`, `PhysicalKey`) VALUES (?, ?)", targets[0].domain, targets[0].key); !IsDuplicateKeyError(err) {
		t.Fatalf("exact duplicate binary target error = %v, want duplicate key", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_ManagedComponentTargetLock`", len(targets), "binary target identities must remain distinct")
	zeroByteTarget := targets[len(targets)-1]
	var storedDomain, storedKey []byte
	if err := db.QueryRow("SELECT `TargetDomain`, `PhysicalKey` FROM `_System_ManagedComponentTargetLock` WHERE `TargetDomain` = ? AND `PhysicalKey` = ?", zeroByteTarget.domain, zeroByteTarget.key).Scan(&storedDomain, &storedKey); err != nil {
		t.Fatalf("read zero-byte target %x/%x: %v", zeroByteTarget.domain, zeroByteTarget.key, err)
	}
	if !bytes.Equal(storedDomain, zeroByteTarget.domain) || !bytes.Equal(storedKey, zeroByteTarget.key) {
		t.Fatalf("zero-byte target round trip = %x/%x, want %x/%x", storedDomain, storedKey, zeroByteTarget.domain, zeroByteTarget.key)
	}

	if _, err := db.Exec(`INSERT INTO _System_ManagedComponentOperationEvidence
		(OperationId, Attempt, ComponentKind, Scope, PlanFingerprint, ProjectionIdentityHash, MutationDisposition, SemanticMutationRows, EffectReceiptHash)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		[]byte{'o', 'p', 0, '1'}, 1, "approval_process", "item", make([]byte, 32), make([]byte, 32), "desired_projection_staged", 1, nil,
	); err != nil {
		t.Fatalf("insert operation evidence: %v", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_ManagedComponentOperationEvidence`", 1, "operation evidence must be insertable")
}

// TestIntegration_DropTable_FailsFastOnInboundFK is test-plan (3): with a real
// inbound FK, DropTable("Widget__c") returns a ConflictError naming the
// constraint and leaves the object's metadata rows fully intact (object live).
func TestIntegration_DropTable_FailsFastOnInboundFK(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db)
	repo := NewSchemaRepository(db, nil)

	// Parent Widget__c with a Gadget__c child referencing it (real FK).
	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY)")
	mustExec(t, db, "CREATE TABLE `Gadget__c` (Id VARCHAR(64) PRIMARY KEY, Widget_Ref__c VARCHAR(64) NULL, "+
		"CONSTRAINT `fk_Gadget__c_Widget_Ref__c` FOREIGN KEY (Widget_Ref__c) REFERENCES `Widget__c` (Id))")
	// Register Widget__c metadata so we can prove it stays live.
	mustExec(t, db, "INSERT INTO `_System_Object` (Id, ApiName) VALUES ('obj_widget', 'Widget__c')")
	mustExec(t, db, "INSERT INTO `_System_Field` (Id, ObjectId, ApiName) VALUES ('fld_widget_id', 'obj_widget', 'Id')")
	mustExec(t, db, "INSERT INTO `_System_Table` (Id, TableName) VALUES ('tbl_widget', 'Widget__c')")

	err := repo.dropTable(testWriterContext(t), "Widget__c")
	var conflict *apperrors.ConflictError
	if !errors.As(err, &conflict) {
		t.Fatalf("DropTable error = %T %v, want *ConflictError (inbound FK must fail fast)", err, err)
	}
	if got := conflict.Error(); !containsAll(got, "fk_Gadget__c_Widget_Ref__c") {
		t.Fatalf("ConflictError = %q, want it to name the referencing constraint", got)
	}
	// Metadata must be fully intact — object still live.
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName='Widget__c'", 1, "object row must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Table` WHERE TableName='Widget__c'", 1, "table registry row must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Field` WHERE ObjectId='obj_widget'", 1, "field rows must survive")
}

func TestIntegration_DropTable_RejectsCatalogOwnedRecordTypeCascade(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db)
	repo := NewSchemaRepository(db, nil)

	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY)")
	mustExec(t, db, "INSERT INTO `_System_Object` (Id, ApiName) VALUES ('obj_widget', 'Widget__c')")
	mustExec(t, db, "INSERT INTO `_System_Field` (Id, ObjectId, ApiName) VALUES ('fld_widget_id', 'obj_widget', 'Id')")
	mustExec(t, db, "INSERT INTO `_System_Table` (Id, TableName) VALUES ('tbl_widget', 'Widget__c')")
	mustExec(t, db, `INSERT INTO _System_RecordType
		(Id, ObjectId, Name, SobjectType, CatalogNamespaceId, CatalogLocalKey)
		VALUES ('rt_widget', 'obj_widget', 'Widget Type', 'Widget__c', ?, ?)`,
		[]byte("official:crm"),
		[]byte("widget-type"),
	)

	err := repo.dropTable(testWriterContext(t), "Widget__c")
	var validation *apperrors.ValidationError
	if !errors.As(err, &validation) {
		t.Fatalf("DropTable error = %T %v, want ValidationError", err, err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='Widget__c'", 1, "physical table must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE Id='obj_widget'", 1, "object metadata must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Field` WHERE ObjectId='obj_widget'", 1, "field metadata must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Table` WHERE TableName='Widget__c'", 1, "table registry must survive")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_RecordType` WHERE Id='rt_widget'", 1, "catalog-owned RecordType must survive")
}

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

	mustExec(t, db, "INSERT INTO `_System_Object` (Id, ApiName) VALUES ('obj_widget', 'Widget__c')")
	mustExec(t, db, `INSERT INTO _System_RecordType
		(Id, ObjectId, Name, SobjectType)
		VALUES ('rt_widget', 'obj_widget', 'Widget Type', 'Widget__c')`)

	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
	defer cancel()
	expectBlockedMutation := func(name string, mutate func(*sql.Tx) error) {
		t.Helper()
		mutationTx, mutationCleanup, err := BeginPessimisticTx(
			withLockWaitTimeoutSecs(ctx, 1),
			db,
		)
		if err != nil {
			t.Fatalf("begin %s transaction: %v", name, err)
		}
		mutationErr := mutate(mutationTx)
		if err := mutationTx.Rollback(); err != nil {
			t.Fatalf("rollback %s transaction: %v", name, err)
		}
		if err := mutationCleanup(); err != nil {
			t.Fatalf("clean up %s transaction: %v", name, err)
		}
		if !IsLockContention(mutationErr) {
			t.Fatalf("%s error = %v, want lock timeout", name, mutationErr)
		}
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	if err := transactions.WithPessimisticTransaction(
		ctx,
		func(txCtx context.Context) error {
			if err := preflightManagedRowMutationCascades(
				txCtx,
				transactions,
				managedRowCascadeDeleteTarget{
					tableName:  constants.TableObject,
					parentKeys: []string{"obj_widget"},
				},
			); err != nil {
				return err
			}
			expectBlockedMutation("concurrent authority assignment", func(tx *sql.Tx) error {
				_, err := tx.ExecContext(ctx, `UPDATE _System_RecordType
					SET CatalogNamespaceId = ?, CatalogLocalKey = ?
					WHERE Id = ?`,
					[]byte("official:crm"),
					[]byte("widget-type"),
					"rt_widget",
				)
				return err
			})
			expectBlockedMutation("concurrent child insert", func(tx *sql.Tx) error {
				_, err := tx.ExecContext(ctx, `INSERT INTO _System_RecordType
					(Id, ObjectId, Name, SobjectType)
					VALUES ('rt_widget_second', 'obj_widget', 'Second Widget Type', 'Widget__c')`)
				return err
			})
			return nil
		},
	); err != nil {
		t.Fatalf("preflightManagedRowMutationCascades(): %v", err)
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_RecordType
		SET CatalogNamespaceId = ?, CatalogLocalKey = ?
		WHERE Id = ?`,
		[]byte("official:crm"),
		[]byte("widget-type"),
		"rt_widget",
	); err != nil {
		t.Fatalf("authority assignment after cascade lock release: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_RecordType
		(Id, ObjectId, Name, SobjectType)
		VALUES ('rt_widget_second', 'obj_widget', 'Second Widget Type', 'Widget__c')`); err != nil {
		t.Fatalf("child insert after cascade lock release: %v", err)
	}
	assertCount(t, db, `SELECT COUNT(*) FROM _System_RecordType
		WHERE Id='rt_widget'
		  AND CatalogNamespaceId IS NOT NULL
		  AND CatalogLocalKey IS NOT NULL`, 1, "authority assignment must complete after lock release")
	assertCount(t, db, `SELECT COUNT(*) FROM _System_RecordType
		WHERE ObjectId='obj_widget'`, 2, "child insert must complete after parent lock release")
}

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

	mustExec(t, db, "INSERT INTO `_System_Object` (Id, ApiName) VALUES ('obj_widget', 'Widget__c')")
	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 = ?
  AND REFERENCED_COLUMN_NAME = ?`,
		constants.TableRecordType,
		constants.FieldObjectID,
		constants.TableObject,
		constants.FieldID,
	).Scan(&constraintName); err != nil {
		t.Fatalf("resolve RecordType object foreign key: %v", err)
	}
	if !isManagedRowCascadeIdentifier(constraintName) {
		t.Fatalf("invalid live constraint name %q", constraintName)
	}
	mustExec(
		t,
		db,
		fmt.Sprintf(
			"ALTER TABLE %s DROP FOREIGN KEY %s",
			quoteManagedRowCascadeIdentifier(constants.TableRecordType),
			quoteManagedRowCascadeIdentifier(constraintName),
		),
	)

	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	err := transactions.WithPessimisticTransaction(
		t.Context(),
		func(txCtx context.Context) error {
			return preflightManagedRowMutationCascades(
				txCtx,
				transactions,
				managedRowCascadeDeleteTarget{
					tableName:  constants.TableObject,
					parentKeys: []string{"obj_widget"},
				},
			)
		},
	)
	if err == nil || !strings.Contains(err.Error(), "ON DELETE CASCADE") {
		t.Fatalf("missing live cascade error = %v", err)
	}
}

// TestIntegration_DropColumnFK_ThenDropTable_RecoveryLoop is test-plan (5): the
// full recovery loop. Dropping the FK-bearing lookup column succeeds (the
// constraint is removed first), and then DropTable("Widget__c") succeeds because
// the inbound FK is gone.
func TestIntegration_DropColumnFK_ThenDropTable_RecoveryLoop(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db)
	repo := NewSchemaRepository(db, nil)

	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY)")
	mustExec(t, db, "CREATE TABLE `Gadget__c` (Id VARCHAR(64) PRIMARY KEY, Widget_Ref__c VARCHAR(64) NULL, "+
		"CONSTRAINT `fk_Gadget__c_Widget_Ref__c` FOREIGN KEY (Widget_Ref__c) REFERENCES `Widget__c` (Id))")
	mustExec(t, db, "INSERT INTO `_System_Object` (Id, ApiName) VALUES ('obj_widget', 'Widget__c')")
	mustExec(t, db, "INSERT INTO `_System_Table` (Id, TableName) VALUES ('tbl_widget', 'Widget__c')")

	// Drop the FK-bearing lookup column: the helper drops the constraint first,
	// so this must succeed (base code would 1828-dead-end here).
	if err := repo.dropPhysicalColumnIfPresent("Gadget__c", "Widget_Ref__c"); err != nil {
		t.Fatalf("dropPhysicalColumnIfPresent on FK column returned error: %v", err)
	}
	assertNoInboundFK(t, db, "Widget__c")

	// With the inbound FK gone, DropTable now succeeds.
	if err := repo.dropTable(testWriterContext(t), "Widget__c"); err != nil {
		t.Fatalf("DropTable after FK removal returned error: %v", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='Widget__c'", 0, "Widget__c table must be dropped")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName='Widget__c'", 0, "object metadata must be gone")
}

// TestIntegration_StrictCreate_FailsClosedOnLingeringTable is test-plan (1) at
// the DDL layer: a pre-existing physical table with a data row and NO metadata
// makes the strict create fail closed (ConflictError). The old row survives and
// no metadata is bound to it — the create-rebind hole is closed.
func TestIntegration_StrictCreate_FailsClosedOnLingeringTable(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db)
	repo := NewSchemaRepository(db, nil)

	// Orphan physical table with a surviving row, no metadata.
	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY, legacy_marker VARCHAR(64) NULL)")
	mustExec(t, db, "INSERT INTO `Widget__c` (Id, legacy_marker) VALUES ('old-1', 'survivor')")

	def := schema.TableDefinition{
		TableName: "Widget__c",
		Columns:   []schema.ColumnDefinition{{Name: "Id", Type: "string"}},
	}
	err := repo.CreateTableWithStrictMetadata(context.Background(), def, widgetMeta())
	var conflict *apperrors.ConflictError
	if !errors.As(err, &conflict) {
		t.Fatalf("strict create error = %T %v, want *ConflictError (must not adopt the orphan)", err, err)
	}
	// Old row survives; no metadata bound.
	assertCount(t, db, "SELECT COUNT(*) FROM `Widget__c` WHERE Id='old-1'", 1, "orphan data row must survive an un-adopted create")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName='Widget__c'", 0, "no object metadata may be bound to the orphan")
}

// TestIntegration_FailedDropThenRecreate_StateMachine is the gate-demanded
// end-to-end regression (test-plan 7) at the chokepoint where both operations
// share a real DB. It manufactures the EXACT committed post-state of a phase-2
// DROP failure — metadata rows removed (byte-identical to what DropTable's
// phase-1 tx commits), the physical table and its rows still present — then
// walks the convergence loop:
//
//  1. strict CreateTableWithStrictMetadata("Widget__c") must FAIL CLOSED
//     (ConflictError), and the old row must NOT be resurrected / rebound.
//  2. DropTable("Widget__c") reconciles the orphan (table dropped).
//  3. strict create then SUCCEEDS and the fresh object has ZERO old rows.
//
// Without the strict plain-CREATE, step 1 silently adopts the orphan (rebind
// hole) and the old row survives under new metadata — the whole bug.
func TestIntegration_FailedDropThenRecreate_StateMachine(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db)
	repo := NewSchemaRepository(db, nil)
	ctx := context.Background()

	def := schema.TableDefinition{
		TableName: "Widget__c",
		Columns:   []schema.ColumnDefinition{{Name: "Id", Type: "string"}},
	}

	// Manufacture the committed post-failed-drop state directly (the harness's
	// minimal _System_* tables cannot round-trip a full strict-metadata insert,
	// so we build the exact committed state by hand — metadata ABSENT, physical
	// table + a survivor row PRESENT, and the DROP-PENDING tombstone row
	// (IsDeleted=1) that markDropPendingTombstone writes on a failed drop).
	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY, legacy_marker VARCHAR(64) NULL)")
	mustExec(t, db, "INSERT INTO `Widget__c` (Id, legacy_marker) VALUES ('old-row-1', 'survivor')")
	mustExec(t, db, "INSERT INTO `_System_Table` (Id, TableName, IsDeleted) VALUES ('tbl_widget', 'Widget__c', 1)")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName='Widget__c'", 0, "precondition: no metadata for the orphan")

	// Step 1: strict create must FAIL CLOSED at the plain CREATE (MySQL 1050)
	// BEFORE any metadata insert — no rebind, no resurrection.
	err := repo.CreateTableWithStrictMetadata(ctx, def, widgetMeta())
	var conflict *apperrors.ConflictError
	if !errors.As(err, &conflict) {
		t.Fatalf("recreate over orphan: error = %T %v, want *ConflictError (must not rebind)", err, err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Object` WHERE ApiName='Widget__c'", 0, "no metadata may be bound to the orphan")
	assertCount(t, db, "SELECT COUNT(*) FROM `Widget__c` WHERE Id='old-row-1'", 1, "orphan row must NOT be resurrected under new metadata")

	// Step 2: DropTable reconciles the orphan (no metadata to unregister; the
	// drop-pending tombstone is cleared inside the metadata transaction; DROP
	// TABLE IF EXISTS clears the physical table).
	if err := repo.dropTable(testWriterContext(t), "Widget__c"); err != nil {
		t.Fatalf("reconcile DropTable of orphan: %v", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='Widget__c'", 0, "orphan table must be dropped by reconcile")
	assertCount(t, db, "SELECT COUNT(*) FROM `_System_Table` WHERE TableName='Widget__c'", 0, "drop-pending tombstone must be cleared by the successful reconcile")

	// Step 3: the strict CREATE now succeeds against the clean name (the metadata
	// insert needs the full production _System_Object shape, out of this harness's
	// scope, so we assert the physical create — the rebind-relevant step — starts
	// from ZERO rows). This proves the name is creatable again and carries no old
	// data.
	if err := repo.createPhysicalTable(ctx, def, false); err != nil {
		t.Fatalf("strict physical create after reconcile: %v", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM `Widget__c`", 0, "recreated table must have ZERO old rows")
}

// TestIntegration_CreatePhysicalTable_And_Batch_IdempotentRerun is test-plan (6):
// the EXPORTED create paths (CreatePhysicalTable, BatchCreatePhysicalTables) must
// stay idempotent — re-running them against tables that already exist returns nil,
// exactly what bootstrap / pack-install / tenant-overlay resume relies on. This is
// the CRITICAL EDGE the strict-create change must not regress: only the strict
// interactive path fails closed on a pre-existing table; the shared exported paths
// keep CREATE TABLE IF NOT EXISTS.
func TestIntegration_CreatePhysicalTable_And_Batch_IdempotentRerun(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	createSchemaMetadataTables(t, db) // BatchCreatePhysicalTables also registers into _System_Table.
	repo := NewSchemaRepository(db, nil)

	def := schema.TableDefinition{
		TableName: "Widget__c",
		Columns:   []schema.ColumnDefinition{{Name: "Id", Type: "string"}},
	}

	// First create.
	if err := repo.CreatePhysicalTable(context.Background(), def); err != nil {
		t.Fatalf("first CreatePhysicalTable: %v", err)
	}
	// Second create against the now-existing table MUST be a no-op, not a 1050.
	if err := repo.CreatePhysicalTable(context.Background(), def); err != nil {
		t.Fatalf("idempotent re-run of CreatePhysicalTable returned error (bootstrap safety regressed): %v", err)
	}

	// Batch path re-run against the same existing table must also succeed.
	if err := repo.BatchCreatePhysicalTables(context.Background(), []schema.TableDefinition{def}); err != nil {
		t.Fatalf("BatchCreatePhysicalTables re-run against existing table returned error: %v", err)
	}
	assertCount(t, db, "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='Widget__c'", 1, "table must exist exactly once after idempotent re-runs")
}

// TestIntegration_TableHasPlatformSystemColumns pins the D10 physical-provenance
// probe against real TiDB: a platform-shaped table (Id/CreatedDate/IsDeleted)
// TestIntegration_HasDropPendingTombstone pins the drop-pending tombstone probe
// the DeleteSchema orphan-reconcile authorizes destructive drops with: ONLY a
// _System_Table row with IsDeleted=1 (written exclusively by a failed phase-2
// DROP) reports true — case-insensitively, mirroring the CI collation the real
// _System_Table uses. A LIVE registration (IsDeleted=0), an unregistered raw
// operator table, and a same-shape CREATE TABLE ... LIKE copy (which defeated
// the earlier system-column shape heuristic) all report false and can never be
// dropped by an absent-metadata DELETE.
func TestIntegration_HasDropPendingTombstone(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	repo := NewSchemaRepository(db, nil)

	// Minimal _System_Table registry mirroring production collation (CI).
	mustExec(t, db, "CREATE TABLE `_System_Table` (TableName VARCHAR(255) PRIMARY KEY, IsDeleted TINYINT(1) NOT NULL DEFAULT 0) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
	// Orphan whose physical DROP failed: drop-pending tombstone.
	mustExec(t, db, "INSERT INTO `_System_Table` (TableName, IsDeleted) VALUES ('Widget__c', 1)")
	// Live registered table: row present but NOT drop-pending.
	mustExec(t, db, "INSERT INTO `_System_Table` (TableName, IsDeleted) VALUES ('Gadget__c', 0)")
	// Same-shape manual copy / raw table: no row at all.
	mustExec(t, db, "CREATE TABLE `Widget__c` (Id VARCHAR(64) PRIMARY KEY, CreatedDate DATETIME NULL, IsDeleted TINYINT(1) NOT NULL DEFAULT 0)")
	mustExec(t, db, "CREATE TABLE `Widget_backup` LIKE `Widget__c`")
	mustExec(t, db, "CREATE TABLE `operator_backup_2026` (pk BIGINT PRIMARY KEY, payload LONGTEXT NULL)")

	cases := []struct {
		name  string
		table string
		want  bool
	}{
		{"drop-pending tombstone", "Widget__c", true},
		{"drop-pending case-insensitive", "widget__c", true},
		{"live registration is NOT drop-pending", "Gadget__c", false},
		{"same-shape LIKE copy, no row", "Widget_backup", false},
		{"raw non-platform, no row", "operator_backup_2026", false},
		{"missing entirely", "does_not_exist_at_all", false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got, err := repo.HasDropPendingTombstone(tc.table)
			if err != nil {
				t.Fatalf("HasDropPendingTombstone(%q): %v", tc.table, err)
			}
			if got != tc.want {
				t.Fatalf("HasDropPendingTombstone(%q) = %v, want %v", tc.table, got, tc.want)
			}
		})
	}
}

func mustExec(t *testing.T, db *sql.DB, q string, args ...interface{}) {
	t.Helper()
	if _, err := db.Exec(q, args...); err != nil {
		t.Fatalf("exec %q: %v", q, err)
	}
}

func assertCount(t *testing.T, db *sql.DB, q string, want int, msg string) {
	t.Helper()
	var got int
	if err := db.QueryRow(q).Scan(&got); err != nil {
		t.Fatalf("count query %q: %v", q, err)
	}
	if got != want {
		t.Fatalf("%s: count=%d, want %d (%q)", msg, got, want, q)
	}
}

func assertNoInboundFK(t *testing.T, db *sql.DB, table string) {
	t.Helper()
	assertCount(t, db,
		fmt.Sprintf("SELECT COUNT(*) FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA=DATABASE() AND REFERENCED_TABLE_NAME='%s'", table),
		0, "inbound FK must be removed")
}