package persistence

// STRICT ALIASING ENFORCEMENT
// This file manipulates physical schema columns.
// - DDL Operations: Must use PHYSICAL names (col.PhysicalName or constants.ColID).
// - Query Builder: Must use LOGICAL names (constants.FieldID) which are then aliased.
// Do not mix these contexts.

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"regexp"
	"sort"
	"strings"

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

var (
	// columnNameRegex validates names for tables and columns.
	// Allows alphanumeric (PascalCase, camelCase, snake_case) + underscores.
	columnNameRegex = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
)

// AddColumn adds a column to the table and registers it
func (r *SchemaRepository) AddColumn(tableName string, col schema.ColumnDefinition) error {
	log.Printf("➕ Adding column %s to table %s", col.Name, tableName)
	if err := validateCatalogColumnWriteProtectionProjection(tableName, col); err != nil {
		return err
	}

	// VALIDATION: Table Name
	// System tables (starting with _System_) are exempt from strict snake_case but we generally don't add columns to them dynamically
	if !constants.IsSystemTable(tableName) {
		if !columnNameRegex.MatchString(tableName) {
			return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name '%s': must be alphanumeric (PascalCase/snake_case)", tableName))
		}
	}

	// VALIDATION: Field Definition
	if err := r.ValidateFieldDefinition(col); err != nil {
		return err
	}

	// 0. IDEMPOTENCY CHECK: Check if column already exists (Zombie/Orphan recovery)
	exists, err := r.checkColumnExists(tableName, col.Name)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", tableName, col.Name), err)
	}
	if err := r.admitAdminFieldPhysicalWriteByObjectAPI(context.Background(), tableName, col.Name, false); err != nil {
		return err
	}

	if exists {
		log.Printf("⚠️  Orphan column detected: %s.%s exists in DB but missing in metadata.", tableName, col.Name)

		// Nullability drift reconciliation now runs in the SSoT
		// metadata-write gate (gatePhysicalShapeBeforeMetadata). The orphan
		// branch no longer needs a per-caller call here — the gate fires
		// before registerField writes the row at the bottom of this
		// function, applying the same relax / auto-tighten / fail-closed
		// rules to every metadata write across all entry points.

		// ROOT FIX: Check if existing column type matches expected type
		// If not, we need to MODIFY the column to fix the type mismatch
		existingType, err := r.getColumnType(tableName, col.Name)
		if err != nil {
			log.Printf("   ⚠️  Could not determine existing column type: %v. Proceeding with adoption.", err)
		} else {
			expectedType := col.Type // col.Type is already the SQL type (mapped in metadata_field_crud.go)
			if !r.isTypeCompatible(existingType, expectedType) {
				log.Printf("   🔄 Type mismatch detected: existing=%s, expected=%s. Executing MODIFY COLUMN...", existingType, expectedType)

				// TiDB Fix: Strip UNIQUE from MODIFY COLUMN and handle separately
				colDDL := r.buildColumnDDL(col, false)
				isUnique := col.Unique
				// No need to replace " UNIQUE" anymore

				modifyDDL := fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN %s", tableName, colDDL)
				log.Printf("   🏁 Executing DDL: %s", modifyDDL)

				if _, err := r.db.Exec(modifyDDL); err != nil {
					log.Printf("   ⚠️  MODIFY COLUMN failed (data may be incompatible): %v. Proceeding with adoption.", err)
				} else {
					log.Printf("   ✅ Column type corrected successfully")

					// Add Unique Index if needed
					if isUnique {
						idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
						idxDDL := fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE KEY `%s` (`%s`)", tableName, idxName, col.Name)
						log.Printf("   🔑 Adding Unique Constraint: %s", idxDDL)
						if _, err := r.db.Exec(idxDDL); err != nil {
							if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
								log.Printf("   ⚠️  Unique Index already exists, skipping.")
							} else {
								log.Printf("   ⚠️  Failed to add Unique Index: %v", err)
							}
						}
					}
				}
			} else {
				log.Printf("   ✅ Existing type is compatible. Adopting column...")

				// Ensure Unique Constraint if required (Idempotent)
				if col.Unique {
					idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
					idxDDL := fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE KEY `%s` (`%s`)", tableName, idxName, col.Name)
					log.Printf("   🔑 Ensuring Unique Constraint: %s", idxDDL)
					if _, err := r.db.Exec(idxDDL); err != nil {
						if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
							log.Printf("   ⚠️  Unique Index already exists, skipping.")
						} else {
							return apperrors.NewInternalError(fmt.Sprintf("failed to add Unique Index during adoption for %s.%s", tableName, col.Name), err)
						}
					}
				}
			}
		}
	} else {
		// 1. DDL: ADD COLUMN — use planner so JSON/TEXT/BLOB NOT NULL columns
		// with metadata defaults take the data-safe add-nullable / backfill /
		// tighten-NOT-NULL sequence. UNIQUE / FK constraints continue to run
		// as separate ALTERs after column creation.
		desiredType := r.ResolveColumnSqlType(col)
		stmts := PlanAddColumnStatements(tableName, col.Name, desiredType, col.Nullable, col.Default, col.CollationName)

		var indexDDL string
		if col.Unique {
			idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
			indexDDL = fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE KEY `%s` (`%s`)", tableName, idxName, col.Name)
		}

		for _, stmt := range stmts {
			log.Printf("   🏁 Executing DDL: %s", stmt)
			if _, err := r.db.Exec(stmt); err != nil {
				log.Printf("   ❌ DDL execution failed: %v", err)
				return apperrors.NewInternalError(fmt.Sprintf("failed to add column to table %s", tableName), err)
			}
		}

		// Execute Index DDL if present
		if indexDDL != "" {
			log.Printf("   🏁 Executing Index DDL: %s", indexDDL)
			if _, err := r.db.Exec(indexDDL); err != nil {
				// If index exists (duplicate), ignore. Else fail.
				if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
					log.Printf("   ⚠️  Unique Index already exists, skipping.")
				} else {
					return apperrors.NewInternalError("failed to add Unique Index", err)
				}
			}
		}

		log.Printf("   ✅ DDL execution complete")
	}

	// If it's a reference field (single target), we handle the Foreign Key in a separate statement
	var fkDDL string
	// Fix: Generate safe column name (max 64 chars) using helper
	safeColName := schema.GenerateSafeIdentifier(col.Name, "col")

	if len(col.ReferenceTo) == 1 && !col.IsPolymorphic && !strings.EqualFold(col.ReferenceTo[0], "parent") {
		rawFKName := fmt.Sprintf("fk_%s_%s", tableName, safeColName)
		fkName := schema.GenerateSafeIdentifier(rawFKName, "fk")

		// Use physical ID column name for FK target
		targetPK := constants.ColID

		fkDDL = fmt.Sprintf("ALTER TABLE `%s` ADD CONSTRAINT `%s` FOREIGN KEY (`%s`) REFERENCES `%s` (`%s`)",
			tableName, fkName, safeColName, col.ReferenceTo[0], targetPK)

		if col.OnDelete != "" {
			fkDDL += fmt.Sprintf(" ON DELETE %s", col.OnDelete)
		}
	}

	// 1.5. DDL: ADD FOREIGN KEY (if applicable)
	if fkDDL != "" {
		log.Printf("   🔗 Adding Foreign Key constraint...")
		if _, err := r.db.Exec(fkDDL); err != nil {
			// Handle "Duplicate constraint" error for idempotency
			if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
				log.Printf("⚠️  FK Constraint %s already exists, skipping...", fmt.Sprintf("fk_%s_%s", tableName, col.Name))
			} else {
				// Only rollback if we actually created the column in this run
				if !exists {
					log.Printf("⚠️ Failed to add FK, rolling back column: %v", err)
					if dropErr := r.dropPhysicalColumnIfPresentAfterCatalogAdmission(tableName, col.Name); dropErr != nil {
						log.Printf("⚠️ Rollback column drop failed: %v", dropErr)
					}
				}
				return apperrors.NewInternalError("failed to add foreign key constraint", err)
			}
		}
	}

	// 2. Register in _System_Field
	if err := r.registerField(tableName, col, r.db); err != nil {
		log.Printf("⚠️  Failed to register field %s.%s: %v. Attempting rollback...", tableName, col.Name, err)

		// COMPENSATION
		if !exists {
			rollbackDDL := fmt.Sprintf("ALTER TABLE `%s` DROP COLUMN `%s`", tableName, col.Name)
			if _, rbErr := r.db.Exec(rollbackDDL); rbErr != nil {
				// Critical error: Data vs Metadata inconsistency
				log.Printf("🔥 CRITICAL: Failed to rollback column %s.%s after metadata failure: %v", tableName, col.Name, rbErr)
				return apperrors.NewInternalError("failed to register field AND failed to rollback DDL (critical inconsistency)", err)
			}
			log.Printf("   ✅ Rollback successful: Column %s.%s dropped", tableName, col.Name)
		} else {
			log.Printf("⚠️  Skipping rollback for adopted orphan column %s.%s to preserve data.", tableName, col.Name)
		}

		return apperrors.NewInternalError(fmt.Sprintf("failed to register field metadata (DDL rolled back: %v)", !exists), err)
	}

	log.Printf("   ✅ Column added/registered: %s.%s", tableName, col.Name)
	return nil
}

