package bootstrapimpl

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"sort"
	"strings"
	"time"

	"github.com/nexuscrm/metadata"
	"github.com/pingcap-inc/nexuscrm-controlplane/packspec"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/persistencewiring"
)

// ReconcileTableColumns ensures all columns defined in table definitions
// exist in the physical DB with the correct type.
//   - Missing columns: ALTER TABLE ADD COLUMN
//   - Type mismatch: ALTER TABLE MODIFY COLUMN (with data safety checks for
//     narrowing and JSON-to-text capacity)
//
// Optimized: uses 1 bulk INFORMATION_SCHEMA query instead of per-table queries.
// Exported for use by control-plane packs (standard object column reconciliation).
func ReconcileTableColumns(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	log.Println("🔄 Reconciling system table columns...")
	start := time.Now()

	existingCols, err := LoadPhysicalColumnShapes(rawDB)
	if err != nil {
		return err
	}
	if err := preflightUnsupportedVarcharToVarbinaryChanges(existingCols, tableDefs); err != nil {
		return err
	}
	if err := preflightUniqueDefaultColumnAdds(rawDB, existingCols, tableDefs); err != nil {
		return err
	}

	var errs []string
	addCount := 0
	modifyCount := 0
	uniqueIndexAddCount := 0
	for _, def := range tableDefs {
		tableCols := existingCols[strings.ToLower(def.TableName)]
		if len(tableCols) == 0 {
			continue
		}

		for _, col := range def.Columns {
			desiredType := DesiredColumnType(col)

			existingShape, exists := tableCols[strings.ToLower(col.Name)]
			if !exists {
				// Column missing — ADD. The planner emits the data-safe sequence
				// for each column shape: a single ALTER for default-bearing types,
				// or add-nullable / backfill / tighten-NOT-NULL for JSON/TEXT/BLOB
				// columns whose metadata default cannot ride a physical DEFAULT
				// clause (MySQL/TiDB Error 1101).
				log.Printf("   ➕ Adding missing column: %s.%s (%s)", def.TableName, col.Name, desiredType)
				stmts := persistence.PlanAddColumnStatements(def.TableName, col.Name, desiredType, col.Nullable, col.Default, col.CollationName)
				stepFailed := false
				for _, stmt := range stmts {
					if _, err := rawDB.Exec(stmt); err != nil {
						log.Printf("   ⚠️  Failed step on %s.%s: %v (stmt=%s)", def.TableName, col.Name, err, stmt)
						errs = append(errs, fmt.Sprintf("add column %s.%s: %v", def.TableName, col.Name, err))
						stepFailed = true
						break
					}
				}
				if !stepFailed {
					addCount++
					if col.Unique {
						added, err := ensureColumnUniqueIndex(rawDB, def.TableName, col.Name)
						if err != nil {
							log.Printf("   ⚠️  Failed to ensure unique index for %s.%s: %v", def.TableName, col.Name, err)
							errs = append(errs, fmt.Sprintf("ensure unique index %s.%s: %v", def.TableName, col.Name, err))
							continue
						}
						if added {
							uniqueIndexAddCount++
						}
					}
				}
				continue
			}

			// Preserve the one reviewed legacy physical representation without
			// weakening the canonical catalog definition. This substitutes only
			// the physical type used for comparison/MODIFY; nullability, default,
			// collation, and unique-index enforcement remain authoritative.
			physicalDesired := physicalReconciliationColumnDefinition(def.TableName, existingShape, col)
			desiredType = DesiredColumnType(physicalDesired)

			// Column exists — check if type matches (normalize aliases first)
			existingType := existingShape.ColumnType
			typeMatches := ColumnTypesMatch(existingType, desiredType)
			if ColumnShapeMatches(existingShape, physicalDesired) {
				if col.Unique {
					added, err := ensureColumnUniqueIndex(rawDB, def.TableName, col.Name)
					if err != nil {
						log.Printf("   ⚠️  Failed to ensure unique index for %s.%s: %v", def.TableName, col.Name, err)
						errs = append(errs, fmt.Sprintf("ensure unique index %s.%s: %v", def.TableName, col.Name, err))
					} else if added {
						uniqueIndexAddCount++
					}
				}
				continue
			}

			// Type mismatch detected. Only auto-MODIFY for known safe changes:
			//   1. VARCHAR widening (128→255): always safe
			//   2. VARCHAR narrowing (255→128): safe if data fits
			//   3. Numeric widening (INT→BIGINT, INT→DECIMAL): safe when integer
			//      capacity and fractional scale do not shrink
			//   4. Temporal fractional-second precision widening within the same
			//      DATETIME or TIMESTAMP family: preserves every stored instant
			//   5. Equivalent aliases (BOOLEAN↔tinyint(1)): skip (no real change)
			//   6. JSON→TEXT-family storage: safe when every retained serialized
			//      value fits the target byte capacity; this intentionally removes
			//      the database JSON codec when the owner needs lexeme-preserving text
			// All other type changes (TEXT→VARCHAR, DECIMAL narrowing)
			// are skipped with a warning — they require explicit migration.
			//
			// TiDB VARCHAR→VARBINARY changes are rejected by the preflight above:
			// this generic reconciliation API cannot quiesce record writers while
			// replacing the charset and any affected index.
			if !typeMatches && !isAutoColumnTypeChangeSupported(existingType, desiredType) {
				log.Printf("   ⚠️  Column %s.%s type drift detected: DB=%s, expected=%s (unsafe automatic change, skipping — use explicit migration)",
					def.TableName, col.Name, existingType, desiredType)
				continue
			}

			if !typeMatches && isVarcharNarrowing(existingType, desiredType) {
				// Narrowing: check data safety before MODIFY
				newWidth := extractVarcharWidth(desiredType)
				if newWidth <= 0 {
					continue
				}
				var maxLen int
				checkSQL := fmt.Sprintf("SELECT COALESCE(MAX(CHAR_LENGTH(`%s`)), 0) FROM `%s`", col.Name, def.TableName)
				if err := rawDB.QueryRow(checkSQL).Scan(&maxLen); err != nil {
					log.Printf("   ⚠️  Failed to check data width for %s.%s: %v", def.TableName, col.Name, err)
					errs = append(errs, fmt.Sprintf("check data width %s.%s: %v", def.TableName, col.Name, err))
					continue
				}
				if maxLen > newWidth {
					msg := fmt.Sprintf(
						"cannot narrow %s.%s from %s to %s: max data length is %d chars, exceeds target %d",
						def.TableName,
						col.Name,
						existingType,
						desiredType,
						maxLen,
						newWidth,
					)
					log.Printf("   ❌ %s", msg)
					errs = append(errs, msg)
					continue
				}
			}

			if !typeMatches && isJSONToTextChange(existingType, desiredType) {
				capacity, _ := textColumnByteCapacity(desiredType)
				var maxBytes int64
				checkSQL := fmt.Sprintf("SELECT COALESCE(MAX(OCTET_LENGTH(`%s`)), 0) FROM `%s`", col.Name, def.TableName)
				if err := rawDB.QueryRow(checkSQL).Scan(&maxBytes); err != nil {
					log.Printf("   ⚠️  Failed to check serialized JSON size for %s.%s: %v", def.TableName, col.Name, err)
					errs = append(errs, fmt.Sprintf("check serialized JSON size %s.%s: %v", def.TableName, col.Name, err))
					continue
				}
				if maxBytes > capacity {
					msg := fmt.Sprintf(
						"cannot convert %s.%s from %s to %s: largest serialized JSON value is %d bytes, exceeds target capacity %d",
						def.TableName, col.Name, existingType, desiredType, maxBytes, capacity,
					)
					log.Printf("   ❌ %s", msg)
					errs = append(errs, msg)
					continue
				}
			}

			// If target is NOT NULL with a default, backfill NULL rows before MODIFY.
			// Mirrors the ADD-column path above: the ALTER would otherwise fail under
			// strict mode (Error 1265) when existing rows violate the new NOT NULL.
			if !col.Nullable {
				var nullCount int
				countSQL := fmt.Sprintf("SELECT COUNT(*) FROM `%s` WHERE `%s` IS NULL", def.TableName, col.Name)
				if err := rawDB.QueryRow(countSQL).Scan(&nullCount); err != nil {
					log.Printf("   ⚠️  Failed to count NULL rows for %s.%s: %v", def.TableName, col.Name, err)
					errs = append(errs, fmt.Sprintf("count nulls %s.%s: %v", def.TableName, col.Name, err))
					continue
				}
				if nullCount > 0 {
					if col.Default == "" {
						msg := fmt.Sprintf("cannot make %s.%s NOT NULL: %d existing NULL row(s) and no default configured",
							def.TableName, col.Name, nullCount)
						log.Printf("   ❌ %s", msg)
						errs = append(errs, msg)
						continue
					}
					backfill := fmt.Sprintf("UPDATE `%s` SET `%s` = %s WHERE `%s` IS NULL",
						def.TableName, col.Name, col.Default, col.Name)
					log.Printf("   🩹 Pre-heal %d NULL row(s) in %s.%s before NOT NULL narrowing (default=%s)",
						nullCount, def.TableName, col.Name, col.Default)
					if _, err := rawDB.Exec(backfill); err != nil {
						log.Printf("   ⚠️  Failed to pre-heal NULLs in %s.%s: %v", def.TableName, col.Name, err)
						errs = append(errs, fmt.Sprintf("pre-heal %s.%s: %v", def.TableName, col.Name, err))
						continue
					}
				}
			}

			// Safe VARCHAR change — MODIFY.
			// Applies the full column definition from system_tables.json (type + nullability + default).
			// This is intentional: the JSON definition is the desired state for all column properties.
			var colDDL strings.Builder
			colDDL.WriteString(fmt.Sprintf("`%s` %s", col.Name, desiredType))
			if strings.TrimSpace(col.CollationName) != "" {
				colDDL.WriteString(fmt.Sprintf(" COLLATE %s", strings.TrimSpace(col.CollationName)))
			}
			if !col.Nullable {
				colDDL.WriteString(" NOT NULL")
			}
			if col.Default != "" && persistence.SqlTypeAllowsDefaultClause(desiredType) {
				colDDL.WriteString(fmt.Sprintf(" DEFAULT %s", col.Default))
			}

			ddl := fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN %s", def.TableName, colDDL.String())
			log.Printf("   🔧 Modifying column shape: %s.%s (%s nullable=%v default=%s → %s nullable=%v default=%s)",
				def.TableName, col.Name, existingType, existingShape.Nullable, printableDefault(existingShape.DefaultVal),
				desiredType, col.Nullable, printableDesiredDefault(col.Default))
			if _, err := rawDB.Exec(ddl); err != nil {
				log.Printf("   ⚠️  Failed to modify column %s.%s: %v", def.TableName, col.Name, err)
				errs = append(errs, fmt.Sprintf("modify column %s.%s: %v", def.TableName, col.Name, err))
			} else {
				modifyCount++
				if col.Unique {
					added, err := ensureColumnUniqueIndex(rawDB, def.TableName, col.Name)
					if err != nil {
						log.Printf("   ⚠️  Failed to ensure unique index for %s.%s: %v", def.TableName, col.Name, err)
						errs = append(errs, fmt.Sprintf("ensure unique index %s.%s: %v", def.TableName, col.Name, err))
					} else if added {
						uniqueIndexAddCount++
					}
				}
			}
		}
	}
	log.Printf("✅ Column reconciliation complete (%d added, %d modified, %d unique indexes added, %v)", addCount, modifyCount, uniqueIndexAddCount, time.Since(start).Round(time.Millisecond))
	if len(errs) > 0 {
		return fmt.Errorf("column reconciliation had %d error(s): %s", len(errs), strings.Join(errs, "; "))
	}
	return nil
}

