package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// CreatePhysicalTable creates the table structure without registering metadata.
//
// It is idempotent for legitimate re-runs (CREATE TABLE IF NOT EXISTS):
// bootstrap, pack install, tenant-overlay apply, and batch resume all re-run it
// against DBs where the target table legitimately already exists — a table a
// prior run of the SAME lane created — and must keep succeeding. The strict
// interactive-authoring path does NOT use this variant — it calls
// createPhysicalTable(ctx, def, false) so a pre-existing table fails closed
// with MySQL 1050 instead of adopting an orphan (#3484).
//
// One pre-existing table is NOT adoptable: a confirmed platform orphan awaiting
// deletion — a table whose name carries the drop-pending tombstone written by a
// failed DropTable phase 2. Adopting it would bind new pack/bootstrap metadata
// to the deleted object's stale table and resurrect its old rows. The guard
// below finishes the interrupted drop first (the tombstone is the same
// authorization the DeleteSchema reconcile uses) and then creates fresh, so
// the install/apply lanes can no longer adopt a delete's residue. A markerless
// pre-existing table (a legitimate resume, or a foreign table) is untouched —
// resume adoption stays byte-identical and foreign tables are never dropped.
func (r *SchemaRepository) CreatePhysicalTable(ctx context.Context, def schema.TableDefinition) error {
	if err := validateCatalogTableWriteProtectionProjection(def); err != nil {
		return err
	}
	if err := validateCatalogObjectAPIName(def.TableName); err != nil {
		return apperrors.NewValidationError("tableName", err.Error())
	}
	dropPending, err := r.HasDropPendingTombstone(def.TableName)
	if err != nil {
		if IsTableNotExistsError(err) {
			// Greenfield bootstrap: _System_Table itself is not provisioned yet,
			// so no drop-pending tombstone can exist — proceed to create.
			dropPending = false
		} else {
			// Fail closed (retryable): an unreadable tombstone state must not
			// silently degrade into adopting a possible orphan. Bootstrap/pack
			// lanes are resume-safe by contract, so surfacing the error is safe.
			return apperrors.NewInternalError(fmt.Sprintf("failed to probe drop-pending tombstone for %s", def.TableName), err)
		}
	}
	if dropPending {
		r.logger.Warn(fmt.Sprintf("table %s carries a drop-pending tombstone (interrupted delete); finishing the drop before re-create so the new object cannot adopt the deleted object's table", def.TableName))
		if dropErr := r.dropTable(ctx, def.TableName); dropErr != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to finish the interrupted drop of %s before re-create", def.TableName), dropErr)
		}
	}
	if err := r.createPhysicalTable(ctx, def, true); err != nil {
		return err
	}
	return r.reconcileManagedPhysicalSchemaEvolution(ctx, def)
}

// createPhysicalTable builds and executes the CREATE TABLE for def. When
// ifNotExists is true it emits `CREATE TABLE IF NOT EXISTS` (idempotent re-run
// semantics for bootstrap/pack/batch). When false it emits a plain
// `CREATE TABLE`, so a pre-existing physical table of the same name makes the
// statement fail with MySQL 1050 — the atomic existence check the strict
// object-create path relies on. This is the single chokepoint; there is no
// separate INFORMATION_SCHEMA probe, so no TOCTOU window and no second read path
// for the "table exists" fact.
func (r *SchemaRepository) createPhysicalTable(ctx context.Context, def schema.TableDefinition, ifNotExists bool) error {
	r.logger.Info(fmt.Sprintf("creating table: %s", def.TableName))
	if err := validateCatalogTableWriteProtectionProjection(def); err != nil {
		return err
	}

	if err := validateCatalogObjectAPIName(def.TableName); err != nil {
		return apperrors.NewValidationError("tableName", err.Error())
	}

	// VALIDATION: Table Name
	// System tables (starting with _System_) are exempt from strict snake_case
	if !constants.IsSystemTable(def.TableName) {
		validName := regexp.MustCompile(constants.TableNamePattern)
		if !validName.MatchString(def.TableName) {
			return apperrors.NewValidationError("tableName", fmt.Sprintf("table name '%s' must be alphanumeric (PascalCase/snake_case)", def.TableName))
		}
	}

	physicalForeignKeys := physicalForeignKeyDefinitions(def.ForeignKeys)
	logicalForeignKeys, err := parseLogicalForeignKeyDefinitions(def.ForeignKeys)
	if err != nil {
		return apperrors.NewValidationError("foreignKeys", err.Error())
	}

	// Build CREATE TABLE statement with indexes inline
	createStmt := "CREATE TABLE IF NOT EXISTS"
	if !ifNotExists {
		createStmt = "CREATE TABLE"
	}
	var ddl strings.Builder
	ddl.WriteString(fmt.Sprintf("%s `%s` (\n", createStmt, def.TableName))

	// Add columns
	for i, col := range def.Columns {
		// VALIDATION: Fail fast if schema assumptions are violated
		if err := r.ValidateFieldDefinition(col); err != nil {
			return apperrors.NewValidationError(col.Name, fmt.Sprintf("invalid column definition for '%s': %v", col.Name, err))
		}

		ddl.WriteString("  ")

		// Physical names now perfectly match Salesforce logical names (e.g. Id)

		ddl.WriteString(r.buildColumnDDL(col, true))
		// Always add comma if there are more columns, indexes, or foreign keys
		if i < len(def.Columns)-1 || len(def.Indices) > 0 || len(physicalForeignKeys) > 0 {
			ddl.WriteString(",")
		}
		ddl.WriteString("\n")
	}

	// Add indexes inline (KEY or UNIQUE KEY)
	for i, idx := range def.Indices {
		ddl.WriteString("  ")
		ddl.WriteString(r.buildIndexDDL(def.TableName, idx))
		if i < len(def.Indices)-1 || len(physicalForeignKeys) > 0 {
			ddl.WriteString(",")
		}
		ddl.WriteString("\n")
	}

	// Add foreign keys. ForeignKeyDefinition.OnDelete is emitted only as a
	// physical DDL referential action; generic record soft-delete does not
	// execute database FK actions.
	for i, fk := range physicalForeignKeys {
		ddl.WriteString("  ")
		ddl.WriteString(r.buildForeignKeyDDL(fk))
		if i < len(physicalForeignKeys)-1 {
			ddl.WriteString(",")
		}
		ddl.WriteString("\n")
	}
	ddl.WriteString(fmt.Sprintf(") %s", constants.DefaultTableOptions))

	// Execute DDL using a dedicated connection because FOREIGN_KEY_CHECKS is
	// session-scoped. BatchCreatePhysicalTables creates definitions in parallel,
	// so a child table may be created before its referenced parent.

	conn, err := r.db.Conn(ctx)
	if err != nil {
		return apperrors.NewInternalError("failed to get database connection", err)
	}
	defer func() { _ = conn.Close() }()

	// Disable foreign key checks only on the pinned DDL connection. A failed SET
	// is ambiguous from the caller's perspective, so discard that connection
	// rather than letting a possibly modified session return to the pool.
	if _, err := conn.ExecContext(ctx, "SET FOREIGN_KEY_CHECKS=0"); err != nil {
		markConnBad(conn)
		return apperrors.NewInternalError(
			fmt.Sprintf("failed to disable foreign key checks for table %s", def.TableName),
			err,
		)
	}
	// The CREATE is durable DDL. Restore the session before Close returns the
	// physical connection to the pool; if restoration fails, quarantine the
	// connection instead of turning a successful CREATE into an orphan-producing
	// retry signal.
	defer func() {
		restoreCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if _, err := conn.ExecContext(restoreCtx, "SET FOREIGN_KEY_CHECKS=1"); err != nil {
			markConnBad(conn)
			r.logger.Error(
				fmt.Sprintf(
					"failed to restore foreign key checks after table DDL attempt for %s; discarded DDL connection",
					def.TableName,
				),
				err,
			)
		}
	}()

	// Normalize FK columns (both source and referenced) to prevent Error 3780.
	// Uses per-column MODIFY instead of table-level CONVERT TO CHARACTER SET
	// because TiDB rejects CONVERT on indexed columns (Error 8200).
	if len(physicalForeignKeys) > 0 {
		physicalDef := def
		physicalDef.ForeignKeys = physicalForeignKeys
		if err := r.normalizeTableForFK(ctx, conn, physicalDef); err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf("failed to normalize foreign-key columns for table %s", def.TableName),
				err,
			)
		}
	}

	r.logger.Debug(fmt.Sprintf("executing DDL for %s: \n%s", def.TableName, ddl.String()))
	if _, err := conn.ExecContext(ctx, ddl.String()); err != nil {
		// Strict path (ifNotExists=false): a 1050 means a physical table of this
		// name already exists that this call did NOT just create — fail closed so
		// new metadata can never be bound to a pre-existing orphan table. Surface
		// it as a ConflictError carrying the reconcile remediation.
		if !ifNotExists && IsTableExistsError(err) {
			r.logger.Warn(fmt.Sprintf("strict create blocked: physical table %s already exists with no live metadata", def.TableName))
			return apperrors.NewConflictErrorWithMessage(def.TableName, fmt.Sprintf(
				"object name %q is blocked by an existing physical table with no live metadata (likely an orphan left by an incomplete delete); retry DELETE for that object name to reconcile the orphan, then re-create; if that DELETE reports not-found, the table is not platform-owned or cannot be safely reconciled, and an operator must rename or drop it manually",
				def.TableName))
		}
		r.logger.Error(fmt.Sprintf("failed to create table %s", def.TableName), err)
		return apperrors.NewInternalError(fmt.Sprintf("failed to create table %s", def.TableName), err)
	}
	r.logger.Info(fmt.Sprintf("DDL executed successfully for %s", def.TableName))

	// Logical-only references are enforced by the request-bound runtime
	// authority, not by a physical parent table. CREATE omits them for new
	// databases; this convergence step removes constraints left by an older
	// schema definition when CREATE IF NOT EXISTS adopts an existing table.
	// Keeping both directions here makes the declarative schema definition the
	// sole owner for greenfield and warm upgrades and avoids one-off migrations
	// for every relationship that moves behind a Catalog authority.
	// A strict create cannot have adopted old constraints; probing after it would
	// only add a post-CREATE failure mode that turns its retry into a 1050
	// conflict. Warm convergence belongs exclusively to the idempotent ensure
	// lane.
	if ifNotExists {
		if err := r.dropExistingLogicalForeignKeys(ctx, conn, def.TableName, logicalForeignKeys); err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf("failed to converge logical-only foreign keys for table %s", def.TableName),
				err,
			)
		}
	}

	return nil
}