// RegisterFieldOnly writes metadata to _System_Field without touching the physical table.
// Used for active formula fields whose values are computed by TiDB Views.
func (r *SchemaRepository) RegisterFieldOnly(tableName string, col schema.ColumnDefinition) error {
	return r.registerField(tableName, col, r.db)
}

// EnsureColumn guarantees a column exists and matches the metadata definition.
// It acts as a "Reconcile" function handling all 4 states:
// 1. Fresh Create (Missing Physical, Missing Meta)
// 2. Orphan Adoption (Existing Physical, Missing Meta)
// 3. Ghost Repair (Missing Physical, Existing Meta)
// 4. Consistency Check (Existing Physical, Existing Meta) - Verifies type & Upserts Valid Meta
func (r *SchemaRepository) EnsureColumn(tableName string, col schema.ColumnDefinition) error {
	// We deliberately delegate to AddColumn instead of short-circuiting on checkColumnExists.
	// AddColumn handles the complex logic of:
	// - Checking if physical column exists (Idempotency)
	// - Checking type compatibility (Drift Repair)
	// - Upserting metadata (Consistency)
	log.Printf("🛡️  Ensuring column %s.%s is reconciled...", tableName, col.Name)
	return r.AddColumn(tableName, col)
}

// checkColumnExists queries INFORMATION_SCHEMA to see if a column exists
func (r *SchemaRepository) checkColumnExists(tableName, columnName string) (bool, error) {
	query := `
		SELECT COUNT(*) 
		FROM INFORMATION_SCHEMA.COLUMNS 
		WHERE TABLE_SCHEMA = DATABASE() 
		  AND TABLE_NAME = ? 
		  AND COLUMN_NAME = ?
	`
	var count int
	if err := r.db.QueryRow(query, tableName, columnName).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

// checkColumnIndexed reports whether the column LEADS an index (SEQ_IN_INDEX =
// 1). A column that is only a trailing member of a composite index cannot serve
// a `COUNT(*) WHERE col = ?` lookup (wrong leading column), so it does not count
// as indexed for the relationship-FK purpose — a dedicated single-column index
// is still created.
func (r *SchemaRepository) checkColumnIndexed(tableName, columnName string) (bool, error) {
	const query = `
		SELECT COUNT(*)
		FROM INFORMATION_SCHEMA.STATISTICS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?
		  AND SEQ_IN_INDEX = 1
	`
	var count int
	if err := r.db.QueryRow(query, tableName, columnName).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

// EnsureColumnIndex creates a single-column secondary index on (tableName,
// columnName) when the column is not already indexed. Idempotent: a column that
// already participates in an index is left untouched, and a concurrent
// duplicate is treated as success. This keeps relationship (lookup /
// master-detail) FK columns indexed so related-list counts, lookups, joins, and
// RLS subqueries do not full-scan the child table. Generic by construction —
// the caller decides which columns are relationship FKs from field metadata; no
// SObject names appear here.
func (r *SchemaRepository) EnsureColumnIndex(tableName, columnName string) error {
	// Defense-in-depth: tableName/columnName are interpolated into the DDL
	// below. Every sibling DDL builder in this file validates identifiers the
	// same way; do so here too since EnsureColumnIndex is on the public
	// ports.SchemaRepository interface and must not trust its caller.
	if !columnNameRegex.MatchString(tableName) {
		return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name '%s': must be alphanumeric (PascalCase/snake_case)", tableName))
	}
	if !columnNameRegex.MatchString(columnName) {
		return apperrors.NewValidationError("columnName", fmt.Sprintf("invalid column name '%s': must be alphanumeric (PascalCase/snake_case)", columnName))
	}
	indexed, err := r.checkColumnIndexed(tableName, columnName)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check index for %s.%s", tableName, columnName), err)
	}
	if indexed {
		return nil
	}
	// Index names share TiDB's 64-char identifier limit. Route through the same
	// guard AddColumn uses for column/FK names: a long object+field pair would
	// otherwise produce a >64-char name and trip error 1059 ("Identifier name …
	// is too long") — which is NOT a duplicate, so it would propagate and the
	// best-effort callers would silently skip the index, leaving exactly the
	// #1407 full-scan this is meant to remove.
	rawName := fmt.Sprintf("idx_%s_%s", strings.ToLower(tableName), strings.ToLower(columnName))
	indexName := schema.GenerateSafeIdentifier(rawName, "idx")
	ddl := fmt.Sprintf("CREATE INDEX `%s` ON `%s` (`%s`)", indexName, tableName, columnName)
	if _, err := r.db.Exec(ddl); err != nil {
		// Lost a race or the index already exists (1061 "Duplicate key name") —
		// benign. Match AddColumn's duplicate convention in this file; a
		// non-duplicate failure (e.g. 1059) must still surface.
		low := strings.ToLower(err.Error())
		if strings.Contains(low, "duplicate") || strings.Contains(low, "already exist") {
			return nil
		}
		return apperrors.NewInternalError(fmt.Sprintf("failed to create index on %s.%s", tableName, columnName), err)
	}
	return nil
}

type physicalColumnSpecIndex map[string]map[string]*physicalColumnSpec

func (idx physicalColumnSpecIndex) get(tableName, columnName string) (*physicalColumnSpec, bool) {
	cols, ok := idx[tableName]
	if !ok {
		return nil, false
	}
	spec, ok := cols[columnName]
	return spec, ok
}

func (r *SchemaRepository) loadPhysicalColumnSpecsForTables(tableNames []string) (physicalColumnSpecIndex, error) {
	if len(tableNames) == 0 {
		return physicalColumnSpecIndex{}, nil
	}

	unique := make(map[string]struct{}, len(tableNames))
	for _, tableName := range tableNames {
		if strings.TrimSpace(tableName) == "" {
			continue
		}
		unique[tableName] = struct{}{}
	}
	if len(unique) == 0 {
		return physicalColumnSpecIndex{}, nil
	}

	sorted := make([]string, 0, len(unique))
	for tableName := range unique {
		sorted = append(sorted, tableName)
	}
	sort.Strings(sorted)

	placeholders := make([]string, 0, len(sorted))
	args := make([]interface{}, 0, len(sorted))
	for _, tableName := range sorted {
		placeholders = append(placeholders, "?")
		args = append(args, tableName)
	}

	q := fmt.Sprintf(`
		SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT, COLLATION_NAME
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME IN (%s)
	`, strings.Join(placeholders, ", "))

	rows, err := r.db.Query(q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	index := make(physicalColumnSpecIndex, len(sorted))
	for rows.Next() {
		var (
			tableName  string
			columnName string
			spec       physicalColumnSpec
			isNullable string
			colDefault sql.NullString
			extra      sql.NullString
			comment    sql.NullString
			collation  sql.NullString
		)
		if err := rows.Scan(
			&tableName, &columnName, &spec.ColumnType, &isNullable, &colDefault, &extra, &comment, &collation,
		); err != nil {
			return nil, err
		}
		spec.IsNullable = strings.EqualFold(isNullable, "YES")
		if colDefault.Valid {
			v := colDefault.String
			spec.ColumnDefault = &v
		}
		if extra.Valid {
			spec.Extra = extra.String
		}
		if comment.Valid {
			spec.Comment = comment.String
		}
		if collation.Valid {
			v := collation.String
			spec.CollationName = &v
		}
		if _, ok := index[tableName]; !ok {
			index[tableName] = make(map[string]*physicalColumnSpec)
		}
		specCopy := spec
		index[tableName][columnName] = &specCopy
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	return index, nil
}

func (r *SchemaRepository) DropColumnPhase1(tableName string, columnName string, expectedOwnerPack string) error {
	log.Printf("➖ Phase 1 dropping column %s from table %s", columnName, tableName)

	if err := validateGenericDropColumnRequest(tableName, columnName); err != nil {
		return err
	}
	expectedOwnerPack = strings.TrimSpace(expectedOwnerPack)
	if expectedOwnerPack == "" {
		return apperrors.NewValidationError("expectedOwnerPack", fmt.Sprintf("field %s.%s requires expected owner pack", tableName, columnName))
	}

	fieldID, ownerKind, ownerPack, err := r.loadFieldOwnership(tableName, columnName)
	if err != nil {
		return err
	}
	if ownerPack != expectedOwnerPack {
		return apperrors.NewValidationError("ownerPack", fmt.Sprintf("field %s.%s OwnerPack = %q, want %q", tableName, columnName, ownerPack, expectedOwnerPack))
	}
	switch ownerKind {
	case constants.FieldOwnerKindPack, constants.FieldOwnerKindPackRetained:
		if _, err := r.db.Exec(
			"UPDATE `_System_Field` SET OwnerKind = ?, OwnerPack = OwnerPack, IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP WHERE Id = ?",
			constants.FieldOwnerKindPackRemoving,
			fieldID,
		); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to mark field %s.%s removing", tableName, columnName), err)
		}
	case constants.FieldOwnerKindPackRemoving:
		if _, err := r.db.Exec(
			"UPDATE `_System_Field` SET IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP WHERE Id = ? AND COALESCE(IsDeleted, 0) = 0",
			fieldID,
		); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to soft-delete field %s.%s removing residue", tableName, columnName), err)
		}
	default:
		return apperrors.NewValidationError("ownerKind", fmt.Sprintf("field %s.%s OwnerKind = %q, want pack, pack_retained, or pack_removing", tableName, columnName, ownerKind))
	}
	if ownerPack == "" {
		return apperrors.NewValidationError("ownerPack", fmt.Sprintf("field %s.%s has empty OwnerPack", tableName, columnName))
	}

	if err := r.dropPhysicalColumnIfPresent(tableName, columnName); err != nil {
		return err
	}
	return r.closeFieldHistoryTrackingAfterPhysicalRemoval(context.Background(), fieldID)
}

func (r *SchemaRepository) DropColumnPhase2(tableName string, columnName string, expectedOwnerPack string) error {
	log.Printf("➖ Phase 2 dropping column metadata %s from table %s", columnName, tableName)

	if err := validateGenericDropColumnRequest(tableName, columnName); err != nil {
		return err
	}
	expectedOwnerPack = strings.TrimSpace(expectedOwnerPack)
	if expectedOwnerPack == "" {
		return apperrors.NewValidationError("expectedOwnerPack", fmt.Sprintf("field %s.%s requires expected owner pack", tableName, columnName))
	}
	exists, err := r.checkColumnExists(tableName, columnName)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", tableName, columnName), err)
	}
	if exists {
		return apperrors.NewValidationError("columnName", fmt.Sprintf("cannot delete metadata for %s.%s while physical column still exists", tableName, columnName))
	}

	return r.deleteFieldMetadataPreservingHistoryAuthority(
		context.Background(), tableName, columnName, constants.FieldOwnerKindPackRemoving, expectedOwnerPack,
	)
}

