package bootstrap

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

	"github.com/nexuscrm/backend/internal/controlplane/packspec"
	"github.com/nexuscrm/backend/internal/features/metadata"

	"github.com/nexuscrm/backend/internal/domain/schema"
	"github.com/nexuscrm/backend/internal/infrastructure/database"
	"github.com/nexuscrm/backend/internal/infrastructure/persistence"
	"github.com/nexuscrm/backend/pkg/logging"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	bootstrapStateTable = "_System_BootstrapState"
	bootstrapStateKey   = "schema_initialize"
)

var bootstrapPhaseOrder = map[string]int{
	"start":                    0,
	"phase0_columns_reconcile": 1,
	"phase0_schema_reconcile":  2,
	"phase1_core_tables":       3,
	"phase1_batch_tables":      4,
	"phase2_object_metadata":   5,
	"phase2_read_object_ids":   6,
	"phase2_field_metadata":    7,
	"phase3_field_reconcile":   8,
	"phase4_name_fields":       9,
	"phase5_auto_numbers":      10,
	"phase6_field_labels":      11,
	"phase7_spec_defaults":     12,
	"phase7b_boolean_fallback": 13,
	"phase8_activity_indexes":  14,
	"completed":                15,
}

type bootstrapStateSnapshot struct {
	RunID  string
	Phase  string
	Status string
}