type logicalForeignKeyDefinition struct {
	column           string
	referencedTable  string
	referencedColumn string
}

func parseLogicalForeignKeyDefinitions(definitions []schema.ForeignKeyDefinition) ([]logicalForeignKeyDefinition, error) {
	identifierPattern := regexp.MustCompile(constants.FieldNamePattern)
	referencePattern := regexp.MustCompile(`^\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\(\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\)\s*$`)
	parsed := make([]logicalForeignKeyDefinition, 0, len(definitions))
	for _, definition := range definitions {
		if !definition.LogicalOnly {
			continue
		}
		column := strings.TrimSpace(definition.Column)
		matches := referencePattern.FindStringSubmatch(definition.References)
		if !identifierPattern.MatchString(column) || len(matches) != 3 {
			return nil, fmt.Errorf(
				"logical-only foreign key %q -> %q must use valid identifiers and the form ParentTable(ParentColumn)",
				definition.Column,
				definition.References,
			)
		}
		parsed = append(parsed, logicalForeignKeyDefinition{
			column:           column,
			referencedTable:  matches[1],
			referencedColumn: matches[2],
		})
	}
	return parsed, nil
}

func (r *SchemaRepository) dropExistingLogicalForeignKeys(
	ctx context.Context,
	conn *sql.Conn,
	tableName string,
	definitions []logicalForeignKeyDefinition,
) error {
	const lookup = `
		SELECT CONSTRAINT_NAME
		FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
		WHERE TABLE_SCHEMA = DATABASE()
		  AND REFERENCED_TABLE_NAME IS NOT NULL
		  AND LOWER(TABLE_NAME) = LOWER(?)
		  AND LOWER(COLUMN_NAME) = LOWER(?)
		  AND LOWER(REFERENCED_TABLE_NAME) = LOWER(?)
		  AND LOWER(REFERENCED_COLUMN_NAME) = LOWER(?)
		ORDER BY CONSTRAINT_NAME`

	for _, definition := range definitions {
		rows, err := conn.QueryContext(
			ctx,
			lookup,
			tableName,
			definition.column,
			definition.referencedTable,
			definition.referencedColumn,
		)
		if err != nil {
			return fmt.Errorf("look up %s.%s logical-only constraint: %w", tableName, definition.column, err)
		}
		var constraintNames []string
		for rows.Next() {
			var constraintName string
			if err := rows.Scan(&constraintName); err != nil {
				_ = rows.Close()
				return fmt.Errorf("scan %s.%s logical-only constraint: %w", tableName, definition.column, err)
			}
			if !regexp.MustCompile(constants.FieldNamePattern).MatchString(constraintName) {
				_ = rows.Close()
				return fmt.Errorf("logical-only constraint on %s.%s has unsafe name %q", tableName, definition.column, constraintName)
			}
			constraintNames = append(constraintNames, constraintName)
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return fmt.Errorf("iterate %s.%s logical-only constraints: %w", tableName, definition.column, err)
		}
		if err := rows.Close(); err != nil {
			return fmt.Errorf("close %s.%s logical-only constraint rows: %w", tableName, definition.column, err)
		}

		for _, constraintName := range constraintNames {
			statement := fmt.Sprintf("ALTER TABLE `%s` DROP FOREIGN KEY `%s`", tableName, constraintName)
			if _, err := conn.ExecContext(ctx, statement); err != nil {
				return fmt.Errorf("drop logical-only constraint %s on %s.%s: %w", constraintName, tableName, definition.column, err)
			}
			r.logger.Info(fmt.Sprintf("dropped obsolete physical foreign key %s for logical-only relationship %s.%s", constraintName, tableName, definition.column))
		}
	}
	return nil
}