// DropColumn drops a column from the table and unregisters it
func (r *SchemaRepository) DropColumn(tableName string, columnName string) error {
	log.Printf("➖ Dropping column %s from table %s", columnName, tableName)

	if err := validateGenericDropColumnRequest(tableName, columnName); err != nil {
		return err
	}
	if err := r.admitAdminFieldPhysicalWriteByObjectAPI(context.Background(), tableName, columnName, true); err != nil {
		return err
	}

	if err := r.dropPhysicalColumnIfPresent(tableName, columnName); err != nil {
		return err
	}

	if err := r.deleteFieldMetadataPreservingHistoryAuthority(context.Background(), tableName, columnName, "", ""); err != nil {
		return err
	}

	log.Printf("   ✅ Column dropped: %s.%s", tableName, columnName)
	return nil
}

func (r *SchemaRepository) closeFieldHistoryTrackingAfterPhysicalRemoval(ctx context.Context, fieldID string) error {
	var tracked bool
	var deleted bool
	if err := r.db.QueryRowContext(ctx, "SELECT TrackHistory, IsDeleted FROM `_System_Field` WHERE Id = ?", fieldID).Scan(&tracked, &deleted); err != nil {
		return apperrors.NewInternalError("failed to read field history removal state", err)
	}
	if !tracked {
		return nil
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return apperrors.NewInternalError("failed to begin field history removal transition", err)
	}
	defer func() { _ = tx.Rollback() }()
	if deleted {
		if _, err := tx.ExecContext(ctx, "UPDATE `_System_Field` SET IsDeleted = 0 WHERE Id = ? AND IsDeleted = 1", fieldID); err != nil {
			return apperrors.NewInternalError("failed to stage field history removal transition", err)
		}
	}
	if _, err := r.ApplyHistoryTrackingTransition(ctx, fieldID, false, tx, false, false); err != nil {
		return err
	}
	if deleted {
		if _, err := tx.ExecContext(ctx, "UPDATE `_System_Field` SET IsDeleted = 1 WHERE Id = ?", fieldID); err != nil {
			return apperrors.NewInternalError("failed to restore field removal state", err)
		}
	}
	if err := tx.Commit(); err != nil {
		return apperrors.NewInternalError("failed to commit field history removal transition", err)
	}
	return nil
}

// deleteFieldMetadataPreservingHistoryAuthority closes any active history
// interval and hard-deletes the field row in one transaction. The durable
// sequence row intentionally survives so deterministic same-name recreation
// cannot reuse tokens against retained closed intervals.
func (r *SchemaRepository) deleteFieldMetadataPreservingHistoryAuthority(ctx context.Context, tableName, columnName, ownerKind, ownerPack string) error {
	if err := rejectCatalogAuthorityColumnDrop(tableName, columnName); err != nil {
		return err
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to begin field metadata delete for %s.%s", tableName, columnName), err)
	}
	defer func() { _ = tx.Rollback() }()

	lookup := "SELECT f.Id, f.TrackHistory FROM `_System_Field` f JOIN `_System_Object` o ON o.Id = f.ObjectId WHERE o.ApiName = ? AND f.ApiName = ?"
	args := []interface{}{tableName, columnName}
	if ownerKind != "" {
		lookup += " AND COALESCE(f.OwnerKind, '') = ? AND COALESCE(f.OwnerPack, '') = ?"
		args = append(args, ownerKind, ownerPack)
	}
	lookup += " LIMIT 1 FOR UPDATE"
	var fieldID string
	var tracked bool
	err = tx.QueryRowContext(ctx, lookup, args...).Scan(&fieldID, &tracked)
	if errors.Is(err, sql.ErrNoRows) {
		return tx.Commit()
	}
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to load field metadata delete authority for %s.%s", tableName, columnName), err)
	}
	if tracked {
		if _, err := r.ApplyHistoryTrackingTransition(ctx, fieldID, false, tx, false, false); err != nil {
			return err
		}
	}

	deleteSQL := "DELETE FROM `_System_Field` WHERE Id = ?"
	if _, err := tx.ExecContext(ctx, deleteSQL, fieldID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to delete field metadata for %s.%s", tableName, columnName), err)
	}
	if err := tx.Commit(); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to commit field metadata delete for %s.%s", tableName, columnName), err)
	}
	return nil
}