func ensureBootstrapStateTable(rawDB *sql.DB) error {
	ddl := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
		`+"`StateKey` VARCHAR(128) NOT NULL PRIMARY KEY,"+`
		`+"`RunID` VARCHAR(64) NOT NULL,"+`
		`+"`Phase` VARCHAR(128) NOT NULL,"+`
		`+"`Status` VARCHAR(32) NOT NULL,"+`
		`+"`LastError` TEXT NULL,"+`
		`+"`CreatedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,"+`
		`+"`LastModifiedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"+`
	) %s`, bootstrapStateTable, constants.DefaultTableOptions)
	_, err := rawDB.Exec(ddl)
	return err
}

func loadBootstrapState(rawDB *sql.DB) (*bootstrapStateSnapshot, error) {
	row := rawDB.QueryRow(
		fmt.Sprintf("SELECT `RunID`, `Phase`, `Status` FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		bootstrapStateKey,
	)
	var s bootstrapStateSnapshot
	if err := row.Scan(&s.RunID, &s.Phase, &s.Status); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return &s, nil
}

func persistBootstrapState(rawDB *sql.DB, runID, phase, status string, lastErr error) error {
	payload := ""
	if lastErr != nil {
		serialized, marshalErr := json.Marshal(map[string]string{
			"message":   lastErr.Error(),
			"timestamp": time.Now().UTC().Format(time.RFC3339),
		})
		if marshalErr != nil {
			return marshalErr
		}
		payload = string(serialized)
	}
	_, err := rawDB.Exec(
		fmt.Sprintf(
			"INSERT INTO `%s` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) VALUES (?,?,?,?,?,NOW(),NOW()) "+
				"ON DUPLICATE KEY UPDATE `RunID`=VALUES(`RunID`),`Phase`=VALUES(`Phase`),`Status`=VALUES(`Status`),`LastError`=VALUES(`LastError`),`LastModifiedDate`=NOW()",
			bootstrapStateTable,
		),
		bootstrapStateKey, runID, phase, status, payload,
	)
	return err
}

func shouldRunBootstrapPhase(resumeFrom, phase string) bool {
	if resumeFrom == "" {
		return true
	}
	resumeOrder, okResume := bootstrapPhaseOrder[resumeFrom]
	phaseOrder, okPhase := bootstrapPhaseOrder[phase]
	if !okResume || !okPhase {
		return true
	}
	return phaseOrder >= resumeOrder
}

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

	var lockResult int
	if err := lockConn.QueryRowContext(ctx,
		"SELECT GET_LOCK('nexuscrm_bootstrap_mutation', 600)",
	).Scan(&lockResult); err != nil || lockResult != 1 {
		return fmt.Errorf("acquiring bootstrap mutation lock (result=%d): %w", lockResult, err)
	}
	defer func() { _, _ = lockConn.ExecContext(ctx, "SELECT RELEASE_LOCK('nexuscrm_bootstrap_mutation')") }()

	return fn()
}

// 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 check for narrowing)
//
// 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()

	// BULK QUERY: Load ALL columns with their physical shape in ONE query
	rows, err := rawDB.Query(
		"SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE()",
	)
	if err != nil {
		return fmt.Errorf("query INFORMATION_SCHEMA.COLUMNS: %w", err)
	}
	defer rows.Close()

	type existingColumnShape struct {
		columnType string
		nullable   bool
		defaultVal sql.NullString
	}

	// lower(tableName) → lower(colName) → physical shape.
	existingCols := make(map[string]map[string]existingColumnShape)
	for rows.Next() {
		var tableName, colName, colType, isNullable string
		var defaultVal sql.NullString
		if err := rows.Scan(&tableName, &colName, &colType, &isNullable, &defaultVal); err != nil {
			return fmt.Errorf("scan INFORMATION_SCHEMA row: %w", err)
		}
		key := strings.ToLower(tableName)
		if existingCols[key] == nil {
			existingCols[key] = make(map[string]existingColumnShape)
		}
		existingCols[key][strings.ToLower(colName)] = existingColumnShape{
			columnType: strings.ToLower(colType),
			nullable:   strings.EqualFold(isNullable, "YES"),
			defaultVal: defaultVal,
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate INFORMATION_SCHEMA: %w", err)
	}

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

		for _, col := range def.Columns {
			desiredType := col.Type
			if desiredType == "" {
				desiredType = "TEXT"
			}

			existingShape, exists := tableCols[strings.ToLower(col.Name)]
			if !exists {
				// Column missing — ADD
				var colDDL strings.Builder
				colDDL.WriteString(fmt.Sprintf("`%s` %s", col.Name, desiredType))
				if !col.Nullable {
					colDDL.WriteString(" NOT NULL")
				}
				if col.Default != "" {
					colDDL.WriteString(fmt.Sprintf(" DEFAULT %s", col.Default))
				}

				ddl := fmt.Sprintf("ALTER TABLE `%s` ADD COLUMN %s", def.TableName, colDDL.String())
				log.Printf("   ➕ Adding missing column: %s.%s (%s)", def.TableName, col.Name, desiredType)
				if _, err := rawDB.Exec(ddl); err != nil {
					log.Printf("   ⚠️  Failed to add column %s.%s: %v", def.TableName, col.Name, err)
					errs = append(errs, fmt.Sprintf("add column %s.%s: %v", def.TableName, col.Name, err))
				} else {
					addCount++
				}

				if !col.Nullable && col.Default != "" {
					backfill := fmt.Sprintf("UPDATE `%s` SET `%s` = %s WHERE `%s` IS NULL",
						def.TableName, col.Name, col.Default, col.Name)
					if _, err := rawDB.Exec(backfill); err != nil {
						log.Printf("   ⚠️  Failed to backfill %s.%s: %v", def.TableName, col.Name, err)
						errs = append(errs, fmt.Sprintf("backfill %s.%s: %v", def.TableName, col.Name, err))
					}
				}
				continue
			}

			// Column exists — check if type matches (normalize aliases first)
			existingType := existingShape.columnType
			typeMatches := normalizedTypeMatch(existingType, desiredType)
			shapeMatches := typeMatches &&
				existingShape.nullable == col.Nullable &&
				defaultShapeMatches(existingShape.defaultVal, col.Default, desiredType)
			if shapeMatches {
				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. Equivalent aliases (BOOLEAN↔tinyint(1)): skip (no real change)
			// All other type changes (TEXT→VARCHAR, DECIMAL precision, INT→BIGINT)
			// are skipped with a warning — they require explicit migration.
			if !typeMatches && !isVarcharChange(existingType, desiredType) {
				log.Printf("   ⚠️  Column %s.%s type drift detected: DB=%s, expected=%s (non-VARCHAR 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 {
					log.Printf("   ❌ 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)
					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 == "" {
						log.Printf("   ❌ Cannot make %s.%s NOT NULL: %d existing NULL row(s) and no default configured",
							def.TableName, col.Name, nullCount)
						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 !col.Nullable {
				colDDL.WriteString(" NOT NULL")
			}
			if col.Default != "" {
				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++
			}
		}
	}
	log.Printf("✅ Column reconciliation complete (%d added, %d modified, %v)", addCount, modifyCount, 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
}

// normalizedTypeMatch returns true if two column types are equivalent after normalizing
// MySQL/TiDB type aliases (e.g., BOOLEAN → tinyint(1), INT → int(11)).
func normalizedTypeMatch(existingType, desiredType string) bool {
	a := normalizeColumnType(existingType)
	b := normalizeColumnType(desiredType)
	return strings.EqualFold(a, b)
}

func defaultShapeMatches(existing sql.NullString, desired, desiredType string) bool {
	if desired == "" {
		return !existing.Valid
	}
	if !existing.Valid {
		return false
	}
	existingDefault := normalizeColumnDefault(existing.String)
	desiredDefault := normalizeColumnDefault(desired)
	if normalizeColumnType(desiredType) == "tinyint(1)" {
		existingBool, existingOK := normalizeBooleanColumnDefault(existingDefault)
		desiredBool, desiredOK := normalizeBooleanColumnDefault(desiredDefault)
		if existingOK && desiredOK {
			return existingBool == desiredBool
		}
	}
	if isTemporalColumnType(desiredType) {
		existingTimestamp, existingOK := normalizeTimestampColumnDefault(existingDefault)
		desiredTimestamp, desiredOK := normalizeTimestampColumnDefault(desiredDefault)
		if existingOK && desiredOK {
			return existingTimestamp == desiredTimestamp
		}
	}
	return strings.EqualFold(existingDefault, desiredDefault)
}

func normalizeColumnDefault(v string) string {
	trimmed := strings.TrimSpace(v)
	if len(trimmed) >= 2 {
		first := trimmed[0]
		last := trimmed[len(trimmed)-1]
		if (first == '\'' && last == '\'') || (first == '"' && last == '"') {
			trimmed = strings.TrimSpace(trimmed[1 : len(trimmed)-1])
		}
	}
	return trimmed
}

func normalizeBooleanColumnDefault(v string) (bool, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "1", "true":
		return true, true
	case "0", "false":
		return false, true
	default:
		return false, false
	}
}

func isTemporalColumnType(t string) bool {
	normalized := normalizeColumnType(t)
	return strings.HasPrefix(normalized, "datetime") || strings.HasPrefix(normalized, "timestamp")
}

func normalizeTimestampColumnDefault(v string) (string, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "current_timestamp", "current_timestamp()":
		return "current_timestamp", true
	default:
		return "", false
	}
}

func printableDefault(v sql.NullString) string {
	if !v.Valid {
		return "<null>"
	}
	return v.String
}

func printableDesiredDefault(v string) string {
	if v == "" {
		return "<null>"
	}
	return v
}

// normalizeColumnType maps MySQL/TiDB type aliases to their canonical INFORMATION_SCHEMA form.
func normalizeColumnType(t string) string {
	lower := strings.ToLower(strings.TrimSpace(t))
	switch lower {
	case "boolean", "bool":
		return "tinyint(1)"
	case "int":
		return "int(11)"
	case "bigint":
		return "bigint(20)"
	case "smallint":
		return "smallint(6)"
	case "tinyint":
		return "tinyint(4)"
	case "integer":
		return "int(11)"
	default:
		return lower
	}
}

// isVarcharChange returns true if both types are VARCHAR (regardless of width).
func isVarcharChange(existing, desired string) bool {
	return extractVarcharWidth(existing) > 0 && extractVarcharWidth(desired) > 0
}

// isVarcharNarrowing returns true if the type change is VARCHAR(X) → VARCHAR(Y) where Y < X.
func isVarcharNarrowing(existing, desired string) bool {
	existWidth := extractVarcharWidth(existing)
	desiredWidth := extractVarcharWidth(desired)
	if existWidth <= 0 || desiredWidth <= 0 {
		return false
	}
	return desiredWidth < existWidth
}

// extractVarcharWidth extracts the width from a VARCHAR(N) type string.
// Returns -1 if not a VARCHAR type.
func extractVarcharWidth(colType string) int {
	lower := strings.ToLower(colType)
	if !strings.HasPrefix(lower, "varchar(") {
		return -1
	}
	inner := strings.TrimPrefix(lower, "varchar(")
	inner = strings.TrimSuffix(inner, ")")
	var width int
	if _, err := fmt.Sscanf(inner, "%d", &width); err != nil {
		return -1
	}
	return width
}

// ---------------------------------------------------------------------------
// 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 {
	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')") }()

	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 := reconcileSystemTableIndices(rawDB, tableDefs); err != nil {
		return fmt.Errorf("index reconciliation: %w", err)
	}
	return nil
}

// 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
}

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):
//   - Merges boolean permission columns (MAX) 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
}

// deduplicatePermTable removes duplicates from a permission table, preserving
// the most permissive boolean values and the newest row per unique group.
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: Merge boolean columns — broadcast MAX to all rows in each group.
	// After this, every row in a group has the same (most permissive) values,
	// so the survivor choice cannot silently revoke permissions.
	maxParts := make([]string, len(boolCols))
	setParts := make([]string, len(boolCols))
	for i, col := range boolCols {
		maxParts[i] = fmt.Sprintf("MAX(`%s`) AS `_max_%s`", col, col)
		setParts[i] = fmt.Sprintf("t.`%s` = agg.`_max_%s`", col, col)
	}
	onParts := make([]string, len(uniqueCols))
	for i, col := range uniqueCols {
		onParts[i] = fmt.Sprintf("t.`%s` = agg.`%s`", col, col)
	}
	mergeSQL := fmt.Sprintf(
		"UPDATE `%s` t INNER JOIN (SELECT %s, %s FROM `%s` GROUP BY %s HAVING COUNT(*) > 1) agg ON %s SET %s",
		tableName, colList, strings.Join(maxParts, ", "), tableName, colList,
		strings.Join(onParts, " AND "), strings.Join(setParts, ", "),
	)
	if _, err := rawDB.Exec(mergeSQL); err != nil {
		return fmt.Errorf("merging boolean columns: %w", err)
	}
	log.Printf("   ✅ %s: boolean merge complete", tableName)

	// Step 2: Materialize duplicate IDs into a staging table (single full-scan).
	// Uses explicit CREATE + INSERT (not CTAS) for TiDB compatibility.
	// Survivor = latest LastModifiedDate, ties broken by Id DESC.
	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)
	}

	// 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
}

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.
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 := dropObsoleteFileServiceIndexes(rawDB, existingTables); err != nil {
		return fmt.Errorf("drop obsolete file service indexes: %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()

	type idxInfo struct {
		name   string
		unique bool
	}
	existingIndices := make(map[string]map[string]idxInfo)
	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]idxInfo)
		}
		existingIndices[key][strings.ToLower(cols)] = idxInfo{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]

		for _, idx := range def.Indices {
			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 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 dropObsoleteFileServiceIndexes(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"},
	}

	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")
}

// 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
}

// InitializeSchema creates core system tables using SchemaManager
// This replaces raw SQL execution with declarative table definitions
func InitializeSchema(db *database.TiDBConnection) error {
	log.Println("🔧 Initializing core system schema...")
	rawDB := db.DB()

	// Create SchemaManager for centralized DDL execution
	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 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
		}

		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
				}
				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,
					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 ensurePluralLabelsAndNameFields(rawDB)
		}); 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
		// `packs.PlatformConfigPack.Apply` 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 reconcileFieldLabels(rawDB, tableDefs)
		}); 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 reconcileStandardFieldDefaultValues(rawDB)
		}); err != nil {
			return err
		}
		if err := runPhase("phase7b_boolean_fallback", func() error {
			return reconcileBooleanDefaultValues(rawDB)
		}); err != nil {
			return err
		}
		if err := runPhase("phase8_activity_indexes", func() error {
			return ensureActivityLookupIndexes(rawDB)
		}); 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.Println("✅ Core system schema initialized (optimized)")
	return nil
}

// VerifySchemaReadiness checks managed schema registry health without mutating schema.
func VerifySchemaReadiness(db *database.TiDBConnection) error {
	repo := persistence.NewSchemaRepository(db.DB(), logging.Default)
	health, err := repo.ValidateSchemaRegistry()
	if err != nil {
		return fmt.Errorf("validate schema registry: %w", err)
	}
	if health == nil {
		return fmt.Errorf("schema registry health check returned nil")
	}
	if health.Status != persistence.StatusHealthy {
		return fmt.Errorf("schema registry unhealthy: expected=%d actual=%d missing=%v",
			health.ExpectedCount, health.ActualCount, health.MissingTables)
	}
	if err := verifyCriticalManagedColumns(db.DB()); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	return nil
}

func verifyCriticalManagedColumns(rawDB *sql.DB) error {
	required := map[string][]string{
		constants.TableObject: {
			constants.FieldSysObject_IsCustomized,
			constants.FieldSysObject_AccessPolicy,
		},
	}

	var missing []string
	for tableName, columns := range required {
		for _, columnName := range columns {
			var count int
			err := rawDB.QueryRow(
				"SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = LOWER(?) AND LOWER(COLUMN_NAME) = LOWER(?)",
				tableName,
				columnName,
			).Scan(&count)
			if err != nil {
				return fmt.Errorf("verify required column %s.%s: %w", tableName, columnName, err)
			}
			if count == 0 {
				missing = append(missing, fmt.Sprintf("%s.%s", tableName, columnName))
			}
		}
	}

	if len(missing) > 0 {
		return fmt.Errorf(
			"schema readiness missing required columns: %s; run 'cmd/setup apply' to reconcile schema",
			strings.Join(missing, ", "),
		)
	}

	return nil
}

// pluralize applies basic English pluralization rules to a label:
//   - ends in consonant + "y" → replace "y" with "ies" (Opportunity → Opportunities)
//   - ends in "z" → double "z" + "es" (Quiz → Quizzes)
//   - ends in sibilant ("ss", "us", "is", "x", "sh", "ch") → append "es"
//   - ends in non-sibilant "s" → already plural, return as-is
//   - otherwise → append "s" (Account → Accounts)
func pluralize(label string) string {
	if label == "" {
		return ""
	}
	lower := strings.ToLower(label)
	n := len(lower)
	if n >= 2 {
		lastChar := lower[n-1]
		prevChar := lower[n-2]
		// consonant + y → ies
		if lastChar == 'y' && !strings.ContainsRune("aeiou", rune(prevChar)) {
			return label[:len(label)-1] + "ies"
		}
		// ends in "sh" or "ch" → es
		if (prevChar == 's' || prevChar == 'c') && lastChar == 'h' {
			return label + "es"
		}
		// sibilant "s" endings: ss, us, is → es
		if lastChar == 's' && (prevChar == 's' || prevChar == 'u' || prevChar == 'i') {
			return label + "es"
		}
	}
	// ends in x → es
	last := lower[n-1]
	if last == 'x' {
		return label + "es"
	}
	// ends in z → double the z + es (Quiz → Quizzes), but not if already doubled (Buzz → Buzzes)
	if last == 'z' {
		if n >= 2 && lower[n-2] == 'z' {
			return label + "es"
		}
		return label + "zes"
	}
	// non-sibilant "s" → already plural
	if last == 's' {
		return label
	}
	return label + "s"
}

// 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.
func defaultSharingModel(tableType string, apiName string) string {
	// Folder tables use dedicated FolderShare access control, not standard RLS.
	if apiName == "ReportFolder" {
		return string(constants.SharingModelFolderSharing)
	}
	// Folder-content tables: visibility derived from FolderId's folder access.
	// "The folder is the atomic unit of sharing" — report/dashboard/template access = folder access.
	if apiName == "Report" || apiName == "Dashboard" || apiName == "_System_EmailTemplate" {
		return string(constants.SharingModelFolderContentSharing)
	}
	switch tableType {
	case "system_core", "system_metadata", "system_data", "system_junction":
		return string(constants.SharingModelPublicReadWrite)
	default:
		return string(constants.SharingModelPrivate)
	}
}

// ReconcileManagedStandardObjectAccessPolicies updates authoritative access
// policy metadata for managed standard objects.
//
// 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 ReconcileManagedStandardObjectAccessPolicies(rawDB *sql.DB) error {
	return reconcileManagedStandardObjectAccessPolicies(rawDB, GetManagedTableDefinitions())
}

// ReconcilePackAccessPolicies reconciles managed access policies using the
// provided table definitions instead of the substrate definitions. This is used
// by the PlatformCore pack whose standard object table defs are not in
// the substrate (system_tables.json).
func ReconcilePackAccessPolicies(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)
}

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,
		}
		if legacy, ok := packspec.LegacyStandardObjectSharingModels()[apiName]; ok {
			query = fmt.Sprintf(
				"UPDATE `%s` SET `%s` = ?, `%s` = NOW() WHERE LOWER(`%s`) = ? AND `%s` = ? AND ((`%s` = 0 OR `%s` IS NULL) OR `%s` = ?) AND `%s` != ?",
				constants.TableObject,
				constants.FieldSysObject_SharingModel,
				constants.FieldLastModifiedDate,
				constants.FieldAPIName,
				constants.FieldSysObject_TableType,
				constants.FieldSysObject_IsCustomized,
				constants.FieldSysObject_IsCustomized,
				constants.FieldSysObject_SharingModel,
				constants.FieldSysObject_SharingModel,
			)
			args = []interface{}{
				desired,
				apiName,
				string(constants.TableTypeStandardObject),
				string(legacy),
				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.TableType != string(constants.TableTypeStandardObject) || def.AccessPolicy == nil {
			continue
		}
		accessPolicyJSON, err := json.Marshal(def.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),
			string(constants.TableTypeStandardObject),
			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 standard object AccessPolicy values from managed metadata", updated)
	}
	return nil
}

// ensurePluralLabelsAndNameFields 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 ensurePluralLabelsAndNameFields(rawDB *sql.DB) 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)
	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`"+` = '' OR `+"`%s`"+` IS NULL`,
		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,
		constants.FieldSysObject_PluralLabel,
		constants.FieldSysObject_PluralLabel,
	)
	res, err := rawDB.Exec(pluralSQL)
	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.
	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)
	}

	// --- IsNameField for autonumber objects ---
	// Standard Salesforce objects that use an autonumber field (not "Name") as their
	// primary display identifier. Mark the autonumber field as nameField only when
	// no field on that object is already marked as nameField.
	type autoNameDef struct {
		ObjectApiName string
		FieldApiName  string
	}
	autoNameDefs := []autoNameDef{
		{"Quote", "QuoteNumber"},
		{"Case", "CaseNumber"},
		{"Order", "OrderNumber"},
		{"Contract", "ContractNumber"},
	}
	autoNameSQL := fmt.Sprintf(
		"UPDATE `%s` f "+
			"JOIN `%s` o ON f.`%s` = o.`%s` "+
			"SET f.`%s` = 1, f.`%s` = NOW() "+
			"WHERE o.`%s` = ? AND f.`%s` = ? "+
			"AND o.`%s` NOT IN ("+
			"  SELECT sub.`%s` FROM ("+
			"    SELECT `%s` FROM `%s` WHERE `%s` = 1"+
			"  ) sub"+
			")",
		constants.TableField,
		constants.TableObject, constants.FieldObjectID, constants.FieldID,
		constants.FieldSysField_IsNameField, constants.FieldLastModifiedDate,
		constants.FieldAPIName, constants.FieldAPIName,
		constants.FieldID,
		constants.FieldObjectID,
		constants.FieldObjectID, constants.TableField, constants.FieldSysField_IsNameField,
	)
	for _, def := range autoNameDefs {
		res, err = rawDB.Exec(autoNameSQL, def.ObjectApiName, def.FieldApiName)
		if err != nil {
			log.Printf("   ⚠️  Failed to backfill IsNameField for %s.%s: %v", def.ObjectApiName, def.FieldApiName, err)
			errs = append(errs, fmt.Sprintf("backfill IsNameField for %s.%s: %v", def.ObjectApiName, def.FieldApiName, err))
		} else if n, _ := res.RowsAffected(); n > 0 {
			log.Printf("   ✅ Backfilled IsNameField for %s.%s", def.ObjectApiName, def.FieldApiName)
		}
	}

	// --- Reference field label cleanup ---
	// Fix labels like "Opportunity ID" → "Opportunity" for reference/masterDetail fields.
	// These labels were auto-generated with "ID" suffix during SF import.
	// Only update labels that end with " ID" and the field is a reference type.
	refLabelSQL := fmt.Sprintf(
		"UPDATE `%s` SET `%s` = LEFT(`%s`, CHAR_LENGTH(`%s`) - 3), `%s` = NOW() "+
			"WHERE `%s` LIKE '%% ID' AND (`%s` = 'reference' OR `%s` = 'masterDetail' OR `%s` = 'masterdetail')",
		constants.TableField,
		constants.FieldSysField_Label, constants.FieldSysField_Label, constants.FieldSysField_Label,
		constants.FieldLastModifiedDate,
		constants.FieldSysField_Label,
		constants.FieldSysField_Type, constants.FieldSysField_Type, constants.FieldSysField_Type,
	)
	res, err = rawDB.Exec(refLabelSQL)
	if err != nil {
		log.Printf("   ⚠️  Failed to fix reference field labels: %v", err)
		errs = append(errs, fmt.Sprintf("fix reference field labels: %v", err))
	} else if n, _ := res.RowsAffected(); n > 0 {
		log.Printf("   ✅ Fixed %d reference field label(s) (removed ' ID' suffix)", n)
	}

	// --- Trailing slash cleanup ---
	// Fix labels with trailing "/" characters (e.g. "Some Label/" → "Some Label")
	trailingSlashSQL := fmt.Sprintf(
		"UPDATE `%s` SET `%s` = TRIM(TRAILING '/' FROM `%s`), `%s` = NOW() "+
			"WHERE `%s` LIKE '%%/'",
		constants.TableField,
		constants.FieldSysField_Label, constants.FieldSysField_Label,
		constants.FieldLastModifiedDate,
		constants.FieldSysField_Label,
	)
	res, err = rawDB.Exec(trailingSlashSQL)
	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 `packs.PlatformConfigPack.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
// `packs.PlatformConfigPack.Apply` 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.

// reconcileFieldLabels updates stale field labels for non-customized fields.
// Two sources of desired labels:
//  1. system_tables.json — explicit "label" values for system table columns
//  2. fieldNameToLabel() — algorithmic derivation for all other fields (camelCase split, Id strip)
//
// Only updates fields where IsCustomized = 0 (or NULL), preserving admin overrides.
// On a warm restart where all labels are already correct, this is a no-op (<100ms).
func reconcileFieldLabels(rawDB *sql.DB, tableDefs []schema.TableDefinition) error {
	log.Println("🔄 Reconciling field labels...")
	start := time.Now()

	// Step 1: Build desired labels from system_tables.json (explicit labels override algorithm)
	// Key: (tableName, fieldApiName) → desired label
	explicitLabels := make(map[string]map[string]string)
	for _, def := range tableDefs {
		m := make(map[string]string)
		for _, col := range def.Columns {
			if col.Label != "" {
				m[col.Name] = col.Label
			}
		}
		if len(m) > 0 {
			explicitLabels[def.TableName] = m
		}
	}

	// Step 2: Single cross-table UPDATE for all explicit labels.
	// Merges ~98 per-table UPDATEs into 1 query using (o.ApiName, f.ApiName) tuple matching.
	step2Start := time.Now()
	sysUpdated := int64(0)
	totalFields := 0

	// Build: UPDATE _System_Field f JOIN _System_Object o ON f.ObjectId = o.Id
	//   SET f.Label = CASE CONCAT(o.ApiName, '.', f.ApiName) WHEN ? THEN ? ... END,
	//       f.LastModifiedDate = NOW()
	//   WHERE (o.ApiName, f.ApiName) IN ((...), (...))
	//     AND (f.IsCustomized = 0 OR f.IsCustomized IS NULL)
	//     AND f.Label != CASE CONCAT(o.ApiName, '.', f.ApiName) WHEN ? THEN ? ... END
	var setCaseParts strings.Builder
	var filterCaseParts strings.Builder
	setCaseArgs := make([]interface{}, 0, 2000)
	filterCaseArgs := make([]interface{}, 0, 2000)
	var tupleArgs []interface{}
	var tuplePlaceholders []string

	concatExpr := fmt.Sprintf("CONCAT(o.`%s`, '.', f.`%s`)", constants.FieldAPIName, constants.FieldAPIName)
	setCaseParts.WriteString("CASE " + concatExpr)
	filterCaseParts.WriteString("CASE " + concatExpr)

	for tableName, fieldLabels := range explicitLabels {
		for fieldName, desiredLabel := range fieldLabels {
			key := tableName + "." + fieldName
			setCaseParts.WriteString(" WHEN ? THEN ?")
			setCaseArgs = append(setCaseArgs, key, desiredLabel)
			filterCaseParts.WriteString(" WHEN ? THEN ?")
			filterCaseArgs = append(filterCaseArgs, key, desiredLabel)
			tuplePlaceholders = append(tuplePlaceholders, "(?, ?)")
			tupleArgs = append(tupleArgs, tableName, fieldName)
			totalFields++
		}
	}
	setCaseParts.WriteString(" ELSE f.`Label` END")
	filterCaseParts.WriteString(" ELSE f.`Label` END")

	if totalFields > 0 {
		// args order: [SET case args...] [tuple args for WHERE IN...] [filter case args...]
		allArgs := make([]interface{}, 0, len(setCaseArgs)+len(tupleArgs)+len(filterCaseArgs))
		allArgs = append(allArgs, setCaseArgs...)
		allArgs = append(allArgs, tupleArgs...)
		allArgs = append(allArgs, filterCaseArgs...)

		query := fmt.Sprintf(
			"UPDATE `%s` f JOIN `%s` o ON f.`%s` = o.`%s` "+
				"SET f.`Label` = %s, f.`%s` = NOW() "+
				"WHERE (o.`%s`, f.`%s`) IN (%s) "+
				"AND (f.IsCustomized = 0 OR f.IsCustomized IS NULL) "+
				"AND f.`Label` != %s",
			constants.TableField, constants.TableObject,
			constants.FieldObjectID, constants.ColID,
			setCaseParts.String(), constants.FieldLastModifiedDate,
			constants.FieldAPIName, constants.FieldAPIName,
			strings.Join(tuplePlaceholders, ", "),
			filterCaseParts.String(),
		)

		res, err := rawDB.Exec(query, allArgs...)
		if err != nil {
			return fmt.Errorf("cross-table label update: %w", err)
		} else if n, _ := res.RowsAffected(); n > 0 {
			sysUpdated = n
		}
	}
	log.Printf("   ⏱  Step 2 (explicit labels): %d fields in 1 query, %d rows updated (%v)",
		totalFields, sysUpdated, time.Since(step2Start).Round(time.Millisecond))

	var errs []string

	// Step 3: Update ALL non-system fields where Label == ApiName (never humanized).
	// These are discovered table fields that were INSERT IGNORE'd with Label = ApiName.
	// Use fieldNameToLabel() algorithm to derive the correct label.
	//
	// We do this in Go (not SQL) because the camelCase splitting algorithm is in Go.
	step3Start := time.Now()
	rows, err := rawDB.Query(
		fmt.Sprintf("SELECT f.`%s`, f.`%s` FROM `%s` f WHERE f.`%s` = f.`%s` AND (f.IsCustomized = 0 OR f.IsCustomized IS NULL)",
			constants.ColID, constants.FieldAPIName,
			constants.TableField,
			"Label", constants.FieldAPIName,
		),
	)
	if err != nil {
		return fmt.Errorf("query stale labels: %w", err)
	}
	defer rows.Close()

	type labelFix struct{ id, desired string }
	var fixes []labelFix
	for rows.Next() {
		var id, apiName string
		if err := rows.Scan(&id, &apiName); err != nil {
			errs = append(errs, fmt.Sprintf("scan stale label row: %v", err))
			continue
		}
		desired := fieldNameToLabel(apiName)
		if desired != apiName { // only fix if the algorithm produces something different
			fixes = append(fixes, labelFix{id, desired})
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating stale labels: %v", err)
		errs = append(errs, fmt.Sprintf("iterate stale labels: %v", err))
	}
	log.Printf("   ⏱  Step 3 SELECT: %d stale labels found (%v)",
		len(fixes), time.Since(step3Start).Round(time.Millisecond))

	// Batch update using CASE WHEN in chunks of 200 to keep query size reasonable.
	// Reduces ~343 individual UPDATEs to ~2 batch statements.
	step3UpdateStart := time.Now()
	algoUpdated := int64(0)
	const batchSize = 200
	for i := 0; i < len(fixes); i += batchSize {
		end := i + batchSize
		if end > len(fixes) {
			end = len(fixes)
		}
		chunk := fixes[i:end]

		// Build: UPDATE _System_Field SET Label = CASE Id WHEN ? THEN ? ... END,
		//   LastModifiedDate = NOW() WHERE Id IN (?, ?, ...)
		var caseParts strings.Builder
		args := make([]interface{}, 0, len(chunk)*3)
		inPlaceholders := make([]string, 0, len(chunk))

		caseParts.WriteString(fmt.Sprintf("CASE `%s`", constants.ColID))
		for _, fix := range chunk {
			caseParts.WriteString(" WHEN ? THEN ?")
			args = append(args, fix.id, fix.desired)
			inPlaceholders = append(inPlaceholders, "?")
		}
		caseParts.WriteString(" END")

		for _, fix := range chunk {
			args = append(args, fix.id)
		}

		query := fmt.Sprintf("UPDATE `%s` SET `Label` = %s, `%s` = NOW() WHERE `%s` IN (%s)",
			constants.TableField, caseParts.String(), constants.FieldLastModifiedDate,
			constants.ColID, strings.Join(inPlaceholders, ","),
		)

		res, err := rawDB.Exec(query, args...)
		if err != nil {
			log.Printf("   ⚠️  Batch algorithmic label update failed (chunk %d-%d): %v", i, end, err)
			errs = append(errs, fmt.Sprintf("batch label update chunk %d-%d: %v", i, end, err))
			continue
		}
		if n, _ := res.RowsAffected(); n > 0 {
			algoUpdated += n
		}
	}

	log.Printf("   ⏱  Step 3 UPDATE: %d algorithmic labels updated (%v)",
		algoUpdated, time.Since(step3UpdateStart).Round(time.Millisecond))
	log.Printf("✅ Field label reconciliation complete: %d system + %d algorithmic labels updated (%v)",
		sysUpdated, algoUpdated, time.Since(start).Round(time.Millisecond))
	if len(errs) > 0 {
		return fmt.Errorf("field label reconciliation had %d error(s): %s", len(errs), strings.Join(errs, "; "))
	}
	return nil
}

// reconcileStandardFieldDefaultValues 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 reconcileStandardFieldDefaultValues(rawDB *sql.DB) 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"`
	}
	var specs []specObject
	if err := json.Unmarshal(packspec.StandardObjectsSpecJSON(), &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 "order" while the spec uses "Order"). 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
}