func physicalForeignKeyDefinitions(definitions []schema.ForeignKeyDefinition) []schema.ForeignKeyDefinition {
	if len(definitions) == 0 {
		return nil
	}
	physical := make([]schema.ForeignKeyDefinition, 0, len(definitions))
	for _, definition := range definitions {
		if definition.LogicalOnly {
			continue
		}
		physical = append(physical, definition)
	}
	return physical
}

// normalizeTableForFK ensures an existing table's string columns use the FK
// source column's catalog-declared collation and the legacy VARCHAR(255) width.
// A source column with no explicit collation retains the historical
// utf8mb4_unicode_ci behavior.
//
// TiDB does NOT support `ALTER TABLE CONVERT TO CHARACTER SET` when indexed columns
// are involved (Error 8200). Instead, we normalize per-column via MODIFY COLUMN,
// which works even on PRIMARY KEY columns.
//
// This function normalizes:
//  1. The table being created (def.TableName) — its FK source columns
//  2. Referenced tables — their target columns (typically Id)
//
// Both sides use the source column's declaration because that is the exact
// child DDL this create call will emit. Rewriting every participant to the
// database default would silently erase byte-exact catalog identities.
func (r *SchemaRepository) normalizeTableForFK(ctx context.Context, conn *sql.Conn, def schema.TableDefinition) error {
	// Collect all (table, column) pairs that participate in FKs
	type colRef struct{ table, column, collation string }
	var targets []colRef

	columns := make(map[string]schema.ColumnDefinition, len(def.Columns))
	for _, column := range def.Columns {
		columns[column.Name] = column
	}

	for _, fk := range def.ForeignKeys {
		sourceColumn, ok := columns[fk.Column]
		if !ok {
			return fmt.Errorf("foreign-key source column %s.%s is absent from its table definition", def.TableName, fk.Column)
		}
		targetCollation := strings.TrimSpace(sourceColumn.CollationName)
		if targetCollation == "" {
			targetCollation = "utf8mb4_unicode_ci"
		}

		// Source column: tableName.fk.Column
		targets = append(targets, colRef{def.TableName, fk.Column, targetCollation})

		// Referenced column: parse "RefTable(RefCol)"
		refTable := fk.References
		refColumn := ""
		if idx := strings.Index(refTable, "("); idx != -1 {
			end := strings.Index(refTable, ")")
			if end > idx {
				refColumn = strings.TrimSpace(refTable[idx+1 : end])
			}
			refTable = strings.TrimSpace(refTable[:idx])
		}
		if refColumn != "" {
			targets = append(targets, colRef{refTable, refColumn, targetCollation})
		}
	}

	for _, t := range targets {
		var colType, collation, isNullable, colKey sql.NullString
		err := conn.QueryRowContext(ctx,
			`SELECT COLUMN_TYPE, COLLATION_NAME, IS_NULLABLE, COLUMN_KEY
			 FROM INFORMATION_SCHEMA.COLUMNS
			 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
			t.table, t.column).Scan(&colType, &collation, &isNullable, &colKey)
		if errors.Is(err, sql.ErrNoRows) {
			// Table or column doesn't exist yet — will be created in this batch
			continue
		}
		if err != nil {
			return fmt.Errorf("inspect foreign-key column %s.%s: %w", t.table, t.column, err)
		}

		upper := strings.ToUpper(colType.String)
		needsFix := false
		targetType := "VARCHAR(255)"

		// Check collation mismatch
		if collation.Valid && !strings.EqualFold(collation.String, t.collation) {
			needsFix = true
		}
		// Check width mismatch (e.g. VARCHAR(18) vs VARCHAR(255))
		if strings.HasPrefix(upper, "VARCHAR") && upper != "VARCHAR(255)" {
			needsFix = true
		}

		if !needsFix {
			continue
		}

		// Build MODIFY COLUMN with correct nullability
		nullClause := "NOT NULL"
		if isNullable.Valid && isNullable.String == "YES" {
			nullClause = "DEFAULT NULL"
		}

		r.logger.Info(fmt.Sprintf("normalizing %s.%s (%s %s -> VARCHAR(255) %s)",
			t.table, t.column, colType.String, collation.String, t.collation))

		modifySQL := fmt.Sprintf(
			"ALTER TABLE `%s` MODIFY COLUMN `%s` %s CHARACTER SET utf8mb4 COLLATE %s %s",
			t.table, t.column, targetType, t.collation, nullClause)
		if _, err := conn.ExecContext(ctx, modifySQL); err != nil {
			return fmt.Errorf("normalize foreign-key column %s.%s: %w", t.table, t.column, err)
		}
	}

	return nil
}

// CreateTableWithStrictMetadata creates a table and binds metadata to it
// atomically for the interactive authoring path (MetadataService.CreateSchema /
// CreateSchemaOptimized). It is the single strict-create chokepoint; its only
// callers are those two service methods.
//
// The physical CREATE is strict (plain `CREATE TABLE`, no IF NOT EXISTS), so a
// pre-existing physical table of the same name fails closed with a ConflictError
// (MySQL 1050) instead of silently rebinding new metadata to a stale orphan
// table — the create-side half of the #3484 invariant: creation may bind new
// metadata ONLY to a table this same call just created.
//
// Because the strict CREATE proves this call created the table moments ago (and
// no platform write path can populate a table that has no metadata yet), a
// PRE-COMMIT metadata failure compensates by dropping the just-created table, so
// an in-band retry starts clean and converges. This replaces the previous
// "preserve the physical table for fail-safe recovery" behavior, which was
// exactly the orphan-adoption mechanism being removed.
//
// Deliberate boundary — transaction ambiguity: the compensating drop fires
// ONLY after a proven rollback. A tx.Commit() or tx.Rollback() error leaves the
// semantic transaction outcome unknown, so we do NOT drop: if metadata did not
// land, the orphan converges via the documented DELETE-then-create reconcile
// loop; if it did land, the object is fully live. Dropping after either lost
// acknowledgement could manufacture live metadata without its table.
func (r *SchemaRepository) CreateTableWithStrictMetadata(ctx context.Context, def schema.TableDefinition, objectMeta *models.ObjectMetadata) (err error) {
	// 1. Create Physical Table (STRICT: fail closed on a pre-existing table).
	if err = r.createPhysicalTable(ctx, def, false); err != nil {
		return err
	}

	// compensateDrop removes the table this call just created. Safe precisely
	// because the strict CREATE above proves the table did not pre-exist. Used
	// only on PRE-COMMIT failures (see the commit-ambiguity boundary above).
	compensateDrop := func(stage string) {
		if _, dropErr := r.db.Exec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`", def.TableName)); dropErr != nil {
			// Log-only: a failed compensating drop degrades to one explicit
			// DELETE reconcile (the strict path's 409 names the remediation).
			r.logger.Warn(fmt.Sprintf("compensating drop of freshly-created table %s failed after %s error: %v", def.TableName, stage, dropErr))
		} else {
			r.logger.Info(fmt.Sprintf("compensated: dropped freshly-created table %s after %s failure", def.TableName, stage))
		}
	}

	if r.permissionStateDirectCoordinator != nil {
		completionProofRegistered := false
		err = r.permissionStateDirectCoordinator.transactions.WithPessimisticTransaction(
			ctx,
			func(txCtx context.Context) error {
				if !ports.RegisterTransactionCompletionHook(
					txCtx,
					func(completed ports.TransactionCompletionOutcome) {
						if completed == ports.TransactionCompletionRolledBack {
							compensateDrop("metadata-transaction")
						}
					},
				) {
					return apperrors.NewInternalError(
						"metadata transaction completion proof is unavailable",
						nil,
					)
				}
				completionProofRegistered = true
				tx := r.permissionStateOwnedTransaction(txCtx)
				if tx == nil {
					return apperrors.NewInternalError(
						"failed to extract metadata transaction",
						nil,
					)
				}
				if err := r.registerTable(def, tx); err != nil {
					return apperrors.NewInternalError(
						fmt.Sprintf("failed to register table %s", def.TableName),
						err,
					)
				}
				if err := r.InsertObjectMetadata(txCtx, objectMeta, tx); err != nil {
					return apperrors.NewInternalError(
						fmt.Sprintf("failed to register object (strict) %s", def.TableName),
						err,
					)
				}
				batchFields := r.prepareStrictCreateFieldsForBatch(def, objectMeta)
				if err := r.BatchSaveFieldMetadata(txCtx, batchFields, tx); err != nil {
					return apperrors.NewInternalError(
						fmt.Sprintf("failed to register fields for %s", def.TableName),
						err,
					)
				}
				return nil
			},
		)
		if err != nil {
			if !completionProofRegistered {
				compensateDrop("metadata-transaction-setup")
			}
			return err
		}
		r.logger.Info(fmt.Sprintf("table created and registered (strict): %s", def.TableName))
		return nil
	}

	// TRANSACTION: Register Metadata
	tx, err := r.db.Begin()
	if err != nil {
		compensateDrop("begin-tx")
		return apperrors.NewInternalError("failed to begin transaction", err)
	}
	// Ensure rollback on panic or error (safe to call after Commit)
	defer func() {
		_ = tx.Rollback()
	}()

	// 1. Register in _System_Table
	if err = r.registerTable(def, tx); err != nil {
		compensateDrop("register-table")
		return apperrors.NewInternalError(fmt.Sprintf("failed to register table %s", def.TableName), err)
	}

	// 2. Register Object Metadata (Strict)
	// We use InsertObjectMetadata instead of BatchSaveObjectMetadata to fail on uniqueness
	if err = r.InsertObjectMetadata(ctx, objectMeta, tx); err != nil {
		compensateDrop("register-object")
		return apperrors.NewInternalError(fmt.Sprintf("failed to register object (strict) %s", def.TableName), err)
	}

	// 3. Register Fields (Batch is OK here as fields belong to new object)
	batchFields := r.prepareStrictCreateFieldsForBatch(def, objectMeta)
	if err = r.BatchSaveFieldMetadata(ctx, batchFields, tx); err != nil {
		compensateDrop("register-fields")
		return apperrors.NewInternalError(fmt.Sprintf("failed to register fields for %s", def.TableName), err)
	}

	// COMMIT: on error do NOT compensate-drop — the commit may have durably
	// landed while the ack was lost (see commit-ambiguity boundary above).
	if err = tx.Commit(); err != nil {
		return apperrors.NewInternalError("failed to commit metadata transaction", err)
	}

	r.logger.Info(fmt.Sprintf("table created and registered (strict): %s", def.TableName))
	return nil
}