func validateDropColumnNames(tableName string, columnName string) error {
	if constants.IsSystemTable(tableName) {
		return nil
	}
	if !columnNameRegex.MatchString(tableName) {
		return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name '%s': must be alphanumeric (PascalCase/snake_case)", tableName))
	}
	if !columnNameRegex.MatchString(columnName) {
		return apperrors.NewValidationError("columnName", fmt.Sprintf("invalid column name '%s': must be alphanumeric (PascalCase/snake_case)", columnName))
	}
	return nil
}

func validateGenericDropColumnRequest(tableName string, columnName string) error {
	if err := validateDropColumnNames(tableName, columnName); err != nil {
		return err
	}
	return rejectCatalogAuthorityColumnDrop(tableName, columnName)
}

func (r *SchemaRepository) dropPhysicalColumnIfPresent(tableName string, columnName string) error {
	if err := rejectCatalogAuthorityColumnDrop(tableName, columnName); err != nil {
		return err
	}
	return r.dropPhysicalColumnIfPresentAfterCatalogAdmission(tableName, columnName)
}

// dropPhysicalColumnIfPresentAfterCatalogAdmission owns physical removal after
// the generic drop gate has admitted the target. AddColumn may also call it to
// compensate a column created by that same attempt after validating the
// catalog projection. Every other removal must use dropPhysicalColumnIfPresent
// so catalog authority columns remain immutable.
func (r *SchemaRepository) dropPhysicalColumnIfPresentAfterCatalogAdmission(tableName string, columnName string) error {
	exists, err := r.checkColumnExists(tableName, columnName)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", tableName, columnName), err)
	}
	if !exists {
		log.Printf("⚠️  Ghost column detected: %s.%s missing from DB. Skipping DDL...", tableName, columnName)
		return nil
	}
	// A relationship (lookup / master-detail) column carries a physical outbound
	// FK constraint installed at AddColumn time. A bare DROP COLUMN on an
	// FK-bearing column fails with MySQL 1828, dead-ending both the interactive
	// DropColumn path and the pack DropColumnPhase1 path — leaving the field
	// soft-deleted but physically present, which then blocks the object delete.
	// Drop the constraint(s) first so the column (and, in turn, the object) can be
	// removed through the platform without operator hand-SQL (#3484).
	if err := r.dropOutboundForeignKeysOnColumn(tableName, columnName); err != nil {
		return err
	}
	ddl := fmt.Sprintf("ALTER TABLE `%s` DROP COLUMN `%s`", tableName, columnName)
	if _, err := r.db.Exec(ddl); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to drop column from table %s", tableName), err)
	}
	return nil
}

// dropOutboundForeignKeysOnColumn drops every physical foreign-key constraint
// this (table, column) participates in as the referencing side, so a subsequent
// DROP COLUMN cannot 1828-dead-end. Case-insensitive matching mirrors the
// inbound-FK probe; a missing constraint is simply not returned by the query
// (no error). ALTER TABLE ... DROP FOREIGN KEY auto-commits — acceptable here
// because the caller is already committed to removing the column.
func (r *SchemaRepository) dropOutboundForeignKeysOnColumn(tableName string, columnName string) error {
	const q = `
		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(?)`
	rows, err := r.db.Query(q, tableName, columnName)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to look up foreign keys for %s.%s", tableName, columnName), err)
	}
	var constraints []string
	for rows.Next() {
		var name string
		if scanErr := rows.Scan(&name); scanErr != nil {
			_ = rows.Close()
			return apperrors.NewInternalError(fmt.Sprintf("failed to scan foreign key for %s.%s", tableName, columnName), scanErr)
		}
		constraints = append(constraints, name)
	}
	if rowsErr := rows.Err(); rowsErr != nil {
		_ = rows.Close()
		return apperrors.NewInternalError(fmt.Sprintf("error iterating foreign keys for %s.%s", tableName, columnName), rowsErr)
	}
	_ = rows.Close()

	for _, name := range constraints {
		ddl := fmt.Sprintf("ALTER TABLE `%s` DROP FOREIGN KEY `%s`", tableName, name)
		if _, err := r.db.Exec(ddl); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to drop foreign key %s on %s.%s", name, tableName, columnName), err)
		}
		log.Printf("   🔗 Dropped foreign key %s on %s.%s before DROP COLUMN", name, tableName, columnName)
	}
	return nil
}

func (r *SchemaRepository) loadFieldOwnership(tableName string, columnName string) (string, string, string, error) {
	var fieldID string
	var ownerKind sql.NullString
	var ownerPack sql.NullString
	err := r.db.QueryRow(
		"SELECT f.Id, COALESCE(f.OwnerKind, ''), COALESCE(f.OwnerPack, '') FROM `_System_Field` f JOIN `_System_Object` o ON o.Id = f.ObjectId WHERE o.ApiName = ? AND f.ApiName = ? LIMIT 1",
		tableName,
		columnName,
	).Scan(&fieldID, &ownerKind, &ownerPack)
	if err == sql.ErrNoRows {
		return "", "", "", apperrors.NewNotFoundError("field", fmt.Sprintf("%s.%s", tableName, columnName))
	}
	if err != nil {
		return "", "", "", apperrors.NewInternalError(fmt.Sprintf("failed to load field ownership for %s.%s", tableName, columnName), err)
	}
	return fieldID, strings.TrimSpace(ownerKind.String), strings.TrimSpace(ownerPack.String), nil
}