// preflightUnsupportedVarcharToVarbinaryChanges rejects charset transitions
// before reconciliation can execute any schema or data mutation.
func preflightUnsupportedVarcharToVarbinaryChanges(existingCols map[string]map[string]PhysicalColumnShape, tableDefs []schema.TableDefinition) error {
	var errs []string
	for _, def := range tableDefs {
		tableCols := existingCols[strings.ToLower(def.TableName)]
		if len(tableCols) == 0 {
			continue
		}
		for _, col := range def.Columns {
			existingShape, exists := tableCols[strings.ToLower(col.Name)]
			if !exists {
				continue
			}
			existingType := existingShape.ColumnType
			desiredType := DesiredColumnType(col)
			if isLegacySessionTokenPhysicalRepresentation(def.TableName, existingShape, col) {
				continue
			}
			if ColumnTypesMatch(existingType, desiredType) ||
				!isVarcharToVarbinaryChange(existingType, desiredType) {
				continue
			}
			reason := "target does not preserve the source VARCHAR's full utf8mb4 byte capacity"
			if isSafeVarcharToVarbinaryChange(existingType, desiredType) {
				reason = "TiDB requires an explicit writer-quiesced migration for charset and index replacement"
			}
			errs = append(errs, fmt.Sprintf(
				"cannot automatically convert %s.%s from %s to %s: %s",
				def.TableName, col.Name, existingType, desiredType, reason,
			))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("unsupported online column type change preflight failed: %s", strings.Join(errs, "; "))
	}
	return nil
}

func preflightUniqueDefaultColumnAdds(rawDB *sql.DB, existingCols map[string]map[string]PhysicalColumnShape, tableDefs []schema.TableDefinition) error {
	var errs []string
	for _, def := range tableDefs {
		tableCols := existingCols[strings.ToLower(def.TableName)]
		if len(tableCols) == 0 {
			continue
		}
		for _, col := range def.Columns {
			if !col.Unique || !uniqueDefaultCanDuplicateExistingRows(col.Default) {
				continue
			}
			if _, exists := tableCols[strings.ToLower(col.Name)]; exists {
				continue
			}
			rowCount, err := countTableRows(rawDB, def.TableName)
			if err != nil {
				errs = append(errs, fmt.Sprintf("count rows before unique default add %s.%s: %v", def.TableName, col.Name, err))
				continue
			}
			if rowCount > 1 {
				errs = append(errs, fmt.Sprintf("cannot add unique column %s.%s with default %s to table with %d existing rows: the default would create duplicate values before unique-index enforcement; add the field without a constant default or backfill unique values explicitly before enabling unique", def.TableName, col.Name, col.Default, rowCount))
			}
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("unique default column preflight failed: %s", strings.Join(errs, "; "))
	}
	return nil
}

func uniqueDefaultCanDuplicateExistingRows(defaultSQL string) bool {
	trimmed := strings.TrimSpace(defaultSQL)
	return trimmed != "" && !strings.EqualFold(trimmed, "NULL")
}

func countTableRows(rawDB *sql.DB, tableName string) (int, error) {
	var rowCount int
	if err := rawDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`", tableName)).Scan(&rowCount); err != nil {
		return 0, err
	}
	return rowCount, nil
}

func ensureColumnUniqueIndex(rawDB *sql.DB, tableName, columnName string) (bool, error) {
	exists, err := columnHasSingleColumnUniqueIndex(rawDB, tableName, columnName)
	if err != nil {
		return false, err
	}
	if exists {
		return false, nil
	}
	indexName := schema.GenerateUniqueIndexName(tableName, columnName)
	ddl := fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE INDEX `%s` (`%s`)", tableName, indexName, columnName)
	log.Printf("   🔑 Adding unique index: %s.%s", tableName, columnName)
	if _, err := rawDB.Exec(ddl); err != nil {
		if isDuplicateIndexNameError(err) {
			exists, checkErr := columnHasSingleColumnUniqueIndex(rawDB, tableName, columnName)
			if checkErr != nil {
				return false, fmt.Errorf("unique index name %s already exists; verifying %s.%s: %w", indexName, tableName, columnName, checkErr)
			}
			if exists {
				return false, nil
			}
		}
		return false, err
	}
	return true, nil
}

func columnHasSingleColumnUniqueIndex(rawDB *sql.DB, tableName, columnName string) (bool, error) {
	const q = `
		SELECT COUNT(*)
		  FROM INFORMATION_SCHEMA.STATISTICS s
		 WHERE s.TABLE_SCHEMA = DATABASE()
		   AND s.TABLE_NAME = ?
		   AND s.COLUMN_NAME = ?
		   AND s.NON_UNIQUE = 0
		   AND NOT EXISTS (
		       SELECT 1
		         FROM INFORMATION_SCHEMA.STATISTICS extra
		        WHERE extra.TABLE_SCHEMA = s.TABLE_SCHEMA
		          AND extra.TABLE_NAME = s.TABLE_NAME
		          AND extra.INDEX_NAME = s.INDEX_NAME
		          AND extra.COLUMN_NAME <> s.COLUMN_NAME
		   )
	`
	var count int
	if err := rawDB.QueryRow(q, tableName, columnName).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

// ReconcileSystemObjectTableColumns repairs the physical _System_Object column
// substrate directly from the authoritative managed table definition.
func ReconcileSystemObjectTableColumns(rawDB *sql.DB) error {
	for _, def := range GetManagedTableDefinitions() {
		if def.TableName == constants.TableObject {
			return ReconcileTableColumns(rawDB, []schema.TableDefinition{def})
		}
	}
	return fmt.Errorf("%s table definition is missing from managed schema definitions", constants.TableObject)
}

// ---------------------------------------------------------------------------
// runSchemaReconciliationUnderLock — single advisory lock for 0b + 0c
// ---------------------------------------------------------------------------

// runSchemaReconciliationUnderLock wraps the permission data migration (0b) and
// index reconciliation (0c) under a single advisory lock. The lock timeout is
// 10 minutes — enough for the one-time dedup of ~5.2M rows. During multi-pod
// rolling deploys, followers block here until the leader finishes, instead of
// crash-looping on a short timeout.
func runSchemaReconciliationUnderLock(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	return withSchemaReconciliationLock(rawDB, func() error {
		if err := migratePermissionDuplicates(rawDB); err != nil {
			return fmt.Errorf("permission data migration: %w", err)
		}
		if err := backfillActionInputGuardKeysBeforeIndex(rawDB); err != nil {
			return fmt.Errorf("action input guard key backfill: %w", err)
		}
		if err := backfillListViewDeveloperNamesBeforeIndex(rawDB); err != nil {
			return fmt.Errorf("list view developer name dedup: %w", err)
		}
		if err := reconcileSystemTableIndices(rawDB, tableDefs); err != nil {
			return fmt.Errorf("index reconciliation: %w", err)
		}
		return nil
	})
}

func withSchemaReconciliationLock(rawDB *sql.DB, fn func() error) error {
	ctx := context.Background()
	lockConn, err := rawDB.Conn(ctx)
	if err != nil {
		return fmt.Errorf("acquiring connection for schema lock: %w", err)
	}
	defer lockConn.Close()

	// 600 seconds = 10 minutes. The one-time 5.2M-row dedup takes several minutes;
	// followers wait here rather than fatal on timeout.
	var lockResult int
	if err := lockConn.QueryRowContext(ctx,
		"SELECT GET_LOCK('nexuscrm_schema_reconcile', 600)",
	).Scan(&lockResult); err != nil || lockResult != 1 {
		return fmt.Errorf("acquiring schema reconciliation lock (result=%d): %w", lockResult, err)
	}
	defer func() { _, _ = lockConn.ExecContext(ctx, "SELECT RELEASE_LOCK('nexuscrm_schema_reconcile')") }()

	return fn()
}

// backfillActionInputGuardKeysBeforeIndex derives GuardKey for legacy
// _System_ActionInputGuard rows before uq_aig_rule is reconciled from the old
// key shape to (StandardActionId, Condition, GuardKey). Soft-deleted tombstones
// are ignored by runtime reads, but the physical unique index still sees them;
// give those rows an Id-suffixed key so historical tombstones cannot block the
// new unique index.
//
// Caller must hold the schema reconciliation advisory lock.
func backfillActionInputGuardKeysBeforeIndex(rawDB *sql.DB) error {
	ready, err := actionInputGuardKeyBackfillColumnsReady(rawDB)
	if err != nil {
		return err
	}
	if !ready {
		log.Println("   ↩ _System_ActionInputGuard GuardKey backfill skipped (table/columns not present)")
		return nil
	}

	result, err := rawDB.Exec(fmt.Sprintf(`
UPDATE %s
SET %s = CASE
	WHEN %s = 1
		THEN CONCAT(
			CASE
				WHEN NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
					THEN CONCAT('permission:', TRIM(%s))
				WHEN NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
					AND NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
					THEN CONCAT('object:', TRIM(%s), ':', TRIM(%s))
				ELSE 'legacy'
			END,
			':deleted:',
			%s
		)
	WHEN NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
		THEN CONCAT('permission:', TRIM(%s))
	WHEN NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
		AND NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
		THEN CONCAT('object:', TRIM(%s), ':', TRIM(%s))
	ELSE %s
END
WHERE (%s IS NULL OR %s = '')
	AND (
		%s = 1
		OR NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
		OR (
			NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
			AND NULLIF(TRIM(COALESCE(%s, '')), '') IS NOT NULL
		)
	)`,
		quoteIdent(constants.TableActionInputGuard),
		quoteIdent(constants.FieldSysActionInputGuard_GuardKey),
		quoteIdent(constants.FieldSysActionInputGuard_IsDeleted),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredPermission),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredPermission),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredObject),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredOperation),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredObject),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredOperation),
		quoteIdent(constants.FieldSysActionInputGuard_ID),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredPermission),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredPermission),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredObject),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredOperation),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredObject),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredOperation),
		quoteIdent(constants.FieldSysActionInputGuard_GuardKey),
		quoteIdent(constants.FieldSysActionInputGuard_GuardKey),
		quoteIdent(constants.FieldSysActionInputGuard_GuardKey),
		quoteIdent(constants.FieldSysActionInputGuard_IsDeleted),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredPermission),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredObject),
		quoteIdent(constants.FieldSysActionInputGuard_RequiredOperation),
	))
	if err != nil {
		return fmt.Errorf("backfilling %s.%s: %w",
			constants.TableActionInputGuard, constants.FieldSysActionInputGuard_GuardKey, err)
	}
	if affected, err := result.RowsAffected(); err == nil && affected > 0 {
		log.Printf("   ✅ Backfilled %d _System_ActionInputGuard GuardKey values before unique-index reconciliation", affected)
	}
	return nil
}

func actionInputGuardKeyBackfillColumnsReady(rawDB *sql.DB) (bool, error) {
	required := map[string]bool{
		strings.ToLower(constants.FieldSysActionInputGuard_ID):                 false,
		strings.ToLower(constants.FieldSysActionInputGuard_GuardKey):           false,
		strings.ToLower(constants.FieldSysActionInputGuard_RequiredPermission): false,
		strings.ToLower(constants.FieldSysActionInputGuard_RequiredObject):     false,
		strings.ToLower(constants.FieldSysActionInputGuard_RequiredOperation):  false,
		strings.ToLower(constants.FieldSysActionInputGuard_IsDeleted):          false,
	}
	rows, err := rawDB.Query(
		"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME IN (?, ?, ?, ?, ?, ?)",
		constants.TableActionInputGuard,
		constants.FieldSysActionInputGuard_ID,
		constants.FieldSysActionInputGuard_GuardKey,
		constants.FieldSysActionInputGuard_RequiredPermission,
		constants.FieldSysActionInputGuard_RequiredObject,
		constants.FieldSysActionInputGuard_RequiredOperation,
		constants.FieldSysActionInputGuard_IsDeleted,
	)
	if err != nil {
		return false, fmt.Errorf("querying %s columns for GuardKey backfill: %w", constants.TableActionInputGuard, err)
	}
	defer rows.Close()

	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return false, fmt.Errorf("scanning %s column for GuardKey backfill: %w", constants.TableActionInputGuard, err)
		}
		key := strings.ToLower(name)
		if _, ok := required[key]; ok {
			required[key] = true
		}
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("iterating %s columns for GuardKey backfill: %w", constants.TableActionInputGuard, err)
	}
	for _, present := range required {
		if !present {
			return false, nil
		}
	}
	return true, nil
}

// backfillListViewDeveloperNamesBeforeIndex repairs historical duplicate
// _System_ListView(ObjectApiName, DeveloperName) rows before the unique index
// is reconciled. The repair is non-destructive: the newest row in each
// object-scoped DeveloperName group keeps the original business key, while the
// loser rows receive conflict-free Id-derived DeveloperNames within the object
// scope.
//
// Caller must hold the schema reconciliation advisory lock.
func backfillListViewDeveloperNamesBeforeIndex(rawDB *sql.DB) error {
	ready, err := listViewDeveloperNameDedupColumnsReady(rawDB)
	if err != nil {
		return err
	}
	if !ready {
		log.Println("   ↩ _System_ListView DeveloperName dedup skipped (table/columns not present)")
		return nil
	}

	rows, err := rawDB.Query(fmt.Sprintf(`
SELECT %s, %s, %s
FROM %s
WHERE %s IS NOT NULL
ORDER BY LOWER(TRIM(%s)), LOWER(TRIM(%s)), %s DESC, %s DESC, %s DESC`,
		quoteIdent(constants.FieldID),
		quoteIdent(constants.FieldObjectAPIName),
		quoteIdent(constants.FieldSysListView_DeveloperName),
		quoteIdent(constants.TableListView),
		quoteIdent(constants.FieldSysListView_DeveloperName),
		quoteIdent(constants.FieldObjectAPIName),
		quoteIdent(constants.FieldSysListView_DeveloperName),
		quoteIdent(constants.FieldSysListView_LastModifiedDate),
		quoteIdent(constants.FieldSysListView_CreatedDate),
		quoteIdent(constants.FieldID),
	))
	if err != nil {
		return fmt.Errorf("loading %s.%s values before unique index dedup: %w",
			constants.TableListView, constants.FieldSysListView_DeveloperName, err)
	}
	defer rows.Close()

	type listViewDeveloperNameRow struct {
		id            string
		objectAPIName string
		developerName string
	}
	var orderedRows []listViewDeveloperNameRow
	namesByObject := make(map[string]map[string]bool)
	groups := make(map[string][]listViewDeveloperNameRow)
	for rows.Next() {
		var row listViewDeveloperNameRow
		if err := rows.Scan(&row.id, &row.objectAPIName, &row.developerName); err != nil {
			return fmt.Errorf("scanning %s.%s values before unique index dedup: %w",
				constants.TableListView, constants.FieldSysListView_DeveloperName, err)
		}
		orderedRows = append(orderedRows, row)
		objectKey := listViewDeveloperNameObjectKey(row.objectAPIName)
		nameKey := listViewDeveloperNameKey(row.developerName)
		if namesByObject[objectKey] == nil {
			namesByObject[objectKey] = make(map[string]bool)
		}
		namesByObject[objectKey][nameKey] = true
		groupKey := objectKey + "\x00" + nameKey
		groups[groupKey] = append(groups[groupKey], row)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterating %s.%s values before unique index dedup: %w",
			constants.TableListView, constants.FieldSysListView_DeveloperName, err)
	}
	if len(orderedRows) == 0 {
		return nil
	}

	renamed := int64(0)
	for _, row := range orderedRows {
		groupKey := listViewDeveloperNameObjectKey(row.objectAPIName) + "\x00" + listViewDeveloperNameKey(row.developerName)
		group := groups[groupKey]
		if len(group) < 2 || group[0].id == row.id {
			continue
		}
		objectKey := listViewDeveloperNameObjectKey(row.objectAPIName)
		nextName, err := nextListViewDedupDeveloperName(row.id, namesByObject[objectKey])
		if err != nil {
			return err
		}
		result, err := rawDB.Exec(fmt.Sprintf("UPDATE %s SET %s = ?, %s = NOW() WHERE %s = ?",
			quoteIdent(constants.TableListView),
			quoteIdent(constants.FieldSysListView_DeveloperName),
			quoteIdent(constants.FieldSysListView_LastModifiedDate),
			quoteIdent(constants.FieldID),
		), nextName, row.id)
		if err != nil {
			return fmt.Errorf("deduplicating %s.%s for row %s before unique index: %w",
				constants.TableListView, constants.FieldSysListView_DeveloperName, row.id, err)
		}
		if affected, err := result.RowsAffected(); err == nil {
			renamed += affected
		}
	}
	if renamed > 0 {
		log.Printf("   ✅ Renamed %d duplicate _System_ListView DeveloperName value(s) before unique-index reconciliation", renamed)
	}
	return nil
}

func listViewDeveloperNameObjectKey(objectAPIName string) string {
	return strings.ToLower(strings.TrimSpace(objectAPIName))
}

func listViewDeveloperNameKey(developerName string) string {
	return strings.ToLower(strings.TrimSpace(developerName))
}

func nextListViewDedupDeveloperName(id string, usedNames map[string]bool) (string, error) {
	const prefix = "ListView_dedup_"
	token := listViewDedupDeveloperNameToken(id)
	for suffix := 1; ; suffix++ {
		suffixText := ""
		if suffix > 1 {
			suffixText = fmt.Sprintf("_%d", suffix)
		}
		maxTokenLen := 255 - len(prefix) - len(suffixText)
		if maxTokenLen < 1 {
			return "", fmt.Errorf("unable to generate %s.%s dedup DeveloperName for row %s: suffix exhausted VARCHAR(255)",
				constants.TableListView, constants.FieldSysListView_DeveloperName, id)
		}
		truncated := token
		if len(truncated) > maxTokenLen {
			truncated = truncated[:maxTokenLen]
		}
		candidate := prefix + truncated + suffixText
		key := listViewDeveloperNameKey(candidate)
		if !usedNames[key] {
			usedNames[key] = true
			return candidate, nil
		}
	}
}

func listViewDedupDeveloperNameToken(id string) string {
	id = strings.TrimSpace(id)
	var b strings.Builder
	b.Grow(len(id))
	for _, r := range id {
		switch {
		case r >= 'a' && r <= 'z':
			b.WriteRune(r)
		case r >= 'A' && r <= 'Z':
			b.WriteRune(r)
		case r >= '0' && r <= '9':
			b.WriteRune(r)
		case r == '_':
			b.WriteRune(r)
		case r == '-':
			b.WriteRune('_')
		default:
			b.WriteRune('_')
		}
	}
	if b.Len() == 0 {
		return "row"
	}
	return b.String()
}

func listViewDeveloperNameDedupColumnsReady(rawDB *sql.DB) (bool, error) {
	required := map[string]bool{
		strings.ToLower(constants.FieldID):                           false,
		strings.ToLower(constants.FieldObjectAPIName):                false,
		strings.ToLower(constants.FieldSysListView_DeveloperName):    false,
		strings.ToLower(constants.FieldSysListView_CreatedDate):      false,
		strings.ToLower(constants.FieldSysListView_LastModifiedDate): false,
	}
	rows, err := rawDB.Query(
		"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME IN (?, ?, ?, ?, ?)",
		constants.TableListView,
		constants.FieldID,
		constants.FieldObjectAPIName,
		constants.FieldSysListView_DeveloperName,
		constants.FieldSysListView_CreatedDate,
		constants.FieldSysListView_LastModifiedDate,
	)
	if err != nil {
		return false, fmt.Errorf("querying %s columns for DeveloperName dedup: %w", constants.TableListView, err)
	}
	defer rows.Close()

	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return false, fmt.Errorf("scanning %s column for DeveloperName dedup: %w", constants.TableListView, err)
		}
		key := strings.ToLower(name)
		if _, ok := required[key]; ok {
			required[key] = true
		}
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("iterating %s columns for DeveloperName dedup: %w", constants.TableListView, err)
	}
	for _, present := range required {
		if !present {
			return false, nil
		}
	}
	return true, nil
}

func quoteIdent(name string) string {
	return "`" + strings.ReplaceAll(name, "`", "``") + "`"
}

// ---------------------------------------------------------------------------
// migratePermissionDuplicates — one-time data migration for known duplicates
// ---------------------------------------------------------------------------

// migratePermissionDuplicates fixes known duplicate data in permission-related tables
// caused by historical boots where unique indexes were missing on existing tables.
//
// This is a targeted, idempotent migration (not a generic framework):
//   - Collapses conflicting permission columns to LEAST-permissive (never
//     widens) and logs each conflict as a destructive repair before selecting
//     survivors
//   - Keeps the row with the latest LastModifiedDate per unique group
//   - Uses Go-side ID batching (no LIMIT-in-subquery SQL compatibility risk)
//   - Fixes ParentId nullability for unique index correctness
//   - Reconciles duplicate profile-owned PermissionSets by re-pointing dependent rows
//   - Deduplicates PermissionSetTabSetting before unique index enforcement
//
// Caller must hold the schema reconciliation advisory lock.
func migratePermissionDuplicates(rawDB *sql.DB) error {
	log.Println("🔄 Checking permission table duplicates...")
	start := time.Now()

	// Fresh-DB guard: skip if tables don't exist yet
	existingTables, err := loadExistingTableSet(rawDB)
	if err != nil {
		return fmt.Errorf("loading existing tables: %w", err)
	}
	if !existingTables[strings.ToLower(constants.TableFieldPermissions)] &&
		!existingTables[strings.ToLower(constants.TableObjectPermissions)] {
		log.Println("   ↩ Permission tables don't exist yet (fresh DB) — skipping")
		return nil
	}

	// Fix ParentId → NOT NULL on both permission tables (required for unique index semantics).
	// In MySQL/TiDB, UNIQUE allows multiple NULLs, so the index only prevents duplicates
	// when the column is NOT NULL. Safe: verified 0 NULL values exist in production data.
	for _, tbl := range []string{constants.TableFieldPermissions, constants.TableObjectPermissions} {
		if !existingTables[strings.ToLower(tbl)] {
			continue
		}
		if _, err := rawDB.Exec(
			fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN `ParentId` VARCHAR(255) NOT NULL", tbl),
		); err != nil {
			return fmt.Errorf("altering %s.ParentId to NOT NULL: %w", tbl, err)
		}
	}

	// Deduplicate FieldPermissions
	if existingTables[strings.ToLower(constants.TableFieldPermissions)] {
		if err := deduplicatePermTable(rawDB,
			constants.TableFieldPermissions,
			[]string{"ParentId", "SobjectType", "Field"},
			[]string{"PermissionsRead", "PermissionsEdit"},
		); err != nil {
			return fmt.Errorf("FieldPermissions dedup: %w", err)
		}
	}

	// Deduplicate ObjectPermissions
	if existingTables[strings.ToLower(constants.TableObjectPermissions)] {
		if err := deduplicatePermTable(rawDB,
			constants.TableObjectPermissions,
			[]string{"ParentId", "SobjectType"},
			[]string{"PermissionsRead", "PermissionsCreate", "PermissionsEdit",
				"PermissionsDelete", "PermissionsViewAllRecords", "PermissionsModifyAllRecords"},
		); err != nil {
			return fmt.Errorf("ObjectPermissions dedup: %w", err)
		}
	}

	// Deduplicate profile-owned PermissionSet rows before enforcing
	// PermissionSet(ProfileId, IsOwnedByProfile) uniqueness.
	if existingTables[strings.ToLower(constants.TablePermissionSet)] {
		if err := deduplicateProfileOwnedPermissionSets(rawDB); err != nil {
			return fmt.Errorf("PermissionSet dedup: %w", err)
		}
	}

	// Deduplicate PermissionSetTabSetting before enforcing idx_permset_tab.
	if existingTables[strings.ToLower("PermissionSetTabSetting")] {
		if err := deduplicateSimpleTableByNewest(rawDB, "PermissionSetTabSetting", []string{"PermissionSetId", "TabName"}); err != nil {
			return fmt.Errorf("PermissionSetTabSetting dedup: %w", err)
		}
	}

	log.Printf("✅ Permission migration complete (%v)", time.Since(start).Round(time.Millisecond))
	return nil
}

// buildPermConflictDetectSQL selects the unique-key of every duplicate group
// whose rows DISAGREE on at least one boolean permission flag (MIN <> MAX, i.e.
// the group holds both true and false for that flag). Such a group cannot be
// merged to a provably-correct value without an authoritative source, so it is
// logged as a destructive repair rather than resolved silently.
func buildPermConflictDetectSQL(tableName string, uniqueCols, boolCols []string) string {
	colList := "`" + strings.Join(uniqueCols, "`, `") + "`"
	conflictExprs := make([]string, len(boolCols))
	for i, col := range boolCols {
		conflictExprs[i] = fmt.Sprintf("MIN(`%s`) <> MAX(`%s`)", col, col)
	}
	// Only live rows: the runtime effective read filters IsDeleted=0, so a
	// soft-deleted (tombstone) duplicate must not be treated as a conflicting
	// grant nor influence the merge.
	return fmt.Sprintf(
		"SELECT %s FROM `%s` WHERE `IsDeleted` = 0 GROUP BY %s HAVING %s",
		colList, tableName, colList, strings.Join(conflictExprs, " OR "),
	)
}

// buildPermLeastPermissiveMergeSQL collapses every row in a duplicate group to
// the LEAST-permissive (MIN) value per boolean column. MIN never widens an
// explicit false into true — dedup repair must never silently grant access.
// After this UPDATE all rows in a group are identical, so the survivor pick in
// Step 2 cannot change any value (no silent widening AND no silent narrowing).
func buildPermLeastPermissiveMergeSQL(tableName string, uniqueCols, boolCols []string) string {
	colList := "`" + strings.Join(uniqueCols, "`, `") + "`"
	minParts := make([]string, len(boolCols))
	setParts := make([]string, len(boolCols))
	for i, col := range boolCols {
		minParts[i] = fmt.Sprintf("MIN(`%s`) AS `_min_%s`", col, col)
		setParts[i] = fmt.Sprintf("t.`%s` = agg.`_min_%s`", col, col)
	}
	onParts := make([]string, 0, len(uniqueCols)+1)
	for _, col := range uniqueCols {
		onParts = append(onParts, fmt.Sprintf("t.`%s` = agg.`%s`", col, col))
	}
	// Only live rows participate, on both sides: a soft-deleted (IsDeleted=1)
	// duplicate must neither pull the group's MIN down (it is excluded from the
	// aggregate) nor be rewritten (t.IsDeleted = 0). Otherwise a tombstone would
	// silently revoke a live grant — the runtime read filters IsDeleted=0.
	onParts = append(onParts, "t.`IsDeleted` = 0")
	return fmt.Sprintf(
		"UPDATE `%s` t INNER JOIN (SELECT %s, %s FROM `%s` WHERE `IsDeleted` = 0 GROUP BY %s HAVING COUNT(*) > 1) agg ON %s SET %s",
		tableName, colList, strings.Join(minParts, ", "), tableName, colList,
		strings.Join(onParts, " AND "), strings.Join(setParts, ", "),
	)
}

// logPermDedupConflicts enumerates and logs duplicate groups whose permission
// flags disagree, so an operator can reconcile them against the import / source
// of truth after dedup collapses them to least-permissive.
func logPermDedupConflicts(rawDB *sql.DB, tableName string, uniqueCols, boolCols []string) error {
	if len(boolCols) == 0 {
		return nil
	}
	rows, err := rawDB.Query(buildPermConflictDetectSQL(tableName, uniqueCols, boolCols))
	if err != nil {
		return err
	}
	defer rows.Close()

	dest := make([]sql.NullString, len(uniqueCols))
	scan := make([]interface{}, len(uniqueCols))
	for i := range dest {
		scan[i] = &dest[i]
	}
	conflicts := 0
	for rows.Next() {
		if err := rows.Scan(scan...); err != nil {
			return err
		}
		parts := make([]string, len(uniqueCols))
		for i, col := range uniqueCols {
			parts[i] = fmt.Sprintf("%s=%q", col, dest[i].String)
		}
		conflicts++
		log.Printf("   ⚠️ %s dedup conflict (%s): rows disagree on a permission flag; "+
			"collapsing to least-permissive — review against authoritative source",
			tableName, strings.Join(parts, ", "))
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if conflicts > 0 {
		log.Printf("   ⚠️ %s: %d duplicate group(s) had conflicting permission values, collapsed to least-permissive (destructive repair)", tableName, conflicts)
	}
	return nil
}

// deduplicatePermTable removes duplicates from a permission table. Conflicting
// groups are logged then collapsed to the LEAST-permissive value (never widened),
// and the newest row per unique group is kept.
func deduplicatePermTable(rawDB *sql.DB, tableName string, uniqueCols, boolCols []string) error {
	colList := "`" + strings.Join(uniqueCols, "`, `") + "`"

	// Check for duplicates
	var dupGroups int64
	if err := rawDB.QueryRow(fmt.Sprintf(
		"SELECT COUNT(*) FROM (SELECT 1 FROM `%s` GROUP BY %s HAVING COUNT(*) > 1) t",
		tableName, colList,
	)).Scan(&dupGroups); err != nil {
		return fmt.Errorf("checking duplicates: %w", err)
	}
	if dupGroups == 0 {
		log.Printf("   ✅ No duplicates in %s", tableName)
		return nil
	}
	log.Printf("   🔧 %s: %d duplicate groups found, migrating...", tableName, dupGroups)

	// Step 1: Collapse each duplicate group to the LEAST-permissive value.
	//
	// The previous implementation broadcast MAX (most permissive), which silently
	// widened an explicit `false` into `true` — e.g. an imported read-only profile
	// whose duplicate row carried a default `Create=true` ended up with Create.
	// MIN never widens. A group whose rows genuinely disagree on a flag is a real
	// conflict that cannot be resolved correctly without an authoritative source,
	// so it is logged as a destructive repair first (neither silently widened nor
	// silently narrowed). After the MIN merge every row in a group is identical,
	// so the Step-2 survivor pick is value-safe.
	if err := logPermDedupConflicts(rawDB, tableName, uniqueCols, boolCols); err != nil {
		return fmt.Errorf("detecting %s dedup conflicts: %w", tableName, err)
	}
	if _, err := rawDB.Exec(buildPermLeastPermissiveMergeSQL(tableName, uniqueCols, boolCols)); err != nil {
		return fmt.Errorf("merging boolean columns (least-permissive): %w", err)
	}
	log.Printf("   ✅ %s: boolean merge complete (least-permissive)", tableName)

	// Step 2: Materialize duplicate IDs into a staging table (single full-scan).
	// Uses explicit CREATE + INSERT (not CTAS) for TiDB compatibility.
	// Survivor = live row first (IsDeleted ASC), then latest LastModifiedDate,
	// ties broken by Id DESC — so dedup never keeps a tombstone in place of a
	// live grant for the same unique key.
	tempTable := fmt.Sprintf("_dedup_%s_%d", strings.ToLower(tableName), time.Now().UnixMilli())
	defer func() { rawDB.Exec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`", tempTable)) }() //nolint:errcheck

	if _, err := rawDB.Exec(fmt.Sprintf(
		"CREATE TABLE `%s` (`Id` VARCHAR(255) NOT NULL PRIMARY KEY)", tempTable,
	)); err != nil {
		return fmt.Errorf("creating staging table %s: %w", tempTable, err)
	}
	if _, err := rawDB.Exec(fmt.Sprintf(
		"INSERT INTO `%s` SELECT `Id` FROM ("+
			"SELECT `Id`, ROW_NUMBER() OVER (PARTITION BY %s ORDER BY `IsDeleted` ASC, `LastModifiedDate` DESC, `Id` DESC) AS rn "+
			"FROM `%s`) sub WHERE sub.rn > 1",
		tempTable, colList, tableName,
	)); err != nil {
		return fmt.Errorf("populating staging table with duplicate IDs: %w", err)
	}

	// Count materialized duplicates
	var totalDups int64
	if err := rawDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`", tempTable)).Scan(&totalDups); err != nil {
		return fmt.Errorf("counting staged duplicates: %w", err)
	}
	log.Printf("   🗑  %s: removing %d duplicate rows...", tableName, totalDups)

	// Step 3: Batch-delete using Go-side ID fetch (TiDB-safe, no LIMIT in subquery).
	const batchSize = 5000
	totalDeleted := int64(0)
	for {
		// Fetch a batch of IDs from the temp table
		rows, err := rawDB.Query(fmt.Sprintf("SELECT `Id` FROM `%s` LIMIT %d", tempTable, batchSize))
		if err != nil {
			return fmt.Errorf("fetching batch from temp table: %w", err)
		}
		var ids []string
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				return fmt.Errorf("scanning ID: %w", err)
			}
			ids = append(ids, id)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return fmt.Errorf("iterating temp table rows: %w", err)
		}
		if len(ids) == 0 {
			break
		}

		// Build parameterized DELETE (no LIMIT-in-subquery)
		placeholders := make([]string, len(ids))
		params := make([]interface{}, len(ids))
		for i, id := range ids {
			placeholders[i] = "?"
			params[i] = id
		}
		inClause := strings.Join(placeholders, ",")

		result, err := rawDB.Exec(
			fmt.Sprintf("DELETE FROM `%s` WHERE `Id` IN (%s)", tableName, inClause), params...)
		if err != nil {
			return fmt.Errorf("batch delete: %w", err)
		}
		affected, _ := result.RowsAffected()
		totalDeleted += affected

		// Remove processed IDs from temp table
		if _, err := rawDB.Exec(
			fmt.Sprintf("DELETE FROM `%s` WHERE `Id` IN (%s)", tempTable, inClause), params...); err != nil {
			return fmt.Errorf("shrinking temp table: %w", err)
		}

		log.Printf("   🗑  %s: deleted %d (total: %d/%d)", tableName, affected, totalDeleted, totalDups)
	}

	log.Printf("   ✅ %s: deduplication complete — removed %d rows", tableName, totalDeleted)
	return nil
}

// deduplicateSimpleTableByNewest removes duplicate rows by unique key while keeping
// the newest row by LastModifiedDate/Id. It does not merge any value columns.
func deduplicateSimpleTableByNewest(rawDB *sql.DB, tableName string, uniqueCols []string) error {
	colList := "`" + strings.Join(uniqueCols, "`, `") + "`"

	var dupGroups int64
	if err := rawDB.QueryRow(fmt.Sprintf(
		"SELECT COUNT(*) FROM (SELECT 1 FROM `%s` GROUP BY %s HAVING COUNT(*) > 1) t",
		tableName, colList,
	)).Scan(&dupGroups); err != nil {
		return fmt.Errorf("checking duplicates: %w", err)
	}
	if dupGroups == 0 {
		log.Printf("   ✅ No duplicates in %s", tableName)
		return nil
	}

	tempTable := fmt.Sprintf("_dedup_%s_%d", strings.ToLower(tableName), time.Now().UnixMilli())
	defer func() { rawDB.Exec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`", tempTable)) }() //nolint:errcheck

	if _, err := rawDB.Exec(fmt.Sprintf(
		"CREATE TABLE `%s` (`Id` VARCHAR(255) NOT NULL PRIMARY KEY)", tempTable,
	)); err != nil {
		return fmt.Errorf("creating staging table %s: %w", tempTable, err)
	}
	if _, err := rawDB.Exec(fmt.Sprintf(
		"INSERT INTO `%s` SELECT `Id` FROM ("+
			"SELECT `Id`, ROW_NUMBER() OVER (PARTITION BY %s ORDER BY `LastModifiedDate` DESC, `Id` DESC) AS rn "+
			"FROM `%s`) sub WHERE sub.rn > 1",
		tempTable, colList, tableName,
	)); err != nil {
		return fmt.Errorf("populating staging table with duplicate IDs: %w", err)
	}

	result, err := rawDB.Exec(fmt.Sprintf("DELETE t FROM `%s` t INNER JOIN `%s` d ON t.`Id` = d.`Id`", tableName, tempTable))
	if err != nil {
		return fmt.Errorf("deleting duplicate rows from %s: %w", tableName, err)
	}
	deleted, _ := result.RowsAffected()
	log.Printf("   ✅ %s: deduplication complete — removed %d rows", tableName, deleted)
	return nil
}

// deduplicateProfileOwnedPermissionSets merges duplicate profile-owned permission sets,
// repoints permission rows to a canonical PermissionSet, and removes extra rows.
func deduplicateProfileOwnedPermissionSets(rawDB *sql.DB) error {
	var dupGroups int64
	err := rawDB.QueryRow(
		"SELECT COUNT(*) FROM (" +
			"SELECT 1 FROM `PermissionSet` " +
			"WHERE `IsOwnedByProfile` = 1 AND `ProfileId` IS NOT NULL AND `ProfileId` <> '' " +
			"GROUP BY `ProfileId`, `IsOwnedByProfile` HAVING COUNT(*) > 1" +
			") t",
	).Scan(&dupGroups)
	if err != nil {
		return fmt.Errorf("checking PermissionSet duplicates: %w", err)
	}
	if dupGroups == 0 {
		log.Printf("   ✅ No duplicates in %s", constants.TablePermissionSet)
		return nil
	}
	log.Printf("   🔧 %s: %d duplicate profile-owned groups found, migrating...", constants.TablePermissionSet, dupGroups)

	tempMap := fmt.Sprintf("_dedup_permissionset_map_%d", time.Now().UnixMilli())
	defer func() { rawDB.Exec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`", tempMap)) }() //nolint:errcheck

	if _, err := rawDB.Exec(fmt.Sprintf(
		"CREATE TABLE `%s` (`LoserId` VARCHAR(255) NOT NULL PRIMARY KEY, `WinnerId` VARCHAR(255) NOT NULL)",
		tempMap,
	)); err != nil {
		return fmt.Errorf("creating PermissionSet dedup map: %w", err)
	}

	insertMapSQL := fmt.Sprintf(
		"INSERT INTO `%s` (`LoserId`, `WinnerId`) "+
			"SELECT loser.`Id`, winner.`Id` FROM ("+
			"  SELECT `Id`, `ProfileId`, ROW_NUMBER() OVER ("+
			"    PARTITION BY `ProfileId`, `IsOwnedByProfile` "+
			"    ORDER BY `IsDeleted` ASC, `IsActive` DESC, `LastModifiedDate` DESC, `Id` DESC"+
			"  ) AS rn "+
			"  FROM `PermissionSet` WHERE `IsOwnedByProfile` = 1 AND `ProfileId` IS NOT NULL AND `ProfileId` <> ''"+
			") loser "+
			"JOIN ("+
			"  SELECT `Id`, `ProfileId` FROM ("+
			"    SELECT `Id`, `ProfileId`, ROW_NUMBER() OVER ("+
			"      PARTITION BY `ProfileId`, `IsOwnedByProfile` "+
			"      ORDER BY `IsDeleted` ASC, `IsActive` DESC, `LastModifiedDate` DESC, `Id` DESC"+
			"    ) AS rn "+
			"    FROM `PermissionSet` WHERE `IsOwnedByProfile` = 1 AND `ProfileId` IS NOT NULL AND `ProfileId` <> ''"+
			"  ) winners WHERE winners.rn = 1"+
			") winner ON winner.`ProfileId` = loser.`ProfileId` "+
			"WHERE loser.rn > 1",
		tempMap,
	)
	if _, err := rawDB.Exec(insertMapSQL); err != nil {
		return fmt.Errorf("building PermissionSet dedup map: %w", err)
	}

	var losers int64
	if err := rawDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`", tempMap)).Scan(&losers); err != nil {
		return fmt.Errorf("counting PermissionSet dedup rows: %w", err)
	}
	if losers == 0 {
		log.Printf("   ✅ No duplicate rows in %s after ranking", constants.TablePermissionSet)
		return nil
	}

	if err := mergeObjectPermissionsForPermissionSetDedup(rawDB, tempMap); err != nil {
		return err
	}
	if err := mergeFieldPermissionsForPermissionSetDedup(rawDB, tempMap); err != nil {
		return err
	}

	// Delete loser PermissionSets after dependent rows are repointed.
	result, err := rawDB.Exec(
		fmt.Sprintf("DELETE ps FROM `PermissionSet` ps INNER JOIN `%s` m ON ps.`Id` = m.`LoserId`", tempMap),
	)
	if err != nil {
		return fmt.Errorf("deleting loser PermissionSet rows: %w", err)
	}
	deleted, _ := result.RowsAffected()
	log.Printf("   ✅ %s: deduplication complete — removed %d rows", constants.TablePermissionSet, deleted)
	return nil
}

// mergeObject/FieldPermissionsForPermissionSetDedup consolidate a profile's
// duplicate profile-owned PermissionSets by folding loser rows into the winner.
// This MUST be access-preserving: the runtime computes effective access as the
// MAX-union over live rows across a profile's PermissionSets, so collapsing two
// of them into one must not change what the profile can do. Hence GREATEST on
// the flags and LEAST on IsDeleted (revive when any contributing row is live).
//
// Note: least-permissive collapse is correct for the INTRA-PermissionSet row
// dedup (deduplicatePermTable), where a duplicate row is a buggy double-write of
// one grant. It is NOT correct here: two profile-owned PermissionSets may carry
// distinct intended grants, and least-permissive would silently revoke access
// that was effective (reachable: an ObjectPermissions row can be soft-deleted
// via the generic SObject delete surface, so a winner tombstone + live loser
// would drop access). Choosing which duplicate PermissionSet is authoritative is
// the provenance-aware identity reconciliation deferred by #1374; until then,
// preserve access (do not regress).
func mergeObjectPermissionsForPermissionSetDedup(rawDB *sql.DB, tempMap string) error {
	mergeSQL := fmt.Sprintf(
		"INSERT INTO `%s` (`Id`, `ParentId`, `SobjectType`, `PermissionsRead`, `PermissionsCreate`, `PermissionsEdit`, `PermissionsDelete`, `PermissionsViewAllRecords`, `PermissionsModifyAllRecords`, `IsDeleted`, `CreatedDate`, `LastModifiedDate`) "+
			"SELECT UUID(), m.`WinnerId`, op.`SobjectType`, op.`PermissionsRead`, op.`PermissionsCreate`, op.`PermissionsEdit`, op.`PermissionsDelete`, op.`PermissionsViewAllRecords`, op.`PermissionsModifyAllRecords`, op.`IsDeleted`, op.`CreatedDate`, NOW() "+
			"FROM `%s` op INNER JOIN `%s` m ON op.`ParentId` = m.`LoserId` "+
			"ON DUPLICATE KEY UPDATE "+
			"`PermissionsRead` = GREATEST(`%s`.`PermissionsRead`, VALUES(`PermissionsRead`)), "+
			"`PermissionsCreate` = GREATEST(`%s`.`PermissionsCreate`, VALUES(`PermissionsCreate`)), "+
			"`PermissionsEdit` = GREATEST(`%s`.`PermissionsEdit`, VALUES(`PermissionsEdit`)), "+
			"`PermissionsDelete` = GREATEST(`%s`.`PermissionsDelete`, VALUES(`PermissionsDelete`)), "+
			"`PermissionsViewAllRecords` = GREATEST(`%s`.`PermissionsViewAllRecords`, VALUES(`PermissionsViewAllRecords`)), "+
			"`PermissionsModifyAllRecords` = GREATEST(`%s`.`PermissionsModifyAllRecords`, VALUES(`PermissionsModifyAllRecords`)), "+
			"`IsDeleted` = LEAST(`%s`.`IsDeleted`, VALUES(`IsDeleted`)), "+
			"`LastModifiedDate` = VALUES(`LastModifiedDate`)",
		constants.TableObjectPermissions, constants.TableObjectPermissions, tempMap,
		constants.TableObjectPermissions, constants.TableObjectPermissions, constants.TableObjectPermissions,
		constants.TableObjectPermissions, constants.TableObjectPermissions, constants.TableObjectPermissions,
		constants.TableObjectPermissions,
	)
	if _, err := rawDB.Exec(mergeSQL); err != nil {
		return fmt.Errorf("merging ObjectPermissions for PermissionSet dedup: %w", err)
	}
	if _, err := rawDB.Exec(fmt.Sprintf(
		"DELETE op FROM `%s` op INNER JOIN `%s` m ON op.`ParentId` = m.`LoserId`",
		constants.TableObjectPermissions, tempMap,
	)); err != nil {
		return fmt.Errorf("deleting loser ObjectPermissions rows: %w", err)
	}
	return nil
}

func mergeFieldPermissionsForPermissionSetDedup(rawDB *sql.DB, tempMap string) error {
	mergeSQL := fmt.Sprintf(
		"INSERT INTO `%s` (`Id`, `ParentId`, `SobjectType`, `Field`, `PermissionsRead`, `PermissionsEdit`, `IsDeleted`, `CreatedDate`, `LastModifiedDate`) "+
			"SELECT UUID(), m.`WinnerId`, fp.`SobjectType`, fp.`Field`, fp.`PermissionsRead`, fp.`PermissionsEdit`, fp.`IsDeleted`, fp.`CreatedDate`, NOW() "+
			"FROM `%s` fp INNER JOIN `%s` m ON fp.`ParentId` = m.`LoserId` "+
			"ON DUPLICATE KEY UPDATE "+
			"`PermissionsRead` = GREATEST(`%s`.`PermissionsRead`, VALUES(`PermissionsRead`)), "+
			"`PermissionsEdit` = GREATEST(`%s`.`PermissionsEdit`, VALUES(`PermissionsEdit`)), "+
			"`IsDeleted` = LEAST(`%s`.`IsDeleted`, VALUES(`IsDeleted`)), "+
			"`LastModifiedDate` = VALUES(`LastModifiedDate`)",
		constants.TableFieldPermissions, constants.TableFieldPermissions, tempMap,
		constants.TableFieldPermissions, constants.TableFieldPermissions, constants.TableFieldPermissions,
	)
	if _, err := rawDB.Exec(mergeSQL); err != nil {
		return fmt.Errorf("merging FieldPermissions for PermissionSet dedup: %w", err)
	}
	if _, err := rawDB.Exec(fmt.Sprintf(
		"DELETE fp FROM `%s` fp INNER JOIN `%s` m ON fp.`ParentId` = m.`LoserId`",
		constants.TableFieldPermissions, tempMap,
	)); err != nil {
		return fmt.Errorf("deleting loser FieldPermissions rows: %w", err)
	}
	return nil
}

// ---------------------------------------------------------------------------
// reconcileSystemTableIndices — generic, safe, additive-only index reconciler
// ---------------------------------------------------------------------------

// reconcileSystemTableIndices adds missing indices defined in system_tables.json
// to existing physical tables. This is a safe, additive reconciler:
//   - Skips tables that don't exist yet (fresh-DB safe)
//   - For non-unique indices: adds, logs on failure (non-fatal)
//   - For unique indices: adds, returns error on failure (fatal)
//   - Handles non-unique → unique upgrade (drop + add)
//   - Idempotent: tolerates "Duplicate key name" from concurrent pods
//   - Does NOT mutate data — data migrations must run first
//
// Caller must hold the schema reconciliation advisory lock.
type reconciledIndexInfo struct {
	name   string
	unique bool
}

func reconcileSystemTableIndices(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	log.Println("🔄 Reconciling system table indices...")
	start := time.Now()

	existingTables, err := loadExistingTableSet(rawDB)
	if err != nil {
		return fmt.Errorf("loading existing tables: %w", err)
	}
	if err := dropObsoleteIndices(rawDB, existingTables); err != nil {
		return fmt.Errorf("drop obsolete indices: %w", err)
	}

	// Load all indices in one query
	rows, err := rawDB.Query(
		`SELECT TABLE_NAME, INDEX_NAME, NON_UNIQUE, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
		 FROM INFORMATION_SCHEMA.STATISTICS
		 WHERE TABLE_SCHEMA = DATABASE()
		 GROUP BY TABLE_NAME, INDEX_NAME, NON_UNIQUE`,
	)
	if err != nil {
		return fmt.Errorf("query INFORMATION_SCHEMA.STATISTICS: %w", err)
	}
	defer rows.Close()

	existingIndices := make(map[string]map[string]reconciledIndexInfo)
	for rows.Next() {
		var tableName, indexName, cols string
		var nonUnique int
		if err := rows.Scan(&tableName, &indexName, &nonUnique, &cols); err != nil {
			return fmt.Errorf("scanning index metadata: %w", err)
		}
		key := strings.ToLower(tableName)
		if existingIndices[key] == nil {
			existingIndices[key] = make(map[string]reconciledIndexInfo)
		}
		existingIndices[key][strings.ToLower(cols)] = reconciledIndexInfo{name: indexName, unique: nonUnique == 0}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterating index metadata: %w", err)
	}

	alterCount := 0
	for _, def := range tableDefs {
		tableKey := strings.ToLower(def.TableName)
		if !existingTables[tableKey] {
			continue // table not created yet — Phase 1 will handle it
		}
		tableIndices := existingIndices[tableKey]

		indices := def.Indices
		if primary, ok := connectedDeliveryColumnPrimaryIndex(def); ok {
			indices = append([]schema.IndexDefinition{primary}, def.Indices...)
		}
		for _, idx := range indices {
			// These readiness-critical indexes require exact canonical name,
			// uniqueness, ordered columns, prefix rejection, fatal DDL handling,
			// and persisted readback. The dedicated reconciler is its single
			// lifecycle owner; the generic composition-key reconciler must not
			// accept or mutate it through a second path.
			if isConnectorWorkItemProjectionOrderIndex(def.TableName, idx) ||
				isRecordEventExecutionScopeIndex(def.TableName, idx) {
				continue
			}
			lowerCols := make([]string, len(idx.Columns))
			for i, c := range idx.Columns {
				lowerCols[i] = strings.ToLower(c)
			}
			colKey := strings.Join(lowerCols, ",")

			desiredName := idx.Name
			if desiredName == "" {
				desiredName = fmt.Sprintf("idx_%s_%s", def.TableName, strings.Join(idx.Columns, "_"))
			}
			columnList := "`" + strings.Join(idx.Columns, "`, `") + "`"

			if idx.Primary {
				if existingPrimaryCols, ok := primaryIndexColumns(tableIndices); ok {
					if existingPrimaryCols == colKey {
						continue
					}
					return fmt.Errorf("primary key drift on %s: existing PRIMARY(%s), expected PRIMARY(%s)",
						def.TableName, existingPrimaryCols, colKey)
				}

				ddl := fmt.Sprintf("ALTER TABLE `%s` ADD PRIMARY KEY (%s)", def.TableName, columnList)
				log.Printf("   ➕ Adding primary key: %s (%s)", def.TableName, strings.Join(idx.Columns, ", "))
				if _, err := rawDB.Exec(ddl); err != nil {
					if isDuplicatePrimaryKeyError(err) {
						log.Printf("   ↩ Primary key already exists (concurrent pod) — skipping")
						continue
					}
					return fmt.Errorf("adding primary key on %s(%s): %w",
						def.TableName, strings.Join(idx.Columns, ", "), err)
				}
				alterCount++
				continue
			}

			if existing, ok := tableIndices[colKey]; ok {
				if idx.Unique && !existing.unique {
					// Non-unique → unique upgrade: drop old, add unique.
					// Tolerate races: another pod may have already completed this upgrade.
					log.Printf("   🔄 Upgrading index %s on %s to unique...", existing.name, def.TableName)
					if _, err := rawDB.Exec(fmt.Sprintf("ALTER TABLE `%s` DROP INDEX `%s`",
						def.TableName, existing.name)); err != nil {
						if strings.Contains(err.Error(), "check that column/key exists") ||
							strings.Contains(err.Error(), "Error 1091") {
							// Index already dropped by another pod — continue to add unique
							log.Printf("   ↩ Index %s already dropped (concurrent pod)", existing.name)
						} else {
							return fmt.Errorf("dropping non-unique index %s on %s: %w", existing.name, def.TableName, err)
						}
					}
					if _, err := rawDB.Exec(fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE INDEX `%s` (%s)",
						def.TableName, desiredName, columnList)); err != nil {
						if isDuplicateIndexNameError(err) {
							log.Printf("   ↩ Unique index already exists (concurrent pod) — skipping")
						} else {
							return fmt.Errorf("adding unique index on %s(%s): %w",
								def.TableName, strings.Join(idx.Columns, ", "), err)
						}
					} else {
						alterCount++
					}
				}
				continue
			}

			// Index missing (by column composition) — add it.
			// But first: if an index with the same NAME exists with different columns
			// (e.g., columns were added/removed), drop the old one first.
			var ddl string
			if idx.Unique {
				ddl = fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE INDEX `%s` (%s)", def.TableName, desiredName, columnList)
			} else {
				ddl = fmt.Sprintf("ALTER TABLE `%s` ADD INDEX `%s` (%s)", def.TableName, desiredName, columnList)
			}
			log.Printf("   ➕ Adding %sindex: %s (%s)",
				map[bool]string{true: "unique ", false: ""}[idx.Unique],
				def.TableName, strings.Join(idx.Columns, ", "))

			if _, err := rawDB.Exec(ddl); err != nil {
				// "Duplicate key name" means an index with the same name exists but
				// different columns. Drop the old one and retry.
				if isDuplicateIndexNameError(err) {
					log.Printf("   🔄 Index %s exists with different columns — dropping and re-creating", desiredName)
					if _, dropErr := rawDB.Exec(fmt.Sprintf("ALTER TABLE `%s` DROP INDEX `%s`", def.TableName, desiredName)); dropErr != nil {
						log.Printf("   ⚠️  Failed to drop old index %s: %v", desiredName, dropErr)
						continue
					}
					if _, retryErr := rawDB.Exec(ddl); retryErr != nil {
						// CRITICAL: old index was dropped but new one failed (e.g., duplicate data).
						// For unique indices this is dangerous — table has no constraint.
						if idx.Unique {
							return fmt.Errorf("CRITICAL: dropped old unique index %s on %s but failed to create replacement — table has no unique constraint: %w",
								desiredName, def.TableName, retryErr)
						}
						log.Printf("   ⚠️  Failed to re-create index %s after drop: %v", desiredName, retryErr)
						continue
					}
					alterCount++
					log.Printf("   ✅ Index %s re-created with updated columns", desiredName)
					continue
				}
				if idx.Unique {
					return fmt.Errorf("adding unique index on %s(%s): %w",
						def.TableName, strings.Join(idx.Columns, ", "), err)
				}
				log.Printf("   ⚠️  Failed to add index on %s(%s): %v (non-critical)",
					def.TableName, strings.Join(idx.Columns, ", "), err)
			} else {
				alterCount++
			}
		}
	}
	log.Printf("✅ Index reconciliation complete (%d alterations, %v)", alterCount, time.Since(start).Round(time.Millisecond))
	return nil
}

type obsoleteIndexSpec struct {
	tableName string
	indexName string
}

func dropObsoleteIndices(rawDB *sql.DB, existingTables map[string]bool) error {
	obsolete := []obsoleteIndexSpec{
		{tableName: "_System_File", indexName: "uk__System_File_compat_projection"},
		{tableName: "_System_File", indexName: "idx__System_File_parent_active"},
		{tableName: "_System_File", indexName: "idx__System_File_parent_kind"},
		{tableName: "_System_File", indexName: "idx__System_File_deleted"},
		{tableName: "_System_FileUploadOperation", indexName: "idx__System_FileUploadOperation_state_expires"},
		{tableName: "_System_FileUploadOperation", indexName: "idx__System_FileUploadOperation_parent"},
		{tableName: "_System_FileUploadOperation", indexName: "idx__System_FileUploadOperation_system_file"},
		// PR #1618 (NCRM-INT-11b) changed the unique key on
		// _System_ConnectorInboundRoute from (InstalledPackId, EventKind)
		// to (InstalledPackId, EventKind, Version). The reconciler's generic
		// ADD-missing-index logic creates the new 3-column unique key, but does
		// NOT drop the old 2-column one. Without this explicit drop, upgraded
		// orgs end up with BOTH unique indices, and INSERTing Version=2 violates
		// the old 2-column unique constraint (same InstalledPackId+EventKind pair).
		//
		// Idempotence note (B13 fix, codex R11): the new non-unique latest-version
		// lookup index on (InstalledPackId, EventKind) is given an explicit name
		// "idx__System_ConnectorInboundRoute_latest_version_lookup" in
		// system_tables.json. The reconciler honours idx.Name when non-empty, so it
		// never auto-generates the legacy name. The drop entry below targets only the
		// auto-generated legacy name, which exists on upgraded orgs and is never
		// recreated. Subsequent setup-apply runs see dropIndexIfExists return nil
		// (isMissingIndexError) and the reconciler sees an already-present explicit-
		// name index — no DDL churn.
		{tableName: "_System_ConnectorInboundRoute", indexName: "idx__System_ConnectorInboundRoute_InstalledPackId_EventKind"},
		// PR #1670 scoped connector callout profiles by PackInstallId. The old
		// unique key used the same explicit name on (OwnerPack, ProfileName), and
		// would block inserting the new install-scoped snapshot for the same pack
		// and profile on upgraded orgs.
		{tableName: "_System_ConnectorCalloutProfile", indexName: "ux_connector_callout_profile"},
		// Source-backed email_template ownership is keyed by (OwnerPack,
		// DeveloperName). Older schemas enforced global Name uniqueness, which
		// prevents two app packs from shipping same-named templates and is not the
		// durable read path for sendEmail.
		{tableName: "_System_EmailTemplate", indexName: "Name"},
		{tableName: "_System_EmailTemplate", indexName: "uniq__System_EmailTemplate_Name"},
	}

	for _, spec := range obsolete {
		if !existingTables[strings.ToLower(spec.tableName)] {
			continue
		}
		if err := dropIndexIfExists(rawDB, spec.tableName, spec.indexName); err != nil {
			return err
		}
	}
	return nil
}

func dropIndexIfExists(rawDB *sql.DB, tableName, indexName string) error {
	ddl := fmt.Sprintf("ALTER TABLE `%s` DROP INDEX `%s`", tableName, indexName)
	if _, err := rawDB.Exec(ddl); err != nil {
		if isMissingIndexError(err, indexName) {
			return nil
		}
		return err
	}
	log.Printf("   🧹 Dropped obsolete index: %s.%s", tableName, indexName)
	return nil
}

func isMissingIndexError(err error, indexName string) bool {
	if err == nil {
		return false
	}
	lower := strings.ToLower(err.Error())
	return strings.Contains(lower, "check that column/key exists") ||
		strings.Contains(lower, "error 1091") ||
		strings.Contains(lower, fmt.Sprintf("index %s doesn't exist", strings.ToLower(indexName)))
}

func isDuplicateIndexNameError(err error) bool {
	if err == nil {
		return false
	}
	lower := strings.ToLower(err.Error())
	return strings.Contains(lower, "duplicate key name") ||
		strings.Contains(lower, "index already exist") ||
		strings.Contains(lower, "already exist")
}

func isDuplicatePrimaryKeyError(err error) bool {
	if err == nil {
		return false
	}
	lower := strings.ToLower(err.Error())
	return strings.Contains(lower, "multiple primary key defined") ||
		strings.Contains(lower, "duplicate key name 'primary'") ||
		strings.Contains(lower, "duplicate key name \"primary\"")
}

func primaryIndexColumns(indices map[string]reconciledIndexInfo) (string, bool) {
	for cols, idx := range indices {
		if strings.EqualFold(idx.name, "PRIMARY") {
			return cols, true
		}
	}
	return "", false
}

// loadExistingTableSet returns a set of lowercase table names that exist in the current database.
// Returns an error if the metadata query fails — callers must not silently treat
// a lookup failure as "no tables exist" (that would skip critical migrations).
func loadExistingTableSet(rawDB *sql.DB) (map[string]bool, error) {
	tables := make(map[string]bool)
	rows, err := rawDB.Query("SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE()")
	if err != nil {
		return nil, fmt.Errorf("querying INFORMATION_SCHEMA.TABLES: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, fmt.Errorf("scanning table name: %w", err)
		}
		tables[strings.ToLower(name)] = true
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterating table names: %w", err)
	}
	return tables, nil
}

// reconcileSystemFieldMetadata ensures all field metadata defined in managed schema metadata
// exist as rows in _System_Field. Runs OUTSIDE the main transaction so a partial
// failure does not prevent field registration.
// Optimized: uses 2 bulk queries instead of 2*N per-table queries (~18s → ~200ms).
func reconcileSystemFieldMetadata(rawDB *sql.DB, repo *persistence.SchemaRepository, tableDefs []schema.TableDefinition) error {
	log.Println("🔄 Reconciling system field metadata...")
	start := time.Now()

	schemaMgr := metadata.NewSchemaManager(repo)

	// BULK QUERY 1: Load ALL objectIDs in one query (ApiName → Id)
	objectIDs := make(map[string]string)
	objRows, err := rawDB.Query(
		fmt.Sprintf("SELECT `%s`, `%s` FROM `%s`",
			constants.FieldAPIName, constants.ColID, constants.TableObject),
	)
	if err != nil {
		return fmt.Errorf("bulk-load object IDs: %w", err)
	}
	defer objRows.Close()
	for objRows.Next() {
		var apiName, id string
		if err := objRows.Scan(&apiName, &id); err != nil {
			return fmt.Errorf("scan object ID row: %w", err)
		}
		objectIDs[apiName] = id
	}
	if err := objRows.Err(); err != nil {
		return fmt.Errorf("iterate object IDs: %w", err)
	}

	// BULK QUERY 2: Load ALL registered field ApiNames grouped by ObjectId
	registeredFields := make(map[string]map[string]bool) // ObjectId → set of ApiNames
	fRows, err := rawDB.Query(
		fmt.Sprintf("SELECT `%s`, `%s` FROM `%s`",
			constants.FieldObjectID, constants.FieldAPIName, constants.TableField),
	)
	if err != nil {
		return fmt.Errorf("bulk-load field metadata: %w", err)
	}
	defer fRows.Close()
	for fRows.Next() {
		var objID, apiName string
		if err := fRows.Scan(&objID, &apiName); err != nil {
			return fmt.Errorf("scan field metadata row: %w", err)
		}
		if registeredFields[objID] == nil {
			registeredFields[objID] = make(map[string]bool)
		}
		registeredFields[objID][apiName] = true
	}
	if err := fRows.Err(); err != nil {
		return fmt.Errorf("iterate field metadata: %w", err)
	}

	// Now iterate table definitions purely in-memory — zero SQL
	var allMissing []schema.FieldWithContext
	for _, def := range tableDefs {
		objectID, ok := objectIDs[def.TableName]
		if !ok {
			continue
		}
		registered := registeredFields[objectID]
		for _, col := range def.Columns {
			if registered[col.Name] {
				continue
			}
			fc := schemaMgr.PrepareFieldForBatch(def.TableName, col)
			fc.ObjectID = objectID
			// Provenance derived at read time by fieldprovenance.ClassifySource
			// from system_tables.json. Issue #1582 RFC v1.6.
			allMissing = append(allMissing, fc)
		}
	}

	if len(allMissing) == 0 {
		log.Printf("✅ Field metadata reconciliation complete (no missing fields, %v)", time.Since(start).Round(time.Millisecond))
		return nil
	}

	log.Printf("   ➕ Adding %d missing field(s) across all system tables...", len(allMissing))
	if err := repo.BatchSaveFieldMetadata(context.Background(), allMissing, nil); err != nil {
		return fmt.Errorf("register missing fields: %w", err)
	}
	log.Printf("   ✅ Registered %d missing field(s)", len(allMissing))

	log.Printf("✅ Field metadata reconciliation complete (%v)", time.Since(start).Round(time.Millisecond))
	return nil
}

func managedObjectPredicate(tableDefs []schema.TableDefinition, objectExpr string) (string, []interface{}) {
	if len(tableDefs) == 0 {
		return "1 = 0", nil
	}
	placeholders := make([]string, 0, len(tableDefs))
	args := make([]interface{}, 0, len(tableDefs))
	for _, def := range tableDefs {
		if strings.TrimSpace(def.TableName) == "" {
			continue
		}
		placeholders = append(placeholders, "?")
		args = append(args, strings.ToLower(def.TableName))
	}
	if len(placeholders) == 0 {
		return "1 = 0", nil
	}
	return fmt.Sprintf("LOWER(%s) IN (%s)", objectExpr, strings.Join(placeholders, ", ")), args
}

func managedFieldTuplePredicate(tableDefs []schema.TableDefinition, objectExpr, fieldExpr string) (string, []interface{}) {
	placeholders := make([]string, 0)
	args := make([]interface{}, 0)
	for _, def := range tableDefs {
		if strings.TrimSpace(def.TableName) == "" {
			continue
		}
		for _, col := range def.Columns {
			if strings.TrimSpace(col.Name) == "" {
				continue
			}
			placeholders = append(placeholders, "(?, ?)")
			args = append(args, strings.ToLower(def.TableName), col.Name)
		}
	}
	if len(placeholders) == 0 {
		return "1 = 0", nil
	}
	return fmt.Sprintf("(LOWER(%s), %s) IN (%s)", objectExpr, fieldExpr, strings.Join(placeholders, ", ")), args
}

type schemaBootstrapMode struct {
	label                       string
	includeSchemaReconciliation bool
	includeManagedMetadata      bool
	scopeManagedMetadata        bool
}

var (
	schemaBootstrapFull = schemaBootstrapMode{
		label:                       "core system schema",
		includeSchemaReconciliation: true,
		includeManagedMetadata:      true,
	}
	schemaBootstrapL0 = schemaBootstrapMode{
		label:                       "L0 system schema and metadata",
		includeSchemaReconciliation: true,
		includeManagedMetadata:      true,
		scopeManagedMetadata:        true,
	}
	schemaBootstrapL0DDLOnly = schemaBootstrapMode{
		label:                       "L0 system schema DDL",
		includeSchemaReconciliation: false,
		includeManagedMetadata:      false,
	}
)

// InitializeSchema creates core system tables using SchemaManager.
// This replaces raw SQL execution with declarative table definitions.
func InitializeSchema(db *database.TiDBConnection) error {
	return initializeSchema(db, schemaBootstrapFull)
}

// EnsureL0SchemaDDL creates only the L0/system physical tables needed to run
// clean-L0 contamination checks. It deliberately does not seed managed metadata
// rows or run app-shaped compatibility phases.
func EnsureL0SchemaDDL(db *database.TiDBConnection) error {
	return initializeSchema(db, schemaBootstrapL0DDLOnly)
}

// InitializeSchemaL0 converges L0/system schema metadata without installing
// application-pack schema or metadata.
func InitializeSchemaL0(db *database.TiDBConnection) error {
	return initializeSchema(db, schemaBootstrapL0)
}

func initializeSchema(db *database.TiDBConnection, mode schemaBootstrapMode) error {
	log.Printf("🔧 Initializing %s...", mode.label)
	rawDB := db.DB()

	repo := persistence.NewSchemaRepository(rawDB, logging.Default)
	schemaMgr := metadata.NewSchemaManager(repo)

	// Get all managed table definitions
	tableDefs := GetManagedTableDefinitions()

	if err := ensureBootstrapStateTable(rawDB); err != nil {
		return fmt.Errorf("ensure bootstrap state table: %w", err)
	}

	lastState, err := loadBootstrapState(rawDB)
	if err != nil {
		return fmt.Errorf("load bootstrap state: %w", err)
	}

	runID := fmt.Sprintf("schema-%d", time.Now().UTC().UnixNano())
	resumeFrom := ""
	if lastState != nil && strings.EqualFold(lastState.Status, "failed") && lastState.Phase != "" {
		resumeFrom = lastState.Phase
		log.Printf("🔁 Resuming schema bootstrap from phase=%s (previous run=%s)", lastState.Phase, lastState.RunID)
	}
	if err := persistBootstrapState(rawDB, runID, "start", "running", nil); err != nil {
		return fmt.Errorf("persist bootstrap start state: %w", err)
	}

	var currentPhase = "start"
	runPhase := func(phase string, fn func() error) error {
		currentPhase = phase
		if !shouldRunBootstrapPhase(resumeFrom, phase) {
			log.Printf("⏭️ Skipping completed bootstrap phase: %s", phase)
			return nil
		}
		if err := persistBootstrapState(rawDB, runID, phase, "running", nil); err != nil {
			return fmt.Errorf("persist bootstrap phase %s: %w", phase, err)
		}
		if err := fn(); err != nil {
			if persistErr := persistBootstrapState(rawDB, runID, phase, "failed", err); persistErr != nil {
				log.Printf("⚠️ Failed to persist bootstrap failure state for phase %s: %v (original error: %v)", phase, persistErr, err)
			}
			return err
		}
		return nil
	}

	if err := runBootstrapMutationUnderLock(rawDB, func() error {
		coreTables := []string{constants.TableTable, constants.TableObject, constants.TableField}
		var remainingDefs []schema.TableDefinition
		for _, def := range tableDefs {
			isCore := false
			for _, core := range coreTables {
				if def.TableName == core {
					isCore = true
					break
				}
			}
			if !isCore {
				remainingDefs = append(remainingDefs, def)
			}
		}

		if err := runPhase("phase0_columns_reconcile", func() error {
			return ReconcileTableColumns(rawDB, tableDefs)
		}); err != nil {
			return err
		}
		if mode.includeSchemaReconciliation {
			if err := runPhase("phase0_schema_reconcile", func() error {
				return runSchemaReconciliationUnderLock(rawDB, tableDefs)
			}); err != nil {
				return fmt.Errorf("schema reconciliation failed: %w", err)
			}
		}

		if err := runPhase("phase1_core_tables", func() error {
			for _, coreName := range coreTables {
				for _, def := range tableDefs {
					if def.TableName == coreName {
						log.Printf("🧱 Pre-creating physical table: %s", coreName)
						if err := schemaMgr.CreatePhysicalTable(context.Background(), def); err != nil {
							log.Printf("⚠️  Warning pre-creating %s: %v", coreName, err)
							return err
						}
						break
					}
				}
			}
			return nil
		}); err != nil {
			return err
		}

		if err := runPhase("phase1_batch_tables", func() error {
			log.Println("⚡️ Creating tables and registering (Super Batch)...")
			if err := schemaMgr.BatchCreatePhysicalTables(context.Background(), remainingDefs); err != nil {
				log.Printf("   ⚠️  Batch physical table creation failed: %v", err)
				return err
			}
			return nil
		}); err != nil {
			return err
		}

		// Runtime-critical physical projections belong to the physical schema
		// lifecycle, not managed metadata or a full Apply-only reconciliation.
		// Run outside resume-skippable phases after tables/columns exist, so the
		// DDL-only recovery entrypoint and an already-created index both converge
		// the same derived invariant before any Runtime producer can start.
		if err := withSchemaReconciliationLock(rawDB, func() error {
			return reconcileRecordEventExecutionScopeProjection(
				rawDB, tableDefs, mode.includeManagedMetadata,
			)
		}); err != nil {
			return fmt.Errorf("runtime-critical physical projection reconciliation: %w", err)
		}

		if mode.includeSchemaReconciliation {
			// Run outside resume-skippable phases. A prior failure after phase 1
			// causes resume to skip both column reconciliation (which cannot create
			// absent tables) and batch table creation. The pinned persistence
			// CreatePhysicalTable contract owns both CREATE TABLE IF NOT EXISTS and
			// catalog-declared managed physical evolution. Replaying it therefore
			// creates missing Receipt tables and performs the guarded retained-row
			// ReceiptProtocolVersion narrowing on an existing parent. Either unsafe
			// retained data or an incompatible physical shape is returned as an
			// apply error; it is never reduced to readiness-only drift.
			if err := reconcileConnectorV2IngressReceiptAuthorityTables(
				context.Background(),
				schemaMgr,
				tableDefs,
			); err != nil {
				return fmt.Errorf("Connector V2 ingress receipt authority table reconciliation: %w", err)
			}
			// Require exact readback in this same apply run. This closes the
			// bootstrap-resume boundary: even if an evolution implementation ever
			// returns without converging every column/index/FK, completion cannot be
			// persisted while runtime readiness remains blocked.
			if err := verifyConnectorV2IngressReceiptAuthoritySchema(rawDB); err != nil {
				return fmt.Errorf("Connector V2 ingress receipt authority post-reconciliation verification: %w", err)
			}

			// Run outside resume-skippable phases so a prior failure in a later
			// phase cannot strand this runtime-critical index. Phase 1 has now
			// created the table on a fresh database; warm databases converge via
			// additive DDL. Reuse the schema advisory lock and require an exact
			// persisted readback before any managed metadata or activation work.
			if err := withSchemaReconciliationLock(rawDB, func() error {
				return reconcileConnectorWorkItemProjectionOrderIndex(rawDB, tableDefs)
			}); err != nil {
				return fmt.Errorf("connector work-item projection-order index reconciliation: %w", err)
			}
		}

		if !mode.includeManagedMetadata {
			return nil
		}

		// InitializeSchema and InitializeSchemaL0 are public managed-schema
		// entry points, not implementation details of ApplyMetadata. Therefore
		// the managed-schema boundary itself owns this prerequisite. Establish
		// the singleton only after its physical table exists, then bind the
		// fail-closed coordinator before the first protected metadata write.
		// EnsureL0SchemaDDL returns above and remains a pre-authority recovery
		// operation that never creates the singleton.
		if err := persistencewiring.EnsurePermissionStateInitialized(context.Background(), db); err != nil {
			return err
		}
		directWriters, err := persistencewiring.NewPermissionStateDirectWriters(db)
		if err != nil {
			return err
		}
		if err := directWriters.BindSchema(repo); err != nil {
			return err
		}

		if err := runPhase("phase2_object_metadata", func() error {
			log.Println("📋 Registering system object metadata (BATCH mode)...")
			var allObjects []*models.ObjectMetadata
			for _, def := range tableDefs {
				objectID := schema.GenerateObjectID(def.TableName)
				isCustom := constants.TableType(def.TableType) == constants.TableTypeCustomObject
				label := def.Label
				if label == "" {
					label = def.Description
				}
				if label == "" {
					label = def.TableName
				}
				if len(label) > 255 {
					return fmt.Errorf(
						"managed object %s resolved label length %d exceeds _System_Object.Label varchar(255); declare a shorter label in metadata",
						def.TableName,
						len(label),
					)
				}
				description := def.Description
				pluralSource := def.Label
				if pluralSource == "" {
					pluralSource = def.TableName
				}
				sharingModel := def.SharingModel
				if sharingModel == "" {
					sharingModel = defaultSharingModel(def.TableType, def.TableName)
				}
				allObjects = append(allObjects, &models.ObjectMetadata{
					ID:                      objectID,
					APIName:                 def.TableName,
					KeyPrefix:               optionalString(def.KeyPrefix),
					Label:                   label,
					PluralLabel:             Pluralize(pluralSource),
					Description:             &description,
					IsCustom:                isCustom,
					SharingModel:            models.SharingModel(sharingModel),
					EnableHierarchySharing:  def.EnableHierarchySharing,
					RequiredWritePermission: def.RequiredWritePermission,
					RequiredReadPermission:  def.RequiredReadPermission,
					AccessPolicy:            def.AccessPolicy,
					DeleteMode:              def.DeleteMode,
					Searchable:              true,
				})
			}
			log.Printf("   📦 Batch inserting %d objects...", len(allObjects))
			if err := schemaMgr.BatchSaveObjectMetadata(context.Background(), allObjects, nil); err != nil {
				log.Printf("   ⚠️  Batch object insert failed: %v", err)
				return err
			}
			return nil
		}); err != nil {
			return err
		}

		realObjectIDs := make(map[string]string)
		if err := runPhase("phase2_read_object_ids", func() error {
			idRows, err := rawDB.Query(
				fmt.Sprintf("SELECT `%s`, `%s` FROM `%s`",
					constants.FieldAPIName, constants.ColID, constants.TableObject),
			)
			if err != nil {
				log.Printf("   ⚠️  Failed to read back object IDs: %v", err)
				return err
			}
			for idRows.Next() {
				var apiName, id string
				if err := idRows.Scan(&apiName, &id); err == nil {
					realObjectIDs[apiName] = id
				}
			}
			idRows.Close()
			if err := idRows.Err(); err != nil {
				log.Printf("   ⚠️  Error iterating object IDs: %v", err)
				return err
			}
			return nil
		}); err != nil {
			return err
		}

		if err := runPhase("phase2_field_metadata", func() error {
			var allFields []metadata.FieldWithContext
			for _, def := range tableDefs {
				realObjID, ok := realObjectIDs[def.TableName]
				if !ok {
					log.Printf("   ⚠️  Object %s has no DB ID, skipping field registration", def.TableName)
					continue
				}
				for _, col := range def.Columns {
					fc := schemaMgr.PrepareFieldForBatch(def.TableName, col)
					fc.ObjectID = realObjID
					// Provenance derived at read time by fieldprovenance.ClassifySource
					// from system_tables.json. Issue #1582 RFC v1.6.
					allFields = append(allFields, fc)
				}
			}
			log.Printf("   📦 Batch inserting %d fields...", len(allFields))
			if err := schemaMgr.BatchSaveFieldMetadata(context.Background(), allFields, nil); err != nil {
				log.Printf("   ⚠️  Batch field insert failed: %v", err)
				return err
			}
			return nil
		}); err != nil {
			return err
		}

		if err := runPhase("phase3_field_reconcile", func() error {
			return reconcileSystemFieldMetadata(rawDB, repo, tableDefs)
		}); err != nil {
			return err
		}
		if err := runPhase("phase4_name_fields", func() error {
			return ensurePluralLabelsAndNameFieldsScoped(rawDB, tableDefs, mode.scopeManagedMetadata)
		}); err != nil {
			return err
		}
		// phase5_auto_numbers intentionally removed (Round-5 Codex
		// V1-P1 / V2-C1). Seeding and verification of
		// `_System_AutoNumberCounter` now run inside
		// the app-pack apply path AFTER `applyAutoNumberFormats`
		// populates `_System_Field`. That
		// order is required: the former phase5 ran BEFORE packs
		// populated the columns it tried to read, so on a fresh DB
		// `loadAutoNumberSeedsFromDB` returned an empty slice and the
		// seed was a no-op. Moving the seed+verify into packs.Apply
		// also lets us pass the generator's full seed slice (with
		// LegacySource preserved) directly — no lossy round-trip
		// through `_System_Field`.
		if err := runPhase("phase6_field_labels", func() error {
			return reconcileFieldLabelsScoped(rawDB, tableDefs, mode.scopeManagedMetadata)
		}); err != nil {
			return err
		}
		// Spec-declared defaults run before the boolean fallback: the fallback
		// writes a hard-coded 'false' that would otherwise satisfy the
		// NULL/empty filter on the next pass and permanently shadow any
		// spec-declared value that differs (e.g. Campaign.IsActive = '0').
		if err := runPhase("phase7_spec_defaults", func() error {
			return reconcileStandardFieldDefaultValuesScoped(rawDB, mode.scopeManagedMetadata)
		}); err != nil {
			return err
		}
		if err := runPhase("phase7b_boolean_fallback", func() error {
			return reconcileBooleanDefaultValuesScoped(rawDB, tableDefs, mode.scopeManagedMetadata)
		}); err != nil {
			return err
		}
		return nil
	}); err != nil {
		if persistErr := persistBootstrapState(rawDB, runID, currentPhase, "failed", err); persistErr != nil {
			log.Printf("⚠️ Failed to persist bootstrap failure state for phase %s: %v (original error: %v)", currentPhase, persistErr, err)
		}
		return err
	}

	if err := persistBootstrapState(rawDB, runID, "completed", "completed", nil); err != nil {
		return fmt.Errorf("persist bootstrap completed state: %w", err)
	}

	log.Printf("✅ %s initialized (optimized)", mode.label)
	return nil
}

func optionalString(value string) *string {
	if value == "" {
		return nil
	}
	return &value
}

// defaultSharingModel returns the fallback SharingModel for objects that do not
// declare one in managed metadata. System tables default to PublicReadWrite;
// everything else defaults to Private. Non-default object sharing models must be
// declared by the owning metadata pack instead of inferred from object names.
func defaultSharingModel(tableType string, _ string) string {
	switch tableType {
	case "system_core", "system_metadata", "system_data", "system_junction":
		return string(constants.SharingModelPublicReadWrite)
	default:
		return string(constants.SharingModelPrivate)
	}
}

// ReconcileManagedStandardObjectAccessPoliciesWithDefs reconciles authoritative
// access policy metadata for the supplied managed table definitions.
//
// This is the exported shim over the package-private
// reconcileManagedStandardObjectAccessPolicies helper. The control-plane
// access-policy reconcile funcs (relocated out of this file in wave5 s5d) call
// this shim so the helper body stays in bootstrap (it is co-resident with the
// reconcileLegacy.../HierarchySharing/AccessPolicyJSON private helpers and the
// schema-package SharingModel/HierarchySharing primitives) and is not duplicated
// across the package boundary.
//
// SharingModel retains legacy-baseline repair semantics so admin overrides are
// preserved while known discovered defaults are corrected. Other immutable
// policy fields such as EnableHierarchySharing are fully authoritative.
func ReconcileManagedStandardObjectAccessPoliciesWithDefs(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	return reconcileManagedStandardObjectAccessPolicies(rawDB, tableDefs)
}

// ReconcileManagedStandardObjectSharingModels is kept as a compatibility wrapper
// for older callers. It now reconciles the full managed access policy set.
func ReconcileManagedStandardObjectSharingModels(rawDB *sql.DB) error {
	return reconcileManagedStandardObjectAccessPolicies(rawDB, GetManagedTableDefinitions())
}

func reconcileManagedStandardObjectAccessPolicies(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	if err := reconcileLegacyManagedStandardObjectSharingModels(rawDB, tableDefs); err != nil {
		return err
	}
	if err := reconcileManagedStandardObjectHierarchySharing(rawDB, tableDefs); err != nil {
		return err
	}
	if err := reconcileManagedStandardObjectAccessPolicyJSON(rawDB, tableDefs); err != nil {
		return err
	}
	return nil
}

func reconcileLegacyManagedStandardObjectSharingModels(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	desiredByAPI := make(map[string]string)
	for _, def := range tableDefs {
		if def.TableType != string(constants.TableTypeStandardObject) || def.SharingModel == "" {
			continue
		}
		desiredByAPI[strings.ToLower(def.TableName)] = def.SharingModel
	}
	if len(desiredByAPI) == 0 {
		return nil
	}

	updated := 0
	for apiName, desired := range desiredByAPI {
		query := fmt.Sprintf(
			"UPDATE `%s` SET `%s` = ?, `%s` = NOW() WHERE LOWER(`%s`) = ? AND `%s` = ? AND (`%s` = 0 OR `%s` IS NULL) AND `%s` != ?",
			constants.TableObject,
			constants.FieldSysObject_SharingModel,
			constants.FieldLastModifiedDate,
			constants.FieldAPIName,
			constants.FieldSysObject_TableType,
			constants.FieldSysObject_IsCustomized,
			constants.FieldSysObject_IsCustomized,
			constants.FieldSysObject_SharingModel,
		)
		args := []interface{}{
			desired,
			apiName,
			string(constants.TableTypeStandardObject),
			desired,
		}
		result, err := rawDB.Exec(query, args...)
		if err != nil {
			return fmt.Errorf("reconcile %s SharingModel to %s: %w", apiName, desired, err)
		}
		if affected, err := result.RowsAffected(); err == nil {
			updated += int(affected)
		}
	}

	if updated > 0 {
		log.Printf("   🔒 Reconciled %d legacy standard object SharingModel values from managed metadata", updated)
	}
	return nil
}

func reconcileManagedStandardObjectHierarchySharing(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	updated := 0
	for _, def := range tableDefs {
		if def.TableType != string(constants.TableTypeStandardObject) {
			continue
		}

		result, err := rawDB.Exec(
			fmt.Sprintf(
				"UPDATE `%s` SET `%s` = ?, `%s` = NOW() WHERE LOWER(`%s`) = ? AND `%s` = ? AND COALESCE(`%s`, 0) != ?",
				constants.TableObject,
				constants.FieldSysObject_EnableHierarchySharing,
				constants.FieldLastModifiedDate,
				constants.FieldAPIName,
				constants.FieldSysObject_TableType,
				constants.FieldSysObject_EnableHierarchySharing,
			),
			def.EnableHierarchySharing,
			strings.ToLower(def.TableName),
			string(constants.TableTypeStandardObject),
			def.EnableHierarchySharing,
		)
		if err != nil {
			return fmt.Errorf("reconcile %s EnableHierarchySharing to %t: %w", def.TableName, def.EnableHierarchySharing, err)
		}
		if affected, err := result.RowsAffected(); err == nil {
			updated += int(affected)
		}
	}

	if updated > 0 {
		log.Printf("   🔒 Reconciled %d managed standard object EnableHierarchySharing values from managed metadata", updated)
	}
	return nil
}

func reconcileManagedStandardObjectAccessPolicyJSON(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	updated := 0
	for _, def := range tableDefs {
		if def.AccessPolicy == nil {
			continue
		}
		accessPolicy, err := managedAccessPolicyWithPackOwnedSidecars(rawDB, def)
		if err != nil {
			return err
		}
		accessPolicyJSON, err := json.Marshal(accessPolicy)
		if err != nil {
			return fmt.Errorf("marshal managed AccessPolicy for %s: %w", def.TableName, err)
		}
		result, err := rawDB.Exec(
			fmt.Sprintf(
				"UPDATE `%s` SET `%s` = CAST(? AS JSON), `%s` = NOW() "+
					"WHERE LOWER(`%s`) = ? AND `%s` = ? "+
					"AND (`%s` IS NULL OR CAST(`%s` AS CHAR) != ?)",
				constants.TableObject,
				constants.FieldSysObject_AccessPolicy,
				constants.FieldLastModifiedDate,
				constants.FieldAPIName,
				constants.FieldSysObject_TableType,
				constants.FieldSysObject_AccessPolicy,
				constants.FieldSysObject_AccessPolicy,
			),
			string(accessPolicyJSON),
			strings.ToLower(def.TableName),
			def.TableType,
			string(accessPolicyJSON),
		)
		if err != nil {
			return fmt.Errorf("reconcile %s AccessPolicy: %w", def.TableName, err)
		}
		if affected, err := result.RowsAffected(); err == nil {
			updated += int(affected)
		}
	}
	if updated > 0 {
		log.Printf("   🔒 Reconciled %d managed AccessPolicy values from managed metadata", updated)
	}
	return nil
}

func cloneObjectAccessPolicy(policy *models.ObjectAccessPolicy) (*models.ObjectAccessPolicy, error) {
	if policy == nil {
		return nil, nil
	}
	raw, err := json.Marshal(policy)
	if err != nil {
		return nil, err
	}
	var out models.ObjectAccessPolicy
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, err
	}
	return &out, nil
}

func managedAccessPolicyWithPackOwnedSidecars(rawDB *sql.DB, def schema.TableDefinition) (*models.ObjectAccessPolicy, error) {
	desired, err := cloneObjectAccessPolicy(def.AccessPolicy)
	if err != nil {
		return nil, fmt.Errorf("clone managed AccessPolicy for %s: %w", def.TableName, err)
	}
	if desired == nil {
		return nil, nil
	}

	var liveRaw, recordMutationPackName, lifecyclePackName, recordMutationPackValue, lifecyclePackValue sql.NullString
	err = rawDB.QueryRow(
		fmt.Sprintf(
			"SELECT `%s`, `%s`, `%s`, `%s`, `%s` FROM `%s` WHERE LOWER(`%s`) = ? AND `%s` = ? AND COALESCE(`%s`, 0) = 0 LIMIT 1",
			constants.FieldSysObject_AccessPolicy,
			constants.FieldSysObject_AccessPolicyRecordMutationPackName,
			constants.FieldSysObject_AccessPolicyLifecyclePackName,
			constants.FieldSysObject_AccessPolicyRecordMutationPackValue,
			constants.FieldSysObject_AccessPolicyLifecyclePackValue,
			constants.TableObject,
			constants.FieldAPIName,
			constants.FieldSysObject_TableType,
			constants.FieldIsDeleted,
		),
		strings.ToLower(def.TableName),
		def.TableType,
	).Scan(&liveRaw, &recordMutationPackName, &lifecyclePackName, &recordMutationPackValue, &lifecyclePackValue)
	if err == sql.ErrNoRows {
		return desired, nil
	}
	if err != nil {
		return nil, fmt.Errorf("query live AccessPolicy owner for %s: %w", def.TableName, err)
	}
	return schema.PreservePackOwnedAccessPolicySidecars(
		desired,
		schema.AccessPolicySidecarSnapshot{
			LiveRaw:                 liveRaw,
			RecordMutationPackName:  recordMutationPackName,
			RecordMutationPackValue: recordMutationPackValue,
			LifecyclePackName:       lifecyclePackName,
			LifecyclePackValue:      lifecyclePackValue,
		},
		schema.AccessPolicySidecarPreserveOptions{
			ObjectAPIName:          def.TableName,
			RequireOwnedProjection: true,
		},
	)
}

// ensurePluralLabelsAndNameFieldsScoped fixes two common metadata gaps:
//  1. Objects with empty PluralLabel get one derived from their Label (append "s").
//  2. Objects where no field has IsNameField=true get their "Name" field auto-marked.
func ensurePluralLabelsAndNameFieldsScoped(rawDB *sql.DB, tableDefs []schema.TableDefinition, scoped bool) error {
	log.Println("🔄 Ensuring PluralLabel and IsNameField for all objects...")
	var errs []string

	// --- PluralLabel ---
	// Derive PluralLabel from Label with basic English pluralization rules:
	//   - ends in "y" (preceded by consonant) → "ies" (Opportunity → Opportunities)
	//   - ends in "z" → double "z" + "es" (Quiz → Quizzes)
	//   - ends in "s", "x", "sh", "ch" → "es" (Address → Addresses)
	//   - otherwise → "s" (Account → Accounts)
	pluralWhere := fmt.Sprintf("`%s` = '' OR `%s` IS NULL", constants.FieldSysObject_PluralLabel, constants.FieldSysObject_PluralLabel)
	var pluralArgs []interface{}
	if scoped {
		objectPredicate, args := managedObjectPredicate(tableDefs, fmt.Sprintf("`%s`", constants.FieldAPIName))
		pluralWhere = "(" + pluralWhere + ") AND " + objectPredicate
		pluralArgs = args
	}
	pluralSQL := fmt.Sprintf(
		`UPDATE `+"`%s`"+` SET `+"`%s`"+` = CASE
			WHEN `+"`%s`"+` REGEXP '[^aeiou]y$' THEN CONCAT(LEFT(`+"`%s`"+`, CHAR_LENGTH(`+"`%s`"+`)-1), 'ies')
			WHEN `+"`%s`"+` REGEXP '[^z]z$' THEN CONCAT(`+"`%s`"+`, 'zes')
			WHEN `+"`%s`"+` REGEXP '(s|x|z|sh|ch)$' THEN CONCAT(`+"`%s`"+`, 'es')
			ELSE CONCAT(`+"`%s`"+`, 's')
		END, `+"`%s`"+` = NOW() WHERE %s`,
		constants.TableObject,
		constants.FieldSysObject_PluralLabel,
		constants.FieldSysObject_Label,
		constants.FieldSysObject_Label, constants.FieldSysObject_Label,
		constants.FieldSysObject_Label, constants.FieldSysObject_Label,
		constants.FieldSysObject_Label, constants.FieldSysObject_Label,
		constants.FieldSysObject_Label,
		constants.FieldLastModifiedDate,
		pluralWhere,
	)
	res, err := rawDB.Exec(pluralSQL, pluralArgs...)
	if err != nil {
		log.Printf("   ⚠️  Failed to backfill PluralLabel: %v", err)
		errs = append(errs, fmt.Sprintf("backfill PluralLabel: %v", err))
	} else if n, _ := res.RowsAffected(); n > 0 {
		log.Printf("   ✅ Backfilled PluralLabel for %d object(s)", n)
	}

	// --- IsNameField ---
	// For each object that has zero fields with IsNameField=true,
	// mark the "Name" field (if it exists) as the name field.
	nameFieldFilter := ""
	var nameFieldArgs []interface{}
	if scoped {
		predicate, args := managedFieldTuplePredicate(tableDefs, fmt.Sprintf("o.`%s`", constants.FieldAPIName), fmt.Sprintf("f.`%s`", constants.FieldAPIName))
		nameFieldFilter = "JOIN `" + constants.TableObject + "` o ON o.`" + constants.ColID + "` = f.`" + constants.FieldObjectID + "` "
		nameFieldArgs = args
		predicate = "AND " + predicate + " "
		nameFieldSQL := fmt.Sprintf(
			"UPDATE `%s` f %sSET f.`%s` = 1, f.`%s` = NOW() "+
				"WHERE f.`%s` = '%s' "+
				"%s"+
				"AND f.`%s` NOT IN ("+
				"  SELECT sub.`%s` FROM ("+
				"    SELECT `%s` FROM `%s` WHERE `%s` = 1"+
				"  ) sub"+
				")",
			constants.TableField, nameFieldFilter,
			constants.FieldSysField_IsNameField,
			constants.FieldLastModifiedDate,
			constants.FieldAPIName, constants.FieldName,
			predicate,
			constants.FieldObjectID,
			constants.FieldObjectID,
			constants.FieldObjectID, constants.TableField, constants.FieldSysField_IsNameField,
		)
		res, err = rawDB.Exec(nameFieldSQL, nameFieldArgs...)
	} else {
		nameFieldSQL := fmt.Sprintf(
			"UPDATE `%s` f SET f.`%s` = 1, f.`%s` = NOW() "+
				"WHERE f.`%s` = '%s' "+
				"AND f.`%s` NOT IN ("+
				"  SELECT sub.`%s` FROM ("+
				"    SELECT `%s` FROM `%s` WHERE `%s` = 1"+
				"  ) sub"+
				")",
			constants.TableField,
			constants.FieldSysField_IsNameField,
			constants.FieldLastModifiedDate,
			constants.FieldAPIName, constants.FieldName,
			constants.FieldObjectID,
			constants.FieldObjectID,
			constants.FieldObjectID, constants.TableField, constants.FieldSysField_IsNameField,
		)
		res, err = rawDB.Exec(nameFieldSQL)
	}
	if err != nil {
		log.Printf("   ⚠️  Failed to backfill IsNameField: %v", err)
		errs = append(errs, fmt.Sprintf("backfill IsNameField: %v", err))
	} else if n, _ := res.RowsAffected(); n > 0 {
		log.Printf("   ✅ Backfilled IsNameField for %d field(s)", n)
	}

	// --- Trailing slash cleanup ---
	// Fix labels with trailing "/" characters (e.g. "Some Label/" → "Some Label")
	trailingSlashJoin := ""
	trailingSlashWhere := fmt.Sprintf(
		"f.`%s` LIKE '%%/' AND (f.`%s` = 0 OR f.`%s` IS NULL) AND COALESCE(f.`%s`, 0) = 0",
		constants.FieldSysField_Label,
		constants.FieldSysField_IsCustomized, constants.FieldSysField_IsCustomized,
		constants.FieldIsDeleted,
	)
	var trailingSlashArgs []interface{}
	if scoped {
		predicate, args := managedFieldTuplePredicate(tableDefs, fmt.Sprintf("o.`%s`", constants.FieldAPIName), fmt.Sprintf("f.`%s`", constants.FieldAPIName))
		trailingSlashJoin = "JOIN `" + constants.TableObject + "` o ON o.`" + constants.ColID + "` = f.`" + constants.FieldObjectID + "` "
		trailingSlashWhere += " AND " + predicate
		trailingSlashArgs = args
	}
	trailingSlashSQL := fmt.Sprintf(
		"UPDATE `%s` f %sSET f.`%s` = TRIM(TRAILING '/' FROM f.`%s`), f.`%s` = NOW() "+
			"WHERE %s",
		constants.TableField,
		trailingSlashJoin,
		constants.FieldSysField_Label, constants.FieldSysField_Label,
		constants.FieldLastModifiedDate,
		trailingSlashWhere,
	)
	res, err = rawDB.Exec(trailingSlashSQL, trailingSlashArgs...)
	if err != nil {
		log.Printf("   ⚠️  Failed to fix trailing slash labels: %v", err)
		errs = append(errs, fmt.Sprintf("fix trailing slash labels: %v", err))
	} else if n, _ := res.RowsAffected(); n > 0 {
		log.Printf("   ✅ Fixed %d label(s) with trailing slash", n)
	}

	log.Println("✅ PluralLabel/IsNameField/Label reconciliation complete")
	if len(errs) > 0 {
		return fmt.Errorf("PluralLabel/IsNameField reconciliation had %d error(s): %s", len(errs), strings.Join(errs, "; "))
	}
	return nil
}

// Autonumber v2 seed+verify has been relocated out of bootstrap
// phase5 (Round-5 Codex V1-P1 / V2-C1). The previous
// `runAutoNumberPhase` + `loadAutoNumberSeedsFromDB` pair read
// `_System_Field` with a 5-column SELECT that dropped LegacySource,
// forcing scope-aware GREATEST to fall back to `max(live+1, start)`
// and silently ignore legacy `_System_AutoNumber.CurrentNumber` floors.
// It also ran BEFORE pack apply populated the
// very columns it tried to read, so on a fresh DB the seed was a
// no-op.
//
// Seeding + V1-V6 verification now run inside the app-pack apply path
// right after `applyAutoNumberFormats`, using the generator's in-memory seed slice
// via `packs.BootstrapSeeds()` so LegacySource reaches
// `BuildScopeSourceMapFromSeeds` without a lossy DB round-trip. The
// outer `nexuscrm_controlplane_apply` advisory lock held by
// `controlplane.ApplyAll` continues to serialize multi-pod concurrent
// applies; `SeedAutoNumberCounters` still does not acquire its own
// GET_LOCK.

// reconcileFieldLabelsScoped updates stale field labels for non-customized,
// live fields. Explicit catalog labels are authoritative, including labels
// ending in " ID". The algorithmic fallback only initializes fields whose
// current label still equals their API name.
func reconcileFieldLabelsScoped(rawDB *sql.DB, tableDefs []schema.TableDefinition, scoped bool) error {
	log.Println("🔄 Reconciling field labels...")
	start := time.Now()

	type labelKey struct {
		objectAPIName string
		fieldAPIName  string
	}
	type labelCandidate struct {
		key       labelKey
		fieldName string
	}

	normalizeKey := func(objectAPIName, fieldAPIName string) labelKey {
		return labelKey{
			objectAPIName: strings.ToLower(objectAPIName),
			fieldAPIName:  strings.ToLower(fieldAPIName),
		}
	}

	explicitLabels := make(map[labelKey]string)
	candidateOrder := make(map[labelKey]int)
	candidates := make([]labelCandidate, 0)
	for _, def := range tableDefs {
		if strings.TrimSpace(def.TableName) == "" {
			continue
		}
		for _, col := range def.Columns {
			if strings.TrimSpace(col.Name) == "" {
				continue
			}
			key := normalizeKey(def.TableName, col.Name)
			if _, exists := candidateOrder[key]; !exists {
				candidateOrder[key] = len(candidates)
				candidates = append(candidates, labelCandidate{
					key:       key,
					fieldName: col.Name,
				})
			}
			if col.Label == "" {
				continue
			}
			if existing, exists := explicitLabels[key]; exists && existing != col.Label {
				return fmt.Errorf(
					"conflicting explicit labels for %s.%s: %q and %q",
					def.TableName, col.Name, existing, col.Label,
				)
			}
			explicitLabels[key] = col.Label
		}
	}

	type labelState struct {
		id            string
		objectAPIName string
		fieldAPIName  string
		currentLabel  sql.NullString
		fieldType     sql.NullString
		customized    bool
		deleted       bool
	}

	statesByID := make(map[string]labelState)
	loadStates := func(query string, args ...interface{}) error {
		rows, err := rawDB.Query(query, args...)
		if err != nil {
			return err
		}
		for rows.Next() {
			var state labelState
			if err := rows.Scan(
				&state.id,
				&state.objectAPIName,
				&state.fieldAPIName,
				&state.currentLabel,
				&state.fieldType,
				&state.customized,
				&state.deleted,
			); err != nil {
				_ = rows.Close()
				return err
			}
			if existing, exists := statesByID[state.id]; exists {
				if existing.objectAPIName != state.objectAPIName || existing.fieldAPIName != state.fieldAPIName {
					_ = rows.Close()
					return fmt.Errorf(
						"field label candidate ID %q maps to both %s.%s and %s.%s",
						state.id,
						existing.objectAPIName, existing.fieldAPIName,
						state.objectAPIName, state.fieldAPIName,
					)
				}
				continue
			}
			statesByID[state.id] = state
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return err
		}
		return rows.Close()
	}

	// Keep each tuple-IN prefetch bounded. On TiDB, a single candidate list can
	// contain thousands of pairs; planning and evaluating that one large
	// LOWER(object),field tuple predicate has unbounded tail latency on a cold
	// branch. The reconciliation is read-only until the later update phase, so
	// chunking preserves its all-or-nothing write semantics while bounding each
	// query to 500 bind parameters.
	const candidateBatchSize = 250
	selectPrefix := fmt.Sprintf(
		"SELECT f.`%s`, o.`%s`, f.`%s`, f.`%s`, f.`%s`, "+
			"COALESCE(f.`%s`, 0), COALESCE(f.`%s`, 0) "+
			"FROM `%s` f JOIN `%s` o ON f.`%s` = o.`%s` ",
		constants.ColID,
		constants.FieldAPIName,
		constants.FieldAPIName,
		constants.FieldSysField_Label,
		constants.FieldSysField_Type,
		constants.FieldSysField_IsCustomized,
		constants.FieldIsDeleted,
		constants.TableField,
		constants.TableObject,
		constants.FieldObjectID,
		constants.ColID,
	)

	for i := 0; i < len(candidates); i += candidateBatchSize {
		end := i + candidateBatchSize
		if end > len(candidates) {
			end = len(candidates)
		}
		placeholders := make([]string, 0, end-i)
		args := make([]interface{}, 0, (end-i)*2)
		for _, candidate := range candidates[i:end] {
			placeholders = append(placeholders, "(?, ?)")
			args = append(args, candidate.key.objectAPIName, candidate.fieldName)
		}
		query := selectPrefix + fmt.Sprintf(
			"WHERE (LOWER(o.`%s`), f.`%s`) IN (%s)",
			constants.FieldAPIName,
			constants.FieldAPIName,
			strings.Join(placeholders, ", "),
		)
		if err := loadStates(query, args...); err != nil {
			return fmt.Errorf("prefetch field label candidates batch %d-%d: %w", i, end, err)
		}
	}

	// Full bootstrap historically initializes discovered fields outside the
	// managed catalog when their label still equals their API name. Preserve
	// that behavior without widening managed/L0-scoped reconciliation.
	if !scoped {
		query := selectPrefix + fmt.Sprintf(
			"WHERE f.`%s` = f.`%s`",
			constants.FieldSysField_Label,
			constants.FieldAPIName,
		)
		if err := loadStates(query); err != nil {
			return fmt.Errorf("prefetch algorithmic field label candidates: %w", err)
		}
	}

	states := make([]labelState, 0, len(statesByID))
	for _, state := range statesByID {
		states = append(states, state)
	}
	sort.Slice(states, func(i, j int) bool {
		leftKey := normalizeKey(states[i].objectAPIName, states[i].fieldAPIName)
		rightKey := normalizeKey(states[j].objectAPIName, states[j].fieldAPIName)
		leftOrder, leftManaged := candidateOrder[leftKey]
		rightOrder, rightManaged := candidateOrder[rightKey]
		if leftManaged != rightManaged {
			return leftManaged
		}
		if leftManaged && leftOrder != rightOrder {
			return leftOrder < rightOrder
		}
		if leftKey != rightKey {
			if leftKey.objectAPIName != rightKey.objectAPIName {
				return leftKey.objectAPIName < rightKey.objectAPIName
			}
			return leftKey.fieldAPIName < rightKey.fieldAPIName
		}
		return states[i].id < states[j].id
	})

	type labelFix struct {
		id       string
		current  sql.NullString
		desired  string
		explicit bool
	}
	fixes := make([]labelFix, 0)
	referenceCandidates := 0
	customized := 0
	deleted := 0
	for _, state := range states {
		switch strings.ToLower(state.fieldType.String) {
		case "reference", "masterdetail":
			referenceCandidates++
		}
		if state.customized {
			customized++
			continue
		}
		if state.deleted {
			deleted++
			continue
		}

		key := normalizeKey(state.objectAPIName, state.fieldAPIName)
		desired, explicit := explicitLabels[key]
		if !explicit {
			if !state.currentLabel.Valid || state.currentLabel.String != state.fieldAPIName {
				continue
			}
			desired = fieldNameToLabel(state.fieldAPIName)
			if desired == state.fieldAPIName {
				continue
			}
		}
		if state.currentLabel.Valid && state.currentLabel.String == desired {
			continue
		}

		fixes = append(fixes, labelFix{
			id:       state.id,
			current:  state.currentLabel,
			desired:  desired,
			explicit: explicit,
		})
	}

	explicitDeltas := 0
	for _, fix := range fixes {
		if fix.explicit {
			explicitDeltas++
		}
	}
	log.Printf(
		"   ⏱  Prefetched %d field label candidates: %d reference, %d customized, %d deleted, %d true deltas (%d explicit, %d algorithmic) (%v)",
		len(states), referenceCandidates, customized, deleted, len(fixes), explicitDeltas, len(fixes)-explicitDeltas,
		time.Since(start).Round(time.Millisecond),
	)

	var updated int64
	const updateBatchSize = 200
	updateStart := time.Now()
	var tx *sql.Tx
	if len(fixes) > 0 {
		var err error
		tx, err = rawDB.Begin()
		if err != nil {
			return fmt.Errorf("begin field label reconciliation transaction: %w", err)
		}
	}
	for i := 0; i < len(fixes); i += updateBatchSize {
		end := i + updateBatchSize
		if end > len(fixes) {
			end = len(fixes)
		}
		chunk := fixes[i:end]

		var desiredRows strings.Builder
		args := make([]interface{}, 0, len(chunk)*3)
		for index, fix := range chunk {
			if index == 0 {
				desiredRows.WriteString(
					"SELECT CAST(? AS CHAR) AS `Id`, CAST(? AS CHAR) AS `CurrentLabel`, CAST(? AS CHAR) AS `DesiredLabel`",
				)
			} else {
				desiredRows.WriteString(
					" UNION ALL SELECT CAST(? AS CHAR), CAST(? AS CHAR), CAST(? AS CHAR)",
				)
			}
			args = append(args, fix.id, fix.current, fix.desired)
		}

		query := fmt.Sprintf(
			"UPDATE `%s` f JOIN (%s) desired ON f.`%s` = desired.`Id` "+
				"SET f.`%s` = desired.`DesiredLabel`, f.`%s` = NOW() "+
				"WHERE (f.`%s` = 0 OR f.`%s` IS NULL) "+
				"AND COALESCE(f.`%s`, 0) = 0 "+
				"AND BINARY f.`%s` <=> BINARY desired.`CurrentLabel` "+
				"AND NOT (BINARY f.`%s` <=> BINARY desired.`DesiredLabel`)",
			constants.TableField,
			desiredRows.String(),
			constants.ColID,
			constants.FieldSysField_Label,
			constants.FieldLastModifiedDate,
			constants.FieldSysField_IsCustomized,
			constants.FieldSysField_IsCustomized,
			constants.FieldIsDeleted,
			constants.FieldSysField_Label,
			constants.FieldSysField_Label,
		)
		res, err := tx.Exec(query, args...)
		if err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("batch field label update chunk %d-%d: %w", i, end, err)
		}
		affected, err := res.RowsAffected()
		if err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("read field label update count for chunk %d-%d: %w", i, end, err)
		}
		if affected != int64(len(chunk)) {
			_ = tx.Rollback()
			return fmt.Errorf(
				"field label update chunk %d-%d matched %d of %d prefetched deltas",
				i, end, affected, len(chunk),
			)
		}
		updated += affected
	}
	if tx != nil {
		if err := tx.Commit(); err != nil {
			return fmt.Errorf("commit field label reconciliation transaction: %w", err)
		}
	}

	log.Printf(
		"✅ Field label reconciliation complete: %d true deltas, %d rows updated (%v update, %v total)",
		len(fixes), updated, time.Since(updateStart).Round(time.Millisecond), time.Since(start).Round(time.Millisecond),
	)
	return nil
}

// reconcileStandardFieldDefaultValuesScoped backfills DefaultValue on live
// _System_Field rows from standard_objects_spec.json whenever the spec
// declares a `default` and the live row has NULL/empty DefaultValue.
//
// This is initialization from spec, not drift-healing:
//   - Only fills NULL/empty — never overwrites an existing non-empty value.
//     Divergent existing defaults are genuine drift and remain reported
//     (verify-only) by VerifyStandardFieldSpecDrift.
//   - Guarded by IsCustomized so admin-overridden rows are preserved.
//
// Without this pass, standard-field defaults declared in the spec only land
// on fresh installs (via buildFieldMetadata) — existing tenants carry
// whatever value was written the first time the field was created, which
// blocks Create payloads that omit a required field relying on the spec
// default (e.g. Order.StatusCode).
func reconcileStandardFieldDefaultValuesScoped(rawDB *sql.DB, scoped bool) error {
	log.Println("🔄 Reconciling standard-spec field defaults...")
	start := time.Now()

	// Use json.RawMessage for Default so a future spec entry that writes the
	// value as a JSON number/boolean (`"default": 0`) instead of a string
	// does not hard-fail the whole bootstrap phase. Coerce to a canonical
	// string below.
	type specField struct {
		APIName string          `json:"apiName"`
		Default json.RawMessage `json:"default,omitempty"`
	}
	type specObject struct {
		APIName string      `json:"apiName"`
		Fields  []specField `json:"fields"`
	}
	specBytes := packspec.StandardObjectsSpecJSON()
	if scoped {
		var err error
		specBytes, err = packspec.StandardObjectsSpecByCategoryJSON("system")
		if err != nil {
			return fmt.Errorf("filter platform-core standard object spec: %w", err)
		}
	}
	var specs []specObject
	if err := json.Unmarshal(specBytes, &specs); err != nil {
		return fmt.Errorf("parse standard_objects_spec.json: %w", err)
	}

	type target struct {
		table, field, def string
	}
	var targets []target
	for _, o := range specs {
		for _, f := range o.Fields {
			def, ok := packspec.CoerceSpecDefault(f.Default)
			if !ok {
				continue
			}
			targets = append(targets, target{table: o.APIName, field: f.APIName, def: def})
		}
	}
	if len(targets) == 0 {
		log.Printf("✅ Standard-spec default reconciliation complete: nothing to apply (%v)",
			time.Since(start).Round(time.Millisecond))
		return nil
	}

	// Single cross-table UPDATE using CASE for SET and tuple IN for WHERE.
	// Case-insensitive on object APIName to match verify_spec_drift (live
	// rows may store lowercase names while the spec uses canonical API casing). Field APIName
	// is matched exactly — Salesforce field API names are case-stable.
	var setCase strings.Builder
	setCase.WriteString(fmt.Sprintf("CASE CONCAT(LOWER(o.`%s`), '.', f.`%s`)", constants.FieldAPIName, constants.FieldAPIName))
	setArgs := make([]interface{}, 0, 2*len(targets))
	tuplePlaceholders := make([]string, 0, len(targets))
	tupleArgs := make([]interface{}, 0, 2*len(targets))
	for _, t := range targets {
		key := strings.ToLower(t.table) + "." + t.field
		setCase.WriteString(" WHEN ? THEN ?")
		setArgs = append(setArgs, key, t.def)
		tuplePlaceholders = append(tuplePlaceholders, "(?, ?)")
		tupleArgs = append(tupleArgs, strings.ToLower(t.table), t.field)
	}
	setCase.WriteString(" ELSE f.`DefaultValue` END")

	// Constrain to live standard_object rows whose (object,field) tuple was
	// declared in the spec. Without o.TableType='standard_object' a pathological
	// custom object whose ApiName lowercases to a standard name would inherit
	// the spec default; unlikely in practice but cheap to lock down.
	query := fmt.Sprintf(
		"UPDATE `%s` f JOIN `%s` o ON f.`%s` = o.`%s` "+
			"SET f.`DefaultValue` = %s, f.`%s` = NOW() "+
			"WHERE (LOWER(o.`%s`), f.`%s`) IN (%s) "+
			"AND o.`%s` = ? "+
			"AND o.`%s` = 0 "+
			"AND (f.`DefaultValue` IS NULL OR f.`DefaultValue` = '') "+
			"AND (f.`IsCustomized` = 0 OR f.`IsCustomized` IS NULL) "+
			"AND f.`%s` = 0",
		constants.TableField, constants.TableObject,
		constants.FieldObjectID, constants.ColID,
		setCase.String(), constants.FieldLastModifiedDate,
		constants.FieldAPIName, constants.FieldAPIName,
		strings.Join(tuplePlaceholders, ", "),
		constants.FieldSysObject_TableType,
		constants.ColIsDeleted,
		constants.ColIsDeleted,
	)

	allArgs := make([]interface{}, 0, len(setArgs)+len(tupleArgs)+1)
	allArgs = append(allArgs, setArgs...)
	allArgs = append(allArgs, tupleArgs...)
	allArgs = append(allArgs, string(constants.TableTypeStandardObject))

	res, err := rawDB.Exec(query, allArgs...)
	if err != nil {
		return fmt.Errorf("reconcile standard-spec defaults: %w", err)
	}
	n, _ := res.RowsAffected()
	log.Printf("✅ Standard-spec default reconciliation complete: %d rows updated across %d spec fields (%v)",
		n, len(targets), time.Since(start).Round(time.Millisecond))
	return nil
}

// reconcileBooleanDefaultValuesScoped ensures every Required boolean field has a
// concrete DefaultValue ('false'). Required booleans with NULL/empty
// DefaultValue fail Create payloads that omit the key, because the REST
// validator treats "key not in body" as missing (Salesforce parity defaults
// unchecked booleans to false on submit). Idempotent: the second run matches
// zero rows. Tenant-customized rows (IsCustomized=1) are preserved.
func reconcileBooleanDefaultValuesScoped(rawDB *sql.DB, tableDefs []schema.TableDefinition, scoped bool) error {
	log.Println("🔄 Reconciling boolean defaults...")
	start := time.Now()

	fieldJoin := ""
	fieldWhere := fmt.Sprintf(
		"f.`%s` = 'boolean' "+
			"AND f.`Required` = 1 "+
			"AND (f.`DefaultValue` IS NULL OR f.`DefaultValue` = '') "+
			"AND (f.`IsCustomized` = 0 OR f.`IsCustomized` IS NULL) "+
			"AND f.`%s` = 0",
		constants.ColType,
		constants.ColIsDeleted,
	)
	var args []interface{}
	if scoped {
		predicate, predicateArgs := managedFieldTuplePredicate(tableDefs, fmt.Sprintf("o.`%s`", constants.FieldAPIName), fmt.Sprintf("f.`%s`", constants.FieldAPIName))
		fieldJoin = "JOIN `" + constants.TableObject + "` o ON f.`" + constants.FieldObjectID + "` = o.`" + constants.ColID + "` "
		fieldWhere += " AND " + predicate
		args = predicateArgs
	}
	query := fmt.Sprintf(
		"UPDATE `%s` f %s"+
			"SET f.`DefaultValue` = 'false', f.`%s` = NOW() "+
			"WHERE %s",
		constants.TableField,
		fieldJoin,
		constants.FieldLastModifiedDate,
		fieldWhere,
	)

	res, err := rawDB.Exec(query, args...)
	if err != nil {
		return fmt.Errorf("reconcile boolean defaults: %w", err)
	}
	n, _ := res.RowsAffected()
	log.Printf("✅ Boolean default reconciliation complete: %d rows updated (%v)",
		n, time.Since(start).Round(time.Millisecond))
	return nil
}