// BatchCreatePhysicalTables performs parallel DDL creation and then batch registers in _System_Table.
// Fail-safe behavior: this method never drops managed tables on failure.
// It returns the first error so callers can retry or resume safely.
func (r *SchemaRepository) BatchCreatePhysicalTables(ctx context.Context, defs []schema.TableDefinition) error {
	sem := make(chan struct{}, 10)
	var wg sync.WaitGroup
	var errMu sync.Mutex
	var ddlErrors []error

	r.logger.Info(fmt.Sprintf("starting parallel DDL for %d tables...", len(defs)))

	for _, def := range defs {
		wg.Add(1)
		go func(d schema.TableDefinition) {
			defer wg.Done()
			sem <- struct{}{}
			defer func() { <-sem }()

			if createErr := r.CreatePhysicalTable(ctx, d); createErr != nil {
				errMu.Lock()
				ddlErrors = append(ddlErrors, apperrors.NewInternalError(
					fmt.Sprintf("failed to create table %s", d.TableName), createErr))
				errMu.Unlock()
			}
		}(def)
	}

	wg.Wait()

	if len(ddlErrors) > 0 {
		firstErr := ddlErrors[0]
		r.logger.Error(fmt.Sprintf("batch DDL had %d error(s); fail-safe mode preserves all tables for resume",
			len(ddlErrors)), firstErr)
		return firstErr
	}

	// 2. Batch Register in _System_Table
	r.logger.Info(fmt.Sprintf("batch registering %d tables in _System_Table...", len(defs)))
	if regErr := r.BatchRegisterTables(defs, r.db); regErr != nil {
		r.logger.Error("batch registration failed; fail-safe mode preserves tables for resume", regErr)
		return apperrors.NewInternalError("failed to batch register tables", regErr)
	}

	return nil
}

// registerTable registers the table in _System_Table registry
func (r *SchemaRepository) registerTable(def schema.TableDefinition, tx interface{}) error {
	var exec Executor
	if tx != nil {
		if e, ok := tx.(Executor); ok {
			exec = e
		}
	}
	if exec == nil {
		exec = r.db
	}
	return r.BatchRegisterTables([]schema.TableDefinition{def}, exec)
}