// ModifyColumn modifies a column's type (widening only: Boolean -> LongTextArea, etc.)
func (r *SchemaRepository) ModifyColumn(tableName, columnName string, newCol schema.ColumnDefinition) error {
	log.Printf("🔧 Modifying column %s.%s to type %s", tableName, columnName, newCol.Type)
	if strings.TrimSpace(newCol.Name) == "" {
		newCol.Name = columnName
	}
	if err := validateCatalogColumnWriteProtectionProjection(tableName, newCol); err != nil {
		return err
	}

	// VALIDATION: Table/Column Name
	if !constants.IsSystemTable(tableName) {
		if !columnNameRegex.MatchString(tableName) {
			return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name '%s': must be alphanumeric (PascalCase/snake_case)", tableName))
		}
		if !columnNameRegex.MatchString(columnName) {
			return apperrors.NewValidationError("columnName", fmt.Sprintf("invalid column name '%s': must be alphanumeric (PascalCase/snake_case)", columnName))
		}
	}
	if err := r.admitAdminFieldPhysicalWriteByObjectAPI(context.Background(), tableName, columnName, true); err != nil {
		return err
	}

	// Check column exists
	exists, err := r.checkColumnExists(tableName, columnName)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", tableName, columnName), err)
	}
	if !exists {
		return apperrors.NewNotFoundError("column", fmt.Sprintf("%s.%s", tableName, columnName))
	}

	// Pre-heal NULL rows when tightening to NOT NULL. The MODIFY emits the
	// resolved type without an inline DEFAULT clause for JSON/TEXT/BLOB
	// (Error 1101), so any pre-existing NULL row would block the ALTER.
	// Mirrors the bootstrap.ReconcileTableColumns MODIFY pre-heal so runtime
	// admin field changes go through the same data-safe path.
	if !newCol.Nullable {
		var nullCount int
		countSQL := fmt.Sprintf("SELECT COUNT(*) FROM `%s` WHERE `%s` IS NULL", tableName, columnName)
		if err := r.db.QueryRow(countSQL).Scan(&nullCount); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("count NULL rows for %s.%s before NOT NULL tighten", tableName, columnName), err)
		}
		if nullCount > 0 {
			if newCol.Default == "" {
				return apperrors.NewValidationError("default", fmt.Sprintf(
					"cannot tighten %s.%s to NOT NULL: %d existing NULL row(s) and no metadata default to backfill",
					tableName, columnName, nullCount,
				))
			}
			backfill := fmt.Sprintf("UPDATE `%s` SET `%s` = %s WHERE `%s` IS NULL",
				tableName, columnName, newCol.Default, columnName)
			log.Printf("   🩹 Pre-heal %d NULL row(s) in %s.%s before NOT NULL tighten (default=%s)",
				nullCount, tableName, columnName, newCol.Default)
			if _, err := r.db.Exec(backfill); err != nil {
				return apperrors.NewInternalError(fmt.Sprintf("pre-heal NULL rows for %s.%s", tableName, columnName), err)
			}
		}
	}

	// Execute ALTER TABLE MODIFY COLUMN
	// TiDB Fix: Separate UNIQUE constraint
	colDDL := r.buildColumnDDL(newCol, false)
	isUnique := newCol.Unique
	// No need to replace " UNIQUE" anymore

	ddl := fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN %s", tableName, colDDL)
	log.Printf("   🏁 Executing DDL: %s", ddl)
	if _, err := r.db.Exec(ddl); err != nil {
		log.Printf("   ❌ DDL execution failed: %v", err)
		return apperrors.NewInternalError(fmt.Sprintf("failed to modify column %s.%s", tableName, columnName), err)
	}

	// Add Unique Index if needed
	if isUnique {
		idxName := schema.GenerateUniqueIndexName(tableName, newCol.Name)
		idxDDL := fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE KEY `%s` (`%s`)", tableName, idxName, newCol.Name)
		log.Printf("   🔑 Adding Unique Constraint: %s", idxDDL)
		if _, err := r.db.Exec(idxDDL); err != nil {
			if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
				log.Printf("   ⚠️  Unique Index already exists, skipping.")
			} else {
				return apperrors.NewInternalError("failed to add Unique Index", err)
			}
		}
	}

	log.Printf("   ✅ DDL execution complete")

	// Update metadata in _System_Field
	if err := r.registerField(tableName, newCol, r.db); err != nil {
		log.Printf("⚠️  Failed to update field metadata for %s.%s: %v", tableName, columnName, err)
		// Don't rollback - the DDL succeeded and data is safe with the new type
		return apperrors.NewInternalError(fmt.Sprintf("column %s.%s modified but metadata update failed", tableName, columnName), err)
	}

	log.Printf("   ✅ Column modified: %s.%s", tableName, columnName)
	return nil
}

// gatePhysicalShapeBeforeMetadata is the SSoT enforcement point for the
// invariant "metadata.Required reflects physical NOT NULL". Every code path
// that persists field metadata (registerField, BatchSaveFieldMetadata)
// routes through this gate before the write is allowed.
//
// Why a single gate instead of per-caller checks: the runtime has many
// column-mutation entry points (AddColumn, BatchAddColumns, ModifyColumn,
// orphan-adoption branches inside each, future paths). Each path used to
// independently decide whether to call reconcileNullability before
// persisting metadata, and Codex repeatedly caught branches that forgot.
// Centralising the check here closes the bug class — any new caller
// automatically inherits the invariant without remembering to wire it up.
//
// The check is a no-op for fresh-create flows where the column does not
// physically exist yet (the caller's CREATE TABLE / ADD COLUMN DDL will
// establish the right shape directly). When the column exists, runs the
// full reconcileNullability decision tree: relax on NULL drift; auto-pre-
// heal + tighten on NOT NULL drift if a metadata default exists; fail
// closed when no default and live NULL rows would block the constraint.
func (r *SchemaRepository) gatePhysicalShapeBeforeMetadata(tableName string, col schema.ColumnDefinition) error {
	exists, err := r.checkColumnExists(tableName, col.Name)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("metadata-write gate: check column %s.%s exists", tableName, col.Name), err)
	}
	if !exists {
		// Fresh-create flow: the caller's DDL establishes the column shape;
		// reconcile would fail on a missing INFORMATION_SCHEMA row. Trust
		// the caller's CREATE/ADD path.
		return nil
	}
	if err := r.reconcileNullability(tableName, col); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("metadata-write gate: nullability reconcile for %s.%s", tableName, col.Name), err)
	}
	return nil
}

func (r *SchemaRepository) gatePhysicalShapeBeforeMetadataWithPrefetchedSpec(tableName string, col schema.ColumnDefinition, spec *physicalColumnSpec, exists bool) error {
	if !exists {
		return nil
	}
	if err := r.reconcileNullabilityWithPhysicalSpec(tableName, col, spec); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("metadata-write gate: nullability reconcile for %s.%s", tableName, col.Name), err)
	}
	return nil
}

// registerField registers a single field in _System_Field
func (r *SchemaRepository) registerField(tableName string, col schema.ColumnDefinition, exec Executor) error {
	if err := validateCatalogColumnWriteProtectionProjection(tableName, col); err != nil {
		return err
	}
	if exec == nil {
		exec = r.db
	}

	// SSoT gate: every metadata-write path enforces metadata/physical-shape
	// agreement here, so individual callers do not need to remember.
	if err := r.gatePhysicalShapeBeforeMetadata(tableName, col); err != nil {
		return err
	}

	// Fix: Lookup actual Object ID from DB to ensure FK integrity
	// We cannot use GenerateObjectID because it is now random/UUID based
	var objectID string
	qr := query.From(constants.TableObject).
		Select([]string{constants.FieldID}).
		Where(constants.FieldSysObject_APIName, tableName).
		Limit(1).
		Build()

	// Use the provided executor (tx or db)
	// QueryRow is supported by both *sql.DB and *sql.Tx
	row := exec.QueryRow(qr.SQL, qr.Params...)
	if err := row.Scan(&objectID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to resolve object ID for table %s", tableName), err)
	}

	fieldID := col.MetadataID
	if fieldID == "" {
		fieldID = schema.GenerateFieldID(tableName, col.Name)
	}
	field := r.buildFieldMetadataFromColumn(col)

	return r.SaveFieldMetadataWithIDs(context.Background(), field, objectID, tableName, fieldID, exec)
}

// physicalColumnSpec captures the attributes of an existing physical column
// that MODIFY COLUMN would silently drop if omitted from the new spec
// (MySQL/TiDB MODIFY replaces the entire column definition).
//
// Sourced from INFORMATION_SCHEMA.COLUMNS, so it reflects the live DB — never
// the logical schema spec (which does not carry COLLATE / COMMENT / ON UPDATE).
type physicalColumnSpec struct {
	ColumnType    string // e.g. "varchar(255)", "longtext", "int(11)"
	IsNullable    bool
	ColumnDefault *string // NULL default is a valid state distinct from "no default"
	Extra         string  // e.g. "on update CURRENT_TIMESTAMP", "auto_increment"
	Comment       string
	CollationName *string // NULL for non-string types
}

// getPhysicalColumnSpec reads the full attribute set of an existing physical
// column so a nullability relax ALTER can rebuild MODIFY COLUMN without
// dropping DEFAULT / COLLATE / COMMENT / ON UPDATE.
func (r *SchemaRepository) getPhysicalColumnSpec(tableName, columnName string) (*physicalColumnSpec, error) {
	const q = `
		SELECT COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT, COLLATION_NAME
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?
	`
	var (
		spec       physicalColumnSpec
		isNullable string
		colDefault sql.NullString
		extra      sql.NullString
		comment    sql.NullString
		collation  sql.NullString
	)
	if err := r.db.QueryRow(q, tableName, columnName).Scan(
		&spec.ColumnType, &isNullable, &colDefault, &extra, &comment, &collation,
	); err != nil {
		return nil, err
	}
	spec.IsNullable = strings.EqualFold(isNullable, "YES")
	if colDefault.Valid {
		v := colDefault.String
		spec.ColumnDefault = &v
	}
	if extra.Valid {
		spec.Extra = extra.String
	}
	if comment.Valid {
		spec.Comment = comment.String
	}
	if collation.Valid {
		v := collation.String
		spec.CollationName = &v
	}
	return &spec, nil
}