// reconcileBooleanDefaultValues 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 reconcileBooleanDefaultValues(rawDB *sql.DB) error {
	log.Println("🔄 Reconciling boolean defaults...")
	start := time.Now()

	query := fmt.Sprintf(
		"UPDATE `%s` "+
			"SET `DefaultValue` = 'false', `%s` = NOW() "+
			"WHERE `%s` = 'boolean' "+
			"AND `Required` = 1 "+
			"AND (`DefaultValue` IS NULL OR `DefaultValue` = '') "+
			"AND (`IsCustomized` = 0 OR `IsCustomized` IS NULL) "+
			"AND `%s` = 0",
		constants.TableField,
		constants.FieldLastModifiedDate,
		constants.ColType,
		constants.ColIsDeleted,
	)

	res, err := rawDB.Exec(query)
	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
}

// ensureActivityLookupIndexes creates indexes on WhatId and WhoId columns
// for the task and event tables. These tables are discovered from Salesforce
// data imports (not defined in system_tables.json), so their indexes must
// be created separately from the normal DDL path.
func ensureActivityLookupIndexes(rawDB *sql.DB) error {
	indexes := []struct {
		table  string
		name   string
		column string
	}{
		{"task", "idx_task_whatid", "WhatId"},
		{"task", "idx_task_whoid", "WhoId"},
		{"event", "idx_event_whatid", "WhatId"},
		{"event", "idx_event_whoid", "WhoId"},
	}

	var errs []string
	for _, idx := range indexes {
		// Check if table exists before attempting index creation
		var tableName string
		err := rawDB.QueryRow("SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?", idx.table).Scan(&tableName)
		if err != nil {
			continue // Table doesn't exist, skip
		}

		ddl := fmt.Sprintf("CREATE INDEX `%s` ON `%s` (`%s`)", idx.name, idx.table, idx.column)
		if _, err := rawDB.Exec(ddl); err != nil {
			if !isDuplicateIndexNameError(err) {
				log.Printf("⚠️  Failed to create index %s on %s: %v", idx.name, idx.table, err)
				errs = append(errs, fmt.Sprintf("create index %s on %s: %v", idx.name, idx.table, err))
			}
			// Duplicate key name means index already exists — expected on subsequent boots
		} else {
			log.Printf("📇 Created index %s on %s(%s)", idx.name, idx.table, idx.column)
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("activity lookup indexes had %d error(s): %s", len(errs), strings.Join(errs, "; "))
	}
	return nil
}