// BatchRegisterTables registers multiple tables in _System_Table registry in a single statement
func (r *SchemaRepository) BatchRegisterTables(defs []schema.TableDefinition, tx interface{}) error {
	if len(defs) == 0 {
		return nil
	}
	var exec Executor
	if tx != nil {
		if e, ok := tx.(Executor); ok {
			exec = e
		}
	}
	if exec == nil {
		exec = r.db
	}

	var batchMaps []map[string]interface{}
	for _, def := range defs {
		id := schema.GenerateTableID(def.TableName)
		record := map[string]interface{}{
			constants.ColID:                       id,
			constants.FieldSysTable_TableName:     def.TableName,
			constants.FieldSysTable_TableType:     def.TableType,
			constants.FieldSysTable_Category:      def.Category,
			constants.FieldSysTable_Description:   def.Description,
			constants.FieldSysTable_IsManaged:     def.IsManaged,
			constants.FieldSysTable_SchemaVersion: def.SchemaVersion,
			constants.FieldSysTable_CreatedBy:     CreatedByBootstrap,
			constants.ColCreatedDate:              time.Now().UTC(),
			constants.FieldSysTable_IsDeleted:     false,
			constants.ColLastModifiedDate:         time.Now().UTC(),
		}
		batchMaps = append(batchMaps, record)
	}

	cols := []string{
		constants.ColID, constants.FieldSysTable_TableName, constants.FieldSysTable_TableType,
		constants.FieldSysTable_Category, constants.FieldSysTable_Description, constants.FieldSysTable_IsManaged,
		constants.FieldSysTable_SchemaVersion, constants.FieldSysTable_CreatedBy,
		constants.ColCreatedDate, constants.FieldSysTable_IsDeleted, constants.ColLastModifiedDate,
	}

	sqlStr, params := query.BulkInsertOrdered(constants.TableTable, cols, batchMaps, nil)

	sqlStr += " ON DUPLICATE KEY UPDATE " + strings.Join([]string{
		tableRegistryValuesUpdate(constants.FieldSysTable_TableType),
		tableRegistryValuesUpdate(constants.FieldSysTable_Category),
		tableRegistryValuesUpdate(constants.FieldSysTable_Description),
		tableRegistryValuesUpdate(constants.FieldSysTable_IsManaged),
		tableRegistryValuesUpdate(constants.FieldSysTable_SchemaVersion),
		tableRegistryValuesUpdate(constants.FieldSysTable_IsDeleted),
		fmt.Sprintf("`%s` = VALUES(`%s`)", constants.ColLastModifiedDate, constants.ColLastModifiedDate),
	}, ", ")

	_, err := exec.Exec(sqlStr, params...)
	return err
}

func tableRegistryValuesUpdate(column string) string {
	return fmt.Sprintf("`%s` = VALUES(`%s`)", column, column)
}

// DropTable unregisters an object and drops its physical table.
//
// Ordering is deliberate and load-bearing (#3484). The object's metadata
// (_System_Field, _System_Object + object/field permissions) is removed in a
// SINGLE transaction FIRST; then the physical table is dropped; only after a
// successful DROP is the _System_Table registry row cleared. TiDB DDL
// (DROP TABLE) auto-commits and cannot be rolled back, so it can never
// participate in the metadata transaction — which is exactly why the metadata
// must go first. The registry row is deliberately kept through the purge as the
// OWNERSHIP TOMBSTONE: it is the proof of platform provenance that survives a
// phase-2 failure and the ONLY thing that authorizes the DeleteSchema reconcile
// to finish the drop later. Every failure state stays consistent:
//
//   - metadata unregister fails  → the transaction rolls back and the object is
//     left fully intact (physical table + metadata). Nothing is half-removed.
//   - DROP TABLE fails afterwards → the object is already unregistered (no longer
//     live); the residual is an orphan table WITH its registry tombstone —
//     provably platform-owned, reconciled by retrying DELETE for the name.
//   - registry cleanup fails after a successful DROP → warn-only residual row;
//     harmless (nothing exists to drop), upserted on re-create, cleared by the
//     next DELETE of the name.
//
// The previous implementation dropped the table first and unregistered metadata
// in best-effort, warn-only statements, so any cleanup failure (or a crash mid
// sequence) left a live _System_Object row pointing at a missing table — an
// orphan that faults generic id→object resolution. This mirrors
// CreateTableWithStrictMetadata's fail-safe: a table without metadata is
// tolerable; live metadata without a table is not.
//
// Boundaries (#3484):
//   - Inbound FK: a phase-0 pre-check (inboundForeignKeyReferences) fails the
//     delete closed BEFORE any metadata mutation when another table references
//     this one, so an FK-blocked delete leaves the object fully live rather than
//     manufacturing a durable orphan. No cascade-drop of inbound FKs (SF parity:
//     the referencing relationship fields must be deleted first — and now can be,
//     because dropPhysicalColumnIfPresent drops the constraint before the column).
//   - Single-writer authoring: in-process ms.mu serializes this against
//     CreateSchema, and the strict plain-CREATE closes cross-replica create-vs-
//     create atomically. A residual reconcile-DROP-vs-remote-create window across
//     replicas cannot be closed transactionally (TiDB DDL auto-commits; pessimistic
//     locks do not gap-lock absent unique keys), so this PR assumes single-writer
//     interactive authoring (one backend replica owns schema authoring). A DB-level
//     authoring name lock is the named follow-up if multi-replica authoring becomes
//     a supported topology.
//   - Controlplane adoption: pack install / tenant-overlay apply deliberately keep
//     CREATE TABLE IF NOT EXISTS + metadata upsert for resume idempotence, so a new
//     pack/overlay object colliding with an interactive orphan still adopts. That
//     is a documented boundary of the strict-create guard, out of this repo.
func (r *SchemaRepository) DropTable(tableName string) (err error) {
	return r.dropTable(context.Background(), tableName)
}