// buildRelaxNullabilityDDL rebuilds a MODIFY COLUMN clause that preserves every
// attribute of the physical column (type, default, collate, comment, extra)
// and only flips NOT NULL -> NULL.
func buildRelaxNullabilityDDL(tableName, columnName string, spec *physicalColumnSpec) string {
	return buildNullabilityDDL(tableName, columnName, spec, true)
}

// buildNullabilityDDL rebuilds a MODIFY COLUMN clause that preserves every
// attribute of the physical column (type, default, collate, comment, extra)
// and flips the NULL/NOT NULL flag according to makeNullable. Both relax and
// tighten paths share this builder so attribute preservation stays in one
// place.
//
// MySQL/TiDB MODIFY COLUMN replaces the full column spec: anything omitted is
// dropped. So we must restate each attribute the physical column carries.
//
// Ordering follows MySQL column-definition syntax:
//
//	<type> [COLLATE] [NULL|NOT NULL] [DEFAULT] [ON UPDATE] [AUTO_INCREMENT] [COMMENT]
func buildNullabilityDDL(tableName, columnName string, spec *physicalColumnSpec, makeNullable bool) string {
	var sb strings.Builder
	fmt.Fprintf(&sb, "ALTER TABLE `%s` MODIFY COLUMN `%s` %s", tableName, columnName, spec.ColumnType)

	if spec.CollationName != nil && *spec.CollationName != "" {
		fmt.Fprintf(&sb, " COLLATE %s", *spec.CollationName)
	}

	if makeNullable {
		sb.WriteString(" NULL")
	} else {
		sb.WriteString(" NOT NULL")
	}

	if spec.ColumnDefault != nil && SqlTypeAllowsDefaultClause(spec.ColumnType) {
		// INFORMATION_SCHEMA returns the default as a bare literal / expression
		// (e.g. "pending", "CURRENT_TIMESTAMP", "0"). Reuse the CURRENT_TIMESTAMP
		// / NULL bareword convention; other values are string-quoted.
		def := *spec.ColumnDefault
		upper := strings.ToUpper(strings.TrimSpace(def))
		if upper == "CURRENT_TIMESTAMP" || strings.HasPrefix(upper, "CURRENT_TIMESTAMP(") || upper == "NULL" {
			fmt.Fprintf(&sb, " DEFAULT %s", def)
		} else {
			fmt.Fprintf(&sb, " DEFAULT '%s'", strings.ReplaceAll(def, "'", "''"))
		}
	}

	extraLower := strings.ToLower(strings.TrimSpace(spec.Extra))
	// EXTRA may contain "on update CURRENT_TIMESTAMP", "auto_increment", etc.
	// Preserve verbatim; MySQL accepts its own EXTRA output as valid DDL input.
	if extraLower != "" {
		fmt.Fprintf(&sb, " %s", spec.Extra)
	}

	if spec.Comment != "" {
		fmt.Fprintf(&sb, " COMMENT '%s'", strings.ReplaceAll(spec.Comment, "'", "''"))
	}

	return sb.String()
}

// reconcileNullability detects nullability drift on an existing physical column
// and relaxes NOT NULL -> NULL when the spec declares the column nullable.
//
// Relax path (spec nullable, physical NOT NULL): emits
//
//	ALTER TABLE <t> MODIFY <col> <type> NULL
//
// This is safe: widening nullability never rejects existing data.
//
// Tighten path (spec NOT NULL, physical nullable): logged as WARN only.
// Auto-tightening is unsafe because existing NULL rows would fail the new
// constraint; the admin must resolve the data drift explicitly.
func (r *SchemaRepository) reconcileNullability(tableName string, col schema.ColumnDefinition) error {
	// Read the full physical spec in one query — we need nullability AND the
	// other attributes (DEFAULT / COLLATE / COMMENT / EXTRA) to rebuild a
	// MODIFY COLUMN clause that does not silently drop them (issue #1918).
	phys, err := r.getPhysicalColumnSpec(tableName, col.Name)
	if err != nil {
		return fmt.Errorf("failed to read physical column spec for %s.%s: %w", tableName, col.Name, err)
	}
	return r.reconcileNullabilityWithPhysicalSpec(tableName, col, phys)
}

func (r *SchemaRepository) reconcileNullabilityWithPhysicalSpec(tableName string, col schema.ColumnDefinition, phys *physicalColumnSpec) error {
	// Never attempt nullability changes on a PRIMARY KEY column — MySQL/TiDB
	// reject MODIFY COLUMN PK ... NULL (Error 1171).
	if col.PrimaryKey {
		return nil
	}

	physicalNullable := phys.IsNullable

	switch {
	case col.Nullable && !physicalNullable:
		// Spec nullable, DB NOT NULL -> relax. Rebuild the full column spec
		// from INFORMATION_SCHEMA so MODIFY COLUMN preserves every attribute
		// (type, default, collate, comment, extra) and only flips NULL.
		ddl := buildRelaxNullabilityDDL(tableName, col.Name, phys)
		log.Printf("   🔓 Relaxing nullability (spec NULL, DB NOT NULL) for %s.%s: %s", tableName, col.Name, ddl)
		if _, err := r.db.Exec(ddl); err != nil {
			return fmt.Errorf("failed to relax nullability for %s.%s: %w", tableName, col.Name, err)
		}
		log.Printf("   ✅ Nullability relaxed: %s.%s is now NULL-able", tableName, col.Name)
	case !col.Nullable && physicalNullable:
		// Spec NOT NULL, DB nullable. Two cases:
		//
		//   1. Restart-safety recovery for the two-phase ADD COLUMN path
		//      (PlanAddColumnStatements step 1 succeeded, step 3 did not, then
		//      AddColumn re-runs and finds an orphan nullable column with the
		//      target metadata declaring it NOT NULL). When the spec carries
		//      a metadata default we can pre-heal NULL rows from it and
		//      tighten via MODIFY ... NOT NULL — the same data-safe sequence
		//      ModifyColumn and bootstrap.ReconcileTableColumns use.
		//
		//   2. Pre-existing orphan with no metadata default. We cannot
		//      synthesize values, so retain the historical WARN-only
		//      behaviour and require explicit admin resolution.
		if col.Default == "" {
			log.Printf("   ⚠️  Nullability drift: spec says %s.%s is NOT NULL but physical column is nullable. "+
				"Auto-tightening is unsafe without a metadata default (existing NULL rows would fail the constraint). "+
				"Admin must resolve via data cleanup + explicit MODIFY.",
				tableName, col.Name)
			return nil
		}
		var nullCount int
		countSQL := fmt.Sprintf("SELECT COUNT(*) FROM `%s` WHERE `%s` IS NULL", tableName, col.Name)
		if err := r.db.QueryRow(countSQL).Scan(&nullCount); err != nil {
			return fmt.Errorf("count NULL rows for %s.%s before NOT NULL tighten: %w", tableName, col.Name, err)
		}
		if nullCount > 0 {
			backfill := fmt.Sprintf("UPDATE `%s` SET `%s` = %s WHERE `%s` IS NULL",
				tableName, col.Name, col.Default, col.Name)
			log.Printf("   🩹 Pre-heal %d NULL row(s) in %s.%s before NOT NULL tighten (default=%s)",
				nullCount, tableName, col.Name, col.Default)
			if _, err := r.db.Exec(backfill); err != nil {
				return fmt.Errorf("pre-heal NULL rows for %s.%s: %w", tableName, col.Name, err)
			}
		}
		ddl := buildNullabilityDDL(tableName, col.Name, phys, false)
		log.Printf("   🔒 Tightening nullability (spec NOT NULL, DB NULL) for %s.%s: %s", tableName, col.Name, ddl)
		if _, err := r.db.Exec(ddl); err != nil {
			return fmt.Errorf("failed to tighten nullability for %s.%s: %w", tableName, col.Name, err)
		}
		log.Printf("   ✅ Nullability tightened: %s.%s is now NOT NULL", tableName, col.Name)
	}
	return nil
}