func (r *SchemaRepository) dropTable(
	ctx context.Context,
	tableName string,
) (err error) {
	if err := rejectCatalogAuthorityTableDrop(tableName); err != nil {
		return err
	}
	r.logger.Warn(fmt.Sprintf("dropping table: %s", tableName))

	// Phase 0 — inbound-FK pre-check, BEFORE any metadata mutation. A physical
	// inbound foreign key (some other table referencing this one) makes DROP
	// TABLE fail durably (MySQL 3730). If we unregistered metadata first and then
	// hit that failure, we would manufacture exactly the durable
	// table-without-metadata orphan this ordering exists to prevent. Probing
	// physical truth (INFORMATION_SCHEMA) rather than live field metadata is
	// deliberate: pack field removal soft-deletes a field (IsDeleted=1) before its
	// DROP COLUMN, so a physical FK can outlive every IsDeleted=false metadata
	// read — the physical probe still catches it. On any inbound FK, fail closed
	// with a ConflictError naming the constraint(s) so the object stays FULLY LIVE
	// and the remediation reaches the client.
	if refs, checkErr := r.inboundForeignKeyReferences(tableName); checkErr != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check inbound foreign keys for %s", tableName), checkErr)
	} else if len(refs) > 0 {
		return apperrors.NewConflictErrorWithMessage(tableName, fmt.Sprintf(
			"cannot delete object %q: it is still referenced by foreign key constraint(s) %s; delete the referencing relationship field(s) first",
			tableName, strings.Join(refs, ", ")))
	}

	// Phase 1 — unregister ALL metadata atomically, before touching the table.
	if err = r.dropTableMetadata(ctx, tableName); err != nil {
		return err
	}

	// Phase 2 — all metadata (including the registry row) is gone; drop the
	// physical table. On FAILURE, write back a DROP-PENDING tombstone — the
	// registry row with IsDeleted=1 — as the ownership proof the DeleteSchema
	// reconcile requires to finish the drop on retry. The tombstone is written
	// ONLY on failure, so a successful drop leaves NOTHING behind that could
	// later authorize destruction of an unrelated same-name table: the
	// stale-authorization class is structurally impossible, not merely unlikely.
	// If the tombstone write itself fails (or the process crashes before it),
	// the orphan is unreconcilable through the API but SAFE — the reconcile
	// fails closed to a 404 and the strict-create 409 remediation routes the
	// operator to manual cleanup. Failure degrades toward less self-healing,
	// never toward false authorization.
	if _, err = r.db.Exec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`", tableName)); err != nil {
		r.markDropPendingTombstone(tableName)
		return apperrors.NewInternalError(fmt.Sprintf("object %s unregistered but physical table drop failed", tableName), err)
	}

	r.logger.Info(fmt.Sprintf("metadata unregistered and table dropped: %s", tableName))
	return nil
}

func (r *SchemaRepository) dropTableMetadata(
	ctx context.Context,
	tableName string,
) error {
	if err := rejectCatalogAuthorityTableDrop(tableName); err != nil {
		return err
	}
	coordinator := r.permissionStateDirectCoordinator
	if coordinator == nil {
		// Directly constructed repositories do not carry the production
		// composition graph. Give this operation one complete graph rather
		// than bypassing the Permission State projection and revision
		// authority; every lock, mutation, and revision advance then shares
		// the exact TransactionManager below.
		transactions := NewTransactionManager(database.NewConnectionFromDB(r.db))
		authority, err := NewPermissionStateAuthority(transactions)
		if err != nil {
			return apperrors.NewInternalError(
				"metadata unregister requires a permission state authority",
				err,
			)
		}
		coordinator, err = NewPermissionStateDirectWriterCoordinator(
			transactions,
			authority,
		)
		if err != nil {
			return apperrors.NewInternalError(
				"metadata unregister requires a direct writer coordinator",
				err,
			)
		}
	}
	transactions := coordinator.transactions
	if transactions == nil {
		return apperrors.NewInternalError(
			"metadata unregister requires a transaction manager",
			nil,
		)
	}
	return transactions.WithPessimisticTransaction(
		ctx,
		func(txCtx context.Context) error {
			return r.dropTableMetadataInOwnedPessimisticTransaction(
				txCtx,
				tableName,
				transactions,
				coordinator,
			)
		},
	)
}

func (r *SchemaRepository) dropTableMetadataInOwnedPessimisticTransaction(
	ctx context.Context,
	tableName string,
	transactions *TransactionManager,
	coordinator *PermissionStateDirectWriterCoordinator,
) error {
	const writerID = "nexuscrm-persistence/schema_ddl_ops.go#SchemaRepository.dropTableMetadataInOwnedPessimisticTransaction"
	if coordinator == nil || coordinator.transactions != transactions {
		return apperrors.NewInternalError(
			"metadata unregister requires one direct writer transaction owner",
			nil,
		)
	}
	if r.permissionStateDirectWriterShouldRun(ctx, writerID) {
		return coordinator.runProjected(
			ctx,
			writerID,
			dropTableMetadataPermissionStateTargets(tableName),
			r.getExecutor,
			func(txCtx context.Context) error {
				return r.dropTableMetadataInOwnedPessimisticTransaction(
					withPermissionStateDirectWriter(txCtx, writerID),
					tableName,
					transactions,
					coordinator,
				)
			},
		)
	}
	if transactions == nil || !transactions.ownsPessimisticTx(ctx) {
		return apperrors.NewInternalError(
			"metadata unregister requires the coordinator-owned pessimistic transaction",
			nil,
		)
	}
	tx, ok := transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return apperrors.NewInternalError(
			"metadata unregister requires the active pessimistic SQL transaction",
			nil,
		)
	}

	parentKeys, err := lockDropTableObjectParentKeys(ctx, tx, tableName)
	if err != nil {
		return err
	}
	if len(parentKeys) > 0 {
		if err := preflightManagedRowMutationCascades(
			ctx,
			transactions,
			managedRowCascadeDeleteTarget{
				tableName:  constants.TableObject,
				parentKeys: parentKeys,
			},
		); err != nil {
			return err
		}
	}

	// Delete fields and the object by the exact locked parent key. The same
	// finite identity drives preflight and mutation; the display/API name never
	// becomes a second destructive predicate.
	if len(parentKeys) == 1 {
		qrFields := query.Delete(constants.TableField).
			Where(constants.FieldObjectID, parentKeys[0]).
			Build()
		if _, err := tx.ExecContext(ctx, qrFields.SQL, qrFields.Params...); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to delete fields for object %s", tableName), err)
		}

		qrObj := query.Delete(constants.TableObject).
			Where(constants.FieldID, parentKeys[0]).
			Build()
		if _, err := tx.ExecContext(ctx, qrObj.SQL, qrObj.Params...); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to delete object metadata %s", tableName), err)
		}
	}

	// If the physical DROP below fails, a DROP-PENDING tombstone is written
	// back as the ownership proof — see markDropPendingTombstone.
	qrTab := query.Delete(constants.TableTable).
		Where(constants.FieldSysTable_TableName, tableName).
		Build()
	if _, err := tx.ExecContext(ctx, qrTab.SQL, qrTab.Params...); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to unregister table %s", tableName), err)
	}

	// Autonumber v2 counter rows are cascade-deleted by the field purge above.
	qrOp := query.Delete(constants.TableObjectPermissions).
		Where(constants.FieldObjectPermissions_SobjectType, tableName).
		Build()
	if _, err := tx.ExecContext(ctx, qrOp.SQL, qrOp.Params...); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to delete object permissions for %s", tableName), err)
	}

	qrFp := query.Delete(constants.TableFieldPermissions).
		Where(constants.FieldFieldPermissions_SobjectType, tableName).
		Build()
	if _, err := tx.ExecContext(ctx, qrFp.SQL, qrFp.Params...); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to delete field permissions for %s", tableName), err)
	}
	return nil
}

func lockDropTableObjectParentKeys(
	ctx context.Context,
	tx *sql.Tx,
	tableName string,
) ([]string, error) {
	rows, err := tx.QueryContext(
		ctx,
		"SELECT `Id` FROM `_System_Object` WHERE LOWER(`ApiName`) = LOWER(?) ORDER BY `Id` FOR UPDATE",
		tableName,
	)
	if err != nil {
		return nil, apperrors.NewInternalError(
			fmt.Sprintf("lock object metadata parent for %s", tableName),
			err,
		)
	}
	defer rows.Close()

	parentKeys := make([]string, 0, 1)
	for rows.Next() {
		var parentKey string
		if err := rows.Scan(&parentKey); err != nil {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("scan object metadata parent for %s", tableName),
				err,
			)
		}
		if strings.TrimSpace(parentKey) == "" {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("object metadata parent for %s has an empty Id", tableName),
				nil,
			)
		}
		parentKeys = append(parentKeys, parentKey)
		if len(parentKeys) > 1 {
			return nil, apperrors.NewInternalError(
				fmt.Sprintf("object metadata parent for %s is ambiguous", tableName),
				nil,
			)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError(
			fmt.Sprintf("iterate object metadata parent for %s", tableName),
			err,
		)
	}
	return parentKeys, nil
}

func dropTableMetadataPermissionStateTargets(tableName string) []permissionStateDirectTarget {
	recordTypeTarget := fmt.Sprintf(
		"%s IN (SELECT %s FROM %s WHERE %s)",
		quotePermissionStateDirectIdentifier(constants.FieldObjectID),
		quotePermissionStateDirectIdentifier(constants.FieldID),
		quotePermissionStateDirectIdentifier(constants.TableObject),
		permissionStateDirectCaseInsensitiveEqual(constants.FieldSysObject_APIName),
	)
	recordTypeVisibilityTarget := fmt.Sprintf(
		"%s IN (SELECT %s FROM %s WHERE %s)",
		quotePermissionStateDirectIdentifier(constants.FieldPermissionSetRecordTypeVisibility_RecordTypeID),
		quotePermissionStateDirectIdentifier(constants.FieldID),
		quotePermissionStateDirectIdentifier(constants.TableRecordType),
		recordTypeTarget,
	)
	return []permissionStateDirectTarget{
		permissionStateDirectTargetWithManagedPermissionSetCurrentOwnerPrelock(
			permissionStateDirectTargetFor(
				constants.TableFieldPermissions,
				permissionStateDirectCaseInsensitiveEqual(constants.FieldFieldPermissions_SobjectType),
				tableName,
			),
			constants.FieldFieldPermissions_ParentID,
		),
		permissionStateDirectTargetFor(
			constants.TableObject,
			permissionStateDirectCaseInsensitiveEqual(constants.FieldSysObject_APIName),
			tableName,
		),
		permissionStateDirectTargetWithManagedPermissionSetCurrentOwnerPrelock(
			permissionStateDirectTargetFor(
				constants.TableObjectPermissions,
				permissionStateDirectCaseInsensitiveEqual(constants.FieldObjectPermissions_SobjectType),
				tableName,
			),
			constants.FieldObjectPermissions_ParentID,
		),
		permissionStateDirectTargetFor(
			constants.TableRecordType,
			recordTypeTarget,
			tableName,
		),
		permissionStateDirectTargetWithPermissionSetDefinitionCurrentOwnerDetachPrelock(
			permissionStateDirectTargetFor(
				constants.TablePermissionSetRecordTypeVisibility,
				recordTypeVisibilityTarget,
				tableName,
			),
			constants.FieldPermissionSetRecordTypeVisibility_PermissionSetID,
		),
	}
}

// markDropPendingTombstone upserts the _System_Table row for a table whose
// physical DROP just failed, with IsDeleted=1 — the DROP-PENDING marker that is
// the ONLY state authorizing the DeleteSchema orphan-reconcile to drop a table
// with no live object metadata. Only this failure path ever writes IsDeleted=1;
// live registrations write IsDeleted=0 (BatchRegisterTables) and a successful
// drop deletes the row in the metadata transaction, so neither can leave a
// destructive authorization behind. Best-effort by design: a failed marker
// write leaves the orphan operator-only (fail-safe), and the caller already
// returns the DROP error.
func (r *SchemaRepository) markDropPendingTombstone(tableName string) {
	now := time.Now().UTC()
	sqlStr := fmt.Sprintf(
		"INSERT INTO %s (`%s`, `%s`, `%s`, `%s`, `%s`, `%s`) VALUES (?, ?, ?, ?, ?, ?) "+
			"ON DUPLICATE KEY UPDATE `%s` = VALUES(`%s`), `%s` = VALUES(`%s`)",
		constants.TableTable,
		constants.ColID, constants.FieldSysTable_TableName, constants.FieldSysTable_CreatedBy,
		constants.ColCreatedDate, constants.FieldSysTable_IsDeleted, constants.ColLastModifiedDate,
		constants.FieldSysTable_IsDeleted, constants.FieldSysTable_IsDeleted,
		constants.ColLastModifiedDate, constants.ColLastModifiedDate,
	)
	if _, err := r.db.Exec(sqlStr,
		schema.GenerateTableID(tableName), tableName, CreatedByBootstrap, now, true, now); err != nil {
		r.logger.Error(fmt.Sprintf(
			"table %s: physical drop failed AND the drop-pending tombstone could not be written; the orphan table is not reconcilable through the API and requires operator cleanup", tableName), err)
	}
}

// inboundForeignKeyReferences returns "CONSTRAINT on TABLE" descriptors for every
// physical foreign key in the current schema that references tableName from a
// DIFFERENT table. Self-referential FKs are excluded because they do not block a
// DROP TABLE. The comparison is case-insensitive (LOWER on both sides) because
// the interactive delete route lowercases the API name before it reaches
// DropTable, while INFORMATION_SCHEMA stores the canonical (possibly mixed-case)
// table name — a case-sensitive match would silently miss the constraint and let
// the delete degrade to a late phase-2 DROP failure.
func (r *SchemaRepository) inboundForeignKeyReferences(tableName string) ([]string, error) {
	const q = `
		SELECT CONSTRAINT_NAME, TABLE_NAME
		FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
		WHERE TABLE_SCHEMA = DATABASE()
		  AND REFERENCED_TABLE_NAME IS NOT NULL
		  AND LOWER(REFERENCED_TABLE_NAME) = LOWER(?)
		  AND LOWER(TABLE_NAME) <> LOWER(?)`
	rows, err := r.db.Query(q, tableName, tableName)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	var refs []string
	for rows.Next() {
		var constraintName, referencingTable string
		if err := rows.Scan(&constraintName, &referencingTable); err != nil {
			return nil, err
		}
		refs = append(refs, fmt.Sprintf("%s on %s", constraintName, referencingTable))
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return refs, nil
}

// HasDropPendingTombstone reports whether a DROP-PENDING tombstone — a
// _System_Table row with IsDeleted=1, written ONLY by markDropPendingTombstone
// when a physical DROP failed after the metadata purge — exists for the name.
// It is the sole proof the DeleteSchema orphan-reconcile accepts before it may
// DROP a table with no live object metadata (#3484). Because a successful drop
// deletes the registry row inside the metadata transaction and never writes
// the marker, no residue of a successful delete can ever authorize destruction
// of a later same-name table; and a live registration (IsDeleted=0) or an
// unregistered raw/copied table never matches. This replaces both the earlier
// system-column shape heuristic (defeated by a same-shape copy) and the
// keep-row-until-success variant (whose post-success cleanup failure left a
// stale row with authorization power).
//
// The name comparison is effectively case-insensitive via the platform's
// utf8mb4 CI collation, matching how the delete route may lowercase the name.
func (r *SchemaRepository) HasDropPendingTombstone(tableName string) (bool, error) {
	q := fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE %s = ? AND %s = ?",
		constants.TableTable, constants.FieldSysTable_TableName, constants.FieldSysTable_IsDeleted)
	var count int
	if err := r.db.QueryRow(q, tableName, true).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

// ValidateSchema checks if a table matches its expected definition by comparing
// the actual database schema (from INFORMATION_SCHEMA) with the registered metadata.
// Returns an error if drift is detected, nil if schema matches or table doesn't exist.
func (r *SchemaRepository) ValidateSchema(tableName string) error {
	// Get expected columns from _System_Field metadata
	metadataOnlyComputedFieldPredicate := fmt.Sprintf(
		"NOT (LOWER(TRIM(COALESCE(`%s`.`%s`, ''))) = 'active' "+
			"AND (TRIM(COALESCE(`%s`.`%s`, '')) != '' OR TRIM(COALESCE(`%s`.`%s`, '')) != ''))",
		constants.TableField, constants.FieldSysField_FormulaMode,
		constants.TableField, constants.FieldSysField_Formula,
		constants.TableField, constants.FieldSysField_ProjectionConfig,
	)
	expectedQR := query.From(constants.TableField).
		Select([]string{
			fmt.Sprintf("`%s`.`%s`", constants.TableField, constants.FieldAPIName),
			fmt.Sprintf("`%s`.`%s`", constants.TableField, constants.FieldType),
		}).
		Join("JOIN", constants.TableObject, "o",
			fmt.Sprintf("`%s`.`%s` = `o`.`%s`", constants.TableField, constants.FieldObjectID, constants.FieldID)).
		WhereRaw(fmt.Sprintf("`o`.`%s` = ?", constants.FieldObjectAPIName), []interface{}{tableName}).
		WhereRaw(fmt.Sprintf("COALESCE(`%s`.`%s`, 0) = 0", constants.TableField, constants.FieldIsDeleted), nil).
		WhereRaw(metadataOnlyComputedFieldPredicate, nil).
		Build()

	expectedRows, err := r.db.Query(expectedQR.SQL, expectedQR.Params...)
	if err != nil {
		return apperrors.NewInternalError("failed to query expected schema", err)
	}
	defer func() { _ = expectedRows.Close() }()

	expectedFields := make(map[string]string) // field_name -> type
	for expectedRows.Next() {
		var fieldName, fieldType string
		if err := expectedRows.Scan(&fieldName, &fieldType); err != nil {
			continue
		}
		expectedFields[fieldName] = fieldType
	}
	if err := expectedRows.Err(); err != nil {
		return apperrors.NewInternalError("error iterating expected fields", err)
	}

	// If no metadata registered, skip validation (table might be new or system-internal)
	if len(expectedFields) == 0 {
		return nil
	}

	// Get actual columns from INFORMATION_SCHEMA (raw SQL is appropriate for system introspection).

	actualQuery := `
		SELECT COLUMN_NAME, DATA_TYPE 
		FROM INFORMATION_SCHEMA.COLUMNS 
		WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
	`
	actualRows, err := r.db.Query(actualQuery, tableName)
	if err != nil {
		return apperrors.NewInternalError("failed to query actual schema", err)
	}
	defer func() { _ = actualRows.Close() }()

	actualFields := make(map[string]string)
	for actualRows.Next() {
		var columnName, dataType string
		if err := actualRows.Scan(&columnName, &dataType); err != nil {
			continue
		}
		actualFields[columnName] = dataType
	}
	if err := actualRows.Err(); err != nil {
		return apperrors.NewInternalError("error iterating actual fields", err)
	}

	// Compare: Find missing columns in database
	var missingColumns []string
	for expectedField := range expectedFields {
		if _, exists := actualFields[expectedField]; !exists {
			missingColumns = append(missingColumns, expectedField)
		}
	}

	if len(missingColumns) > 0 {
		return apperrors.NewValidationError("schema", fmt.Sprintf("schema drift detected for %s: missing columns %v", tableName, missingColumns))
	}

	return nil
}

// ValidateFieldDefinition validates the field schema against core assertions
func (r *SchemaRepository) ValidateFieldDefinition(field schema.ColumnDefinition) error {
	// 1. Naming Convention: snake_case (allow leading underscores for system fields)
	validName := regexp.MustCompile(constants.FieldNamePattern)
	if !validName.MatchString(field.Name) {
		return apperrors.NewValidationError("fieldName", fmt.Sprintf("field name '%s' must be alphanumeric (PascalCase/snake_case)", field.Name))
	}

	// 2. Type-Specific Assertions
	// Strict Validation: Ensure the type is resolvable
	if _, err := ResolveSQLType(field.Type); err != nil {
		return apperrors.NewValidationError(field.Name, fmt.Sprintf("invalid field type '%s' for column '%s': %v", field.Type, field.Name, err))
	}

	// Use LogicalType for validation to catch metadata issues (e.g. Picklist options)
	// regardless of the underlying SQL storage type (e.g. VARCHAR)
	checkType := field.LogicalType
	if checkType == "" {
		checkType = field.Type
	}

	switch checkType {
	case string(constants.FieldTypeReference):
		// User assumption: Lookups must specific a target
		if len(field.ReferenceTo) == 0 {
			return apperrors.NewValidationError(field.Name, fmt.Sprintf("lookup field '%s' must have a valid 'reference_to' target", field.Name))
		}
	case string(constants.FieldTypePicklist):
		// User assumption: Picklists must have options
		if len(field.Options) == 0 {
			// Fail Fast: Reject metadata that maps to a Picklist but has no options.
			return apperrors.NewValidationError(field.Name, fmt.Sprintf("picklist field '%s' must have at least one option", field.Name))
		}
	case string(constants.FieldTypeFormula):
		if field.Formula == "" {
			return apperrors.NewValidationError(field.Name, fmt.Sprintf("formula field '%s' must have a formula expression", field.Name))
		}
		if field.ReturnType == "" {
			return apperrors.NewValidationError(field.Name, fmt.Sprintf("formula field '%s' must have a valid return_type", field.Name))
		}
	}

	return nil
}

// SchemaHealth removed (moved to domain/schema)

// ValidateSchemaRegistry compares the registry against actual DB tables
func (r *SchemaRepository) ValidateSchemaRegistry() (*schema.SchemaHealth, error) {
	// Get expected tables
	qr := query.From(constants.TableTable).
		Select([]string{constants.FieldSysTable_TableName}).
		Where(constants.FieldSysTable_IsManaged, 1).
		Build()

	expectedRows, err := r.db.Query(qr.SQL, qr.Params...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = expectedRows.Close() }()

	expected := make(map[string]bool)
	for expectedRows.Next() {
		var tableName string
		if err := expectedRows.Scan(&tableName); err != nil {
			continue
		}
		expected[tableName] = true
	}
	if err := expectedRows.Err(); err != nil {
		return nil, apperrors.NewInternalError("error iterating expected tables", err)
	}

	// Get actual tables
	actualRows, err := r.db.Query(`
		SELECT TABLE_NAME 
		FROM INFORMATION_SCHEMA.TABLES 
		WHERE TABLE_SCHEMA = DATABASE()
	`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = actualRows.Close() }()

	actual := make(map[string]bool)
	for actualRows.Next() {
		var tableName string
		if err := actualRows.Scan(&tableName); err != nil {
			continue
		}
		actual[tableName] = true
	}
	if err := actualRows.Err(); err != nil {
		return nil, apperrors.NewInternalError("error iterating actual tables", err)
	}

	// Find missing tables
	var missing []string
	for exp := range expected {
		if !actual[exp] {
			missing = append(missing, exp)
		}
	}

	status := StatusHealthy
	if len(missing) > 0 {
		status = StatusUnhealthy
	}

	return &schema.SchemaHealth{
		Status:         status,
		ExpectedCount:  len(expected),
		ActualCount:    len(actual),
		MissingTables:  missing,
		RegistryHealth: StatusOperational,
	}, nil
}