// getColumnType retrieves the actual column type from the database
func (r *SchemaRepository) getColumnType(tableName, columnName string) (string, error) {
	query := `
		SELECT COLUMN_TYPE 
		FROM INFORMATION_SCHEMA.COLUMNS 
		WHERE TABLE_SCHEMA = DATABASE() 
		  AND TABLE_NAME = ? 
		  AND COLUMN_NAME = ?
	`
	var columnType string
	err := r.db.QueryRow(query, tableName, columnName).Scan(&columnType)
	if err != nil {
		return "", err
	}
	return strings.ToUpper(columnType), nil
}

// isTypeCompatible checks if existing DB type is compatible with expected type.
// This prevents unnecessary MODIFY operations for compatible types.
// Returns true if existing type can safely hold data of expected type.
func (r *SchemaRepository) isTypeCompatible(existing, expected string) bool {
	existing = strings.ToUpper(existing)
	expected = strings.ToUpper(expected)

	// Exact match (fastest path)
	if existing == expected {
		return true
	}

	// Extract base types: "DECIMAL(18,6)" -> "DECIMAL", "VARCHAR(255)" -> "VARCHAR"
	// Use package-level extractBaseType from schema_types.go
	existingBase := extractBaseType(existing)
	expectedBase := extractBaseType(expected)

	// Numeric types:
	// We require strict base type matching to avoid precision loss or overflow.
	// e.g. INT != DECIMAL, FLOAT != DECIMAL
	// Fall through to strict equality check at the end.

	// Text types: TEXT, MEDIUMTEXT, LONGTEXT are compatible with each other
	textTypes := map[string]bool{
		"TEXT": true, "MEDIUMTEXT": true, "LONGTEXT": true,
	}
	if textTypes[expectedBase] {
		return textTypes[existingBase]
	}

	// VARCHAR: exact base match (VARCHAR == VARCHAR)
	if existingBase == "VARCHAR" && expectedBase == "VARCHAR" {
		return true
	}

	// Default: require exact base type match
	return existingBase == expectedBase
}

// BatchAddColumns adds multiple columns and registers their metadata (Strict/Optimized)
func (r *SchemaRepository) BatchAddColumns(tableName string, fields []models.FieldMetadata) error {
	log.Printf("➕➕ Batch adding %d columns to table %s", len(fields), tableName)

	if len(fields) == 0 {
		return nil
	}

	// VALIDATION: Table Name
	if !constants.IsSystemTable(tableName) {
		if !columnNameRegex.MatchString(tableName) {
			return apperrors.NewValidationError("tableName", fmt.Sprintf("invalid table name '%s': must be alphanumeric (PascalCase/snake_case)", tableName))
		}
	}

	// VALIDATION: Column Names (Fail Fast)
	for _, field := range fields {
		if !columnNameRegex.MatchString(field.APIName) {
			return apperrors.NewValidationError("columnName", fmt.Sprintf("invalid column name '%s': must be alphanumeric (PascalCase/snake_case)", field.APIName))
		}
		if err := validateCatalogWriteProtectionProjection(tableName, &field); err != nil {
			return err
		}
	}

	// Resolve Object ID ONCE for metadata registration
	var objectID string
	qr := query.From(constants.TableObject).
		Select([]string{constants.FieldID}).
		Where(constants.FieldSysObject_APIName, tableName).
		Limit(1).
		Build()

	if err := r.db.QueryRow(qr.SQL, qr.Params...).Scan(&objectID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to resolve object ID for table %s", tableName), err)
	}

	var newCols []schema.ColumnDefinition
	var colsToModify []schema.ColumnDefinition
	var fkDDLs []string
	var uniqueClauses []string // Global scope for batch
	var fieldsToRegister []schema.FieldWithContext

	// 1. Process Fields
	for _, field := range fields {
		// Validating metadata structure implicitly via Service, but good to be safe?
		// Repo usually assumes valid domain objects from Service.

		preparedField, col, err := r.prepareBatchField(tableName, field)
		if err != nil {
			return apperrors.NewValidationError(field.APIName, err.Error())
		}
		// Check existence
		exists, err := r.checkColumnExists(tableName, col.Name)
		if err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to check column %s", col.Name), err)
		}
		if err := r.admitAdminFieldPhysicalWriteByObjectAPI(context.Background(), tableName, col.Name, false); err != nil {
			return err
		}

		// Prepare for Metadata Registration (Always register/update metadata for consistency)
		fieldID := preparedField.ID
		if fieldID == "" {
			fieldID = schema.GenerateFieldID(tableName, preparedField.APIName)
		}
		f := preparedField // copy

		// Ensure PhysicalName is always populated (Explicit Mapping)
		f.PhysicalName = col.Name

		fieldsToRegister = append(fieldsToRegister, schema.FieldWithContext{
			Field:                &f,
			ColumnDefinition:     col,
			TableName:            tableName,
			ObjectID:             objectID,
			FieldID:              fieldID,
			OwnerKind:            constants.FieldOwnerKindAdmin,
			CapabilitiesDeclared: f.GeneratedForPolymorphicReference != "",
		})

		if formulastorage.IsActiveComputedField(&field) {
			log.Printf("⚡ Active formula %s.%s — registering metadata only (no physical column)", tableName, field.APIName)
			continue
		}

		if exists {
			// Check Type Mismatch
			// We only modify if types are incompatible (e.g. INT -> VARCHAR) AND not identical.
			existingType, err := r.getColumnType(tableName, col.Name)
			if err == nil && !r.isTypeCompatible(existingType, col.Type) {
				log.Printf("🔄 Type mismatch for %s.%s: %s vs %s. Queueing MODIFY.", tableName, col.Name, existingType, col.Type)
				colsToModify = append(colsToModify, col)
			} else {
				if err != nil {
					log.Printf("⚠️ Could not determine existing type for %s.%s: %v", tableName, col.Name, err)
				}
				log.Printf("⚠️ Column %s.%s already exists (compatible). Skipping DDL, will ensure metadata.", tableName, col.Name)
				// Restart-safety for the two-phase ADD COLUMN flow (a prior
				// batch could have left step 1 done and step 3 missing) is
				// enforced in the SSoT metadata-write gate inside
				// BatchSaveFieldMetadata, so no per-caller reconcile is
				// needed here.
			}

			// Ensure Unique Constraint checks (Idempotent fix)
			if col.Unique {
				idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
				uniqueClauses = append(uniqueClauses, fmt.Sprintf("ADD UNIQUE KEY `%s` (`%s`)", idxName, col.Name))
			}
			continue
		}

		newCols = append(newCols, col)
	}

	// 2. Build Bulk DDL
	var ddlClauses []string

	// Two-phase columns (JSON/TEXT/BLOB NOT NULL with metadata default) cannot
	// share a batched ALTER because the safe sequence is multi-statement
	// (ADD nullable -> backfill -> MODIFY NOT NULL). Run them upfront as
	// singletons; everything else is batched as before.
	if len(newCols) > 0 {
		for _, col := range newCols {
			col.Name = schema.GenerateSafeIdentifier(col.Name, "col")

			desiredType := r.ResolveColumnSqlType(col)
			if IsAddColumnTwoPhase(col.Nullable, col.Default, desiredType) {
				stmts := PlanAddColumnStatements(tableName, col.Name, desiredType, col.Nullable, col.Default, col.CollationName)
				for _, stmt := range stmts {
					log.Printf("   🏁 Executing two-phase ADD step: %s", stmt)
					if _, err := r.db.Exec(stmt); err != nil {
						return apperrors.NewInternalError(fmt.Sprintf("two-phase add column %s.%s", tableName, col.Name), err)
					}
				}
			} else {
				colDDL := r.buildColumnDDL(col, false)
				ddlClauses = append(ddlClauses, fmt.Sprintf("ADD COLUMN %s", colDDL))
			}

			if col.Unique {
				idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
				uniqueClauses = append(uniqueClauses, fmt.Sprintf("ADD UNIQUE KEY `%s` (`%s`)", idxName, col.Name))
			}

			// Collect FKs
			if len(col.ReferenceTo) == 1 && !col.IsPolymorphic && !strings.EqualFold(col.ReferenceTo[0], "parent") {
				// Fix: Generate safe FK name using helper
				rawFKName := fmt.Sprintf("fk_%s_%s", tableName, col.Name)
				fkName := schema.GenerateSafeIdentifier(rawFKName, "fk")

				targetPK := constants.ColID

				fkDDL := fmt.Sprintf("ALTER TABLE `%s` ADD CONSTRAINT `%s` FOREIGN KEY (`%s`) REFERENCES `%s` (`%s`)",
					tableName, fkName, col.Name, col.ReferenceTo[0], targetPK)

				if col.OnDelete != "" {
					fkDDL += fmt.Sprintf(" ON DELETE %s", col.OnDelete)
				}
				fkDDLs = append(fkDDLs, fkDDL)
			}
		}
	}

	// Process Modifications
	if len(colsToModify) > 0 {
		for _, col := range colsToModify {
			// Fix: Generate safe column name
			col.Name = schema.GenerateSafeIdentifier(col.Name, "col")

			colDDL := r.buildColumnDDL(col, false)
			ddlClauses = append(ddlClauses, fmt.Sprintf("MODIFY COLUMN %s", colDDL))

			// Unique constraints are handled via uniqueClauses (global)
			// modifyClauses only needs to handle the type change
		}
	}

	if len(ddlClauses) > 0 {
		// Execute Column Additions/Modifications
		fullDDL := fmt.Sprintf("ALTER TABLE `%s` %s", tableName, strings.Join(ddlClauses, ", "))
		log.Printf("   🏁 Executing Batch DDL: %s", fullDDL)

		if _, err := r.db.Exec(fullDDL); err != nil {
			return apperrors.NewInternalError("batch DDL failed", err)
		}
		log.Printf("   ✅ Batch Columns Processed")
	}

	// 3. Execute Unique Keys (Separate Statements, includes keys for existing columns if needed)
	if len(uniqueClauses) > 0 {
		for _, clause := range uniqueClauses {
			stmt := fmt.Sprintf("ALTER TABLE `%s` %s", tableName, clause)
			log.Printf("   🏁 Executing Unique Key DDL: %s", stmt)
			if _, err := r.db.Exec(stmt); err != nil {
				if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
					log.Printf("   ⚠️  Unique Index already exists, skipping.")
				} else {
					return apperrors.NewInternalError(fmt.Sprintf("unique DDL failed (%s)", stmt), err)
				}
			}
		}
	}

	// 3. Execute FK DDLs
	var fkErrors []string
	for _, ddl := range fkDDLs {
		log.Printf("   🔗 Adding Foreign Key constraint...")
		if _, err := r.db.Exec(ddl); err != nil {
			// Skip duplicate FK (idempotent) - these are expected in resume scenarios
			if strings.Contains(err.Error(), "Duplicate") || strings.Contains(err.Error(), "already exists") {
				log.Printf("   ⚠️ FK already exists, skipping")
				continue
			}
			log.Printf("❌ FK creation failed: %v", err)
			fkErrors = append(fkErrors, err.Error())
		}
	}
	if len(fkErrors) > 0 {
		return apperrors.NewInternalError(fmt.Sprintf("FK creation failed: %v", fkErrors), fmt.Errorf("foreign key errors: %s", strings.Join(fkErrors, "; ")))
	}

	// 4. Register Metadata (Optimized Batch Insert)
	if err := r.BatchSaveFieldMetadata(context.Background(), fieldsToRegister, r.db); err != nil {
		return apperrors.NewInternalError("failed to batch register field metadata", err)
	}

	return nil
}

// mapFieldToColumnDef converts FieldMetadata to ColumnDefinition for DDL generation
func (r *SchemaRepository) mapFieldToColumnDef(field models.FieldMetadata) schema.ColumnDefinition {
	// Logic mirrors BatchCreateFields service logic but centralized here
	var relationshipName string
	if field.RelationshipName != nil {
		relationshipName = *field.RelationshipName
	}

	// Target physical name is identical to API name under Salesforce Native model
	physicalName := field.APIName

	colDef := schema.ColumnDefinition{
		MetadataID:            field.ID,
		Name:                  physicalName,
		Label:                 field.Label,
		Type:                  r.MapFieldTypeToSQL(string(field.Type)),
		LogicalType:           string(field.Type),
		Nullable:              fieldPhysicalNullable(field),
		Unique:                field.IsUnique,
		IsMasterDetail:        field.IsMasterDetail,
		RelationshipName:      relationshipName,
		RollupConfig:          schema.CloneRollupConfig(field.RollupConfig),
		WritableReferenceTo:   append([]string(nil), field.WritableReferenceTo...),
		WriteProtectionPolicy: field.WriteProtectionPolicy,
		ReadProtectionPolicy:  field.ReadProtectionPolicy,
	}
	if field.SiblingMaxParentField != nil {
		colDef.SiblingMaxParentField = strings.TrimSpace(*field.SiblingMaxParentField)
	}

	if field.DefaultValue != nil {
		// SECURITY: Escape single quotes to prevent SQL injection in DEFAULT value (e.g. O'Reilly -> O''Reilly)
		escapedVal := strings.ReplaceAll(*field.DefaultValue, "'", "''")
		colDef.Default = "'" + escapedVal + "'"
	}
	colDef.ReferenceTo = append([]string(nil), field.ReferenceTo...)
	if field.Type == constants.FieldTypeReference && field.IsPolymorphic {
		colDef.IsPolymorphic = true
	}
	colDef.GeneratedForPolymorphicReference = field.GeneratedForPolymorphicReference
	if field.Formula != nil {
		colDef.Formula = *field.Formula
	}
	if field.ReturnType != nil {
		colDef.ReturnType = string(*field.ReturnType)
	}
	if field.FormulaMode != nil {
		colDef.FormulaMode = *field.FormulaMode
	}
	if field.FormulaStorageMode != nil {
		colDef.FormulaStorageMode = *field.FormulaStorageMode
	}
	if field.ProjectionConfig != nil {
		colDef.ProjectionConfig = cloneFieldProjectionConfig(field.ProjectionConfig)
	}
	if field.Validator != nil {
		colDef.Validator = *field.Validator
	}
	if field.ValidatorConfig != nil {
		colDef.ValidatorConfig = *field.ValidatorConfig
	}

	// Delete Rule
	if field.DeleteRule != nil {
		switch constants.NormalizeDeleteRule(*field.DeleteRule) {
		case constants.DeleteRuleCascade:
			colDef.OnDelete = "CASCADE"
		case constants.DeleteRuleSetNull:
			colDef.OnDelete = "SET NULL"
		default:
			colDef.OnDelete = "RESTRICT"
		}
	}

	if len(field.Options) > 0 {
		colDef.Options = field.Options
	}

	return colDef
}

func fieldPhysicalNullable(field models.FieldMetadata) bool {
	if field.IsSystem {
		return !field.Required
	}
	if field.IsMasterDetail || constants.NormalizeFieldType(string(field.Type)) == constants.FieldTypeMasterDetail {
		return false
	}
	return true
}

// camelCaseSplitRe matches a lowercase letter followed by an uppercase letter.
var camelCaseSplitRe = regexp.MustCompile(`([a-z])([A-Z])`)

// splitCamelCase inserts a space before each uppercase letter that follows a lowercase letter.
// e.g. "FirstName" -> "First Name", "SystemModstamp" -> "System Modstamp"
func splitCamelCase(s string) string {
	return camelCaseSplitRe.ReplaceAllString(s, "$1 $2")
}