package bootstrapimpl

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/fieldcapabilities"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	defaultEnsureIndexesWorkers = 8
	envEnsureIndexesWorkers     = "ENSURE_INDEXES_WORKERS"
)

func ensureIndexesWorkerCount() int {
	raw := strings.TrimSpace(os.Getenv(envEnsureIndexesWorkers))
	if raw == "" {
		return defaultEnsureIndexesWorkers
	}
	workers, err := strconv.Atoi(raw)
	if err != nil || workers <= 0 {
		log.Printf("   ⚠️  Invalid %s=%q; using default %d workers", envEnsureIndexesWorkers, raw, defaultEnsureIndexesWorkers)
		return defaultEnsureIndexesWorkers
	}
	return workers
}

type standardIndexJob struct {
	table, indexName string
	cols             []string
	unique           bool
}

// executeStandardIndexJobs creates the collected index set with bounded
// concurrency. Unique-field jobs are the metadata contract and fail closed;
// existing best-effort search-index jobs retain their compatibility behavior.
func executeStandardIndexJobs(rawDB *sql.DB, jobs []standardIndexJob, workers int) (int, error) {
	sem := make(chan struct{}, workers)
	var mu sync.Mutex
	created := 0
	var failures []error
	var wg sync.WaitGroup
	for _, job := range jobs {
		wg.Add(1)
		go func(j standardIndexJob) {
			defer wg.Done()
			sem <- struct{}{}        // acquire
			defer func() { <-sem }() // release

			if j.unique {
				wasCreated, err := ensureColumnUniqueIndex(rawDB, j.table, j.cols[0])
				if err != nil {
					mu.Lock()
					failures = append(failures, fmt.Errorf("ensure unique index %s on %s: %w", j.indexName, j.table, err))
					mu.Unlock()
					return
				}
				if wasCreated {
					mu.Lock()
					created++
					mu.Unlock()
				}
				return
			}

			quotedCols := make([]string, 0, len(j.cols))
			for _, col := range j.cols {
				quotedCols = append(quotedCols, fmt.Sprintf("`%s`", col))
			}
			ddl := fmt.Sprintf("CREATE INDEX `%s` ON `%s` (%s)", j.indexName, j.table, strings.Join(quotedCols, ", "))
			if _, err := rawDB.Exec(ddl); err != nil {
				// A concurrent control-plane run can finish the same index first.
				// Treat only that duplicate condition as idempotent.
				low := strings.ToLower(err.Error())
				if strings.Contains(low, "duplicate") || strings.Contains(low, "already exist") {
					return
				}
				log.Printf("   ⚠️  Could not create standard index %s on %s: %v", j.indexName, j.table, err)
				return
			}

			mu.Lock()
			created++
			mu.Unlock()
		}(job)
	}
	wg.Wait()

	if len(failures) > 0 {
		return created, fmt.Errorf("index convergence failed: %w", errors.Join(failures...))
	}
	return created, nil
}

// DiscoverAndRegisterPhysicalTables scans the DB for physical tables that exist
// but are not yet registered in _System_Object. This handles the case where
// application tables were imported via data migration and need to be
// registered as metadata objects.
func DiscoverAndRegisterPhysicalTables(dbInstance *database.TiDBConnection) error {
	rawDB := dbInstance.DB()

	// Platform field-metadata overrides must run on every startup, not only when new
	// tables are discovered. The early returns for "all tables already
	// registered" and "no missing tables" previously short-circuited past the
	// reconcile, leaving steady-state environments stuck with stale values.
	// Defer makes it the last step of every success path; the overrides are
	// idempotent so running on no-op paths is safe.
	//
	// `reconcileOK` defaults to false so any error path — including ones
	// added later — is fail-safe: the UPDATEs only fire after the caller has
	// committed to a clean `return nil`. This covers not just DB-read
	// failures but partial-write failures in registerMissingTables etc.,
	// where running UPDATEs against half-populated _System_Object rows
	// would join on missing parents and produce spurious log noise.
	reconcileOK := false
	defer func() {
		if reconcileOK {
			reconcilePlatformDiscoveredFieldOverrides(rawDB)
		}
	}()

	// Step 1: Get all physical tables in the DB
	physicalTables, err := getPhysicalTables(rawDB)
	if err != nil {
		return fmt.Errorf("failed to get physical tables: %w", err)
	}

	// Step 2: Get all already-registered objects in _System_Object
	registeredObjects, err := getRegisteredObjects(rawDB)
	if err != nil {
		return fmt.Errorf("failed to get registered objects: %w", err)
	}

	// Step 3: Find missing tables (physical tables not in _System_Object).
	// An active unmanaged _System_Table row is the catalog-owned declaration
	// that a physical table must stay outside object metadata. Deleted rows do
	// not carry that authority, so their physical tables remain discoverable.
	missing, err := getMissingPhysicalTables(rawDB, physicalTables, registeredObjects)
	if err != nil {
		return fmt.Errorf("failed to get physical-table discovery exclusions: %w", err)
	}

	// Also check for objects that are registered but have 0 fields.
	// Use a single aggregate query instead of per-table COUNT(*) (~355 → 1 SQL call).
	var needFields []string
	zeroFieldRows, err := rawDB.Query(
		fmt.Sprintf(
			"SELECT o.`%s` FROM `%s` o LEFT JOIN `%s` f ON o.Id = f.ObjectId "+
				"WHERE o.`%s` NOT LIKE '_System_%%' AND o.`%s` NOT LIKE '_fivetran_%%' AND o.`%s` NOT LIKE 'v_%%' "+
				"GROUP BY o.`%s`, o.Id HAVING COUNT(f.Id) = 0",
			constants.FieldAPIName, constants.TableObject, constants.TableField,
			constants.FieldAPIName, constants.FieldAPIName, constants.FieldAPIName,
			constants.FieldAPIName,
		),
	)
	if err != nil {
		log.Printf("   ⚠️  Failed to find zero-field objects: %v", err)
	} else {
		defer zeroFieldRows.Close()
		for zeroFieldRows.Next() {
			var name string
			if zeroFieldRows.Scan(&name) == nil {
				needFields = append(needFields, name)
			}
		}
		if err := zeroFieldRows.Err(); err != nil {
			log.Printf("   ⚠️  Error iterating zero-field objects: %v", err)
		}
	}

	if len(missing) == 0 && len(needFields) == 0 {
		log.Println("✅ All physical tables are already registered in _System_Object")
		reconcileOK = true
		return nil
	}

	if len(needFields) > 0 {
		log.Printf("🔍 Found %d registered objects with 0 fields, registering fields from physical schema...", len(needFields))
		if err := registerFieldsForDiscoveredTables(rawDB, needFields); err != nil {
			log.Printf("   ⚠️  Failed to register fields: %v", err)
		}
	}

	if len(missing) == 0 {
		reconcileOK = true
		return nil
	}

	log.Printf("🔍 Discovered %d unregistered physical tables, registering...", len(missing))

	// Step 4: Register missing tables
	// Use a single batch INSERT IGNORE to avoid conflicts with Label UNIQUE constraint
	err = registerMissingTables(rawDB, missing, registeredObjects)
	if err != nil {
		return fmt.Errorf("failed to register missing tables: %w", err)
	}

	log.Printf("✅ Registered %d physical tables in _System_Object", len(missing))

	// Step 5: Reconcile orphaned _System_Field records
	// Business objects imported before bootstrap may have _System_Field records
	// pointing to old ObjectIds that no longer exist in _System_Object.
	if err := reconcileOrphanedFields(rawDB); err != nil {
		log.Printf("⚠️  Failed to reconcile orphaned fields: %v", err)
	}

	// Step 6: Correct platform field metadata overrides — handled by the deferred
	// reconcilePlatformDiscoveredFieldOverrides call at the top of this function so
	// steady-state environments also converge on the desired state.

	reconcileOK = true
	return nil
}

func getMissingPhysicalTables(db *sql.DB, physicalTables, registeredObjects []string) ([]string, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, COALESCE(`%s`, 0) FROM `%s` WHERE COALESCE(`%s`, 0) = 0",
		constants.FieldSysTable_TableName,
		constants.FieldSysTable_IsDeleted,
		constants.TableTable,
		constants.FieldSysTable_IsManaged,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	unmanagedLower := make(map[string]bool)
	for rows.Next() {
		var name string
		var deleted int
		if err := rows.Scan(&name, &deleted); err != nil {
			return nil, err
		}
		if deleted == 0 {
			unmanagedLower[strings.ToLower(name)] = true
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	registeredLower := make(map[string]bool)
	for _, name := range registeredObjects {
		registeredLower[strings.ToLower(name)] = true
	}

	var missing []string
	for _, table := range physicalTables {
		lower := strings.ToLower(table)
		// Skip system tables, internal state tables, data-pipeline checkpoints, and already registered.
		//   _system_       — core CRM metadata tables
		//   _fivetran_     — Fivetran sync state
		//   _dev_          — dev session and audit tables (internal runtime state)
		//   _bq_           — BigQuery sync checkpoints (internal pipeline state)
		//   v_             — views, not physical tables
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "_dev_") ||
			strings.HasPrefix(lower, "_bq_") ||
			strings.HasPrefix(lower, "v_") ||
			unmanagedLower[lower] ||
			registeredLower[lower] {
			continue
		}
		missing = append(missing, table)
	}
	return missing, nil
}

func getPhysicalTables(db *sql.DB) ([]string, error) {
	// Use SHOW TABLES instead of INFORMATION_SCHEMA to avoid stale schema cache
	// after bootstrap DDL operations. SHOW TABLES always reflects the current state.
	rows, err := db.Query("SHOW TABLES")
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		tables = append(tables, name)
	}
	return tables, rows.Err()
}

func getRegisteredObjects(db *sql.DB) ([]string, error) {
	rows, err := db.Query(
		fmt.Sprintf("SELECT `%s` FROM `%s`", constants.FieldAPIName, constants.TableObject),
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var names []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		names = append(names, name)
	}
	return names, rows.Err()
}

func registerMissingTables(db *sql.DB, missing []string, existingObjects []string) error {
	// Build a set of existing Labels (case-insensitive) to avoid UNIQUE constraint violations
	existingLabels := make(map[string]bool)
	rows, err := db.Query(
		fmt.Sprintf("SELECT `%s` FROM `%s`", "Label", constants.TableObject),
	)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var label string
		if err := rows.Scan(&label); err != nil {
			return err
		}
		existingLabels[strings.ToLower(label)] = true
	}
	if err := rows.Err(); err != nil {
		return err
	}

	// Build INSERT statements
	var valueParts []string
	var params []interface{}

	for _, table := range missing {
		label := tableNameToLabel(table)
		// If label would conflict, append " (Data)" suffix
		if existingLabels[strings.ToLower(label)] {
			label = label + " (Data)"
		}
		// Double-check after suffix
		if existingLabels[strings.ToLower(label)] {
			label = label + " " + table
		}

		plural := Pluralize(label)
		isCustom := strings.HasSuffix(strings.ToLower(table), "__c")
		customInt := 0
		if isCustom {
			customInt = 1
		}

		// Use the canonical (lowercase-normalized) ID scheme so all write paths —
		// discovery, pack apply, admin UI — converge on the same Id for the
		// same logical object. Previously discovery used `disc_<table>` which
		// collided with pack apply's `obj_<ApiName>_<hash>` on the
		// case-insensitive UNIQUE(ApiName) and produced orphan field rows.
		id := schema.GenerateObjectID(table)
		valueParts = append(valueParts, "(?, ?, 'custom_object', ?, ?, ?, ?, 1, 'Private', 0, NOW(), NOW())")
		params = append(params, id, table, label, plural, table, customInt)

		existingLabels[strings.ToLower(label)] = true // Track for next iteration
	}

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

	// Table name is interpolated via Sprintf because SQL parameters (?) cannot substitute identifiers.
	// constants.TableObject is a compile-time constant — no injection risk.
	sql := fmt.Sprintf(
		"INSERT IGNORE INTO `%s` (`Id`, `ApiName`, `TableType`, `Label`, `PluralLabel`, `Description`, `IsCustom`, `Searchable`, `SharingModel`, `IsDeleted`, `CreatedDate`, `LastModifiedDate`) VALUES %s",
		constants.TableObject,
		strings.Join(valueParts, ", "),
	)

	result, err := db.Exec(sql, params...)
	if err != nil {
		return fmt.Errorf("batch insert failed: %w", err)
	}

	affected, _ := result.RowsAffected()
	log.Printf("   📦 Inserted %d/%d tables (some may have been skipped due to conflicts)", affected, len(missing))

	// Step 5: Also register fields for the newly discovered tables
	// by scanning INFORMATION_SCHEMA.COLUMNS
	if err := registerFieldsForDiscoveredTables(db, missing); err != nil {
		log.Printf("   ⚠️  Failed to register fields for discovered tables: %v", err)
		// Non-fatal: objects will show with 0 fields, but at least they'll be visible
	}

	return nil
}

func registerFieldsForDiscoveredTables(db *sql.DB, tables []string) error {
	// Get ObjectId for each table
	for _, table := range tables {
		var objectId string
		err := db.QueryRow(
			fmt.Sprintf("SELECT `Id` FROM `%s` WHERE `%s` = ?", constants.TableObject, constants.FieldAPIName),
			table,
		).Scan(&objectId)
		if err != nil {
			log.Printf("   ⚠️  Could not find ObjectId for table %s: %v", table, err)
			continue
		}

		// Get columns from INFORMATION_SCHEMA
		colRows, err := db.Query(
			"SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? ORDER BY ORDINAL_POSITION",
			table,
		)
		if err != nil {
			log.Printf("   ⚠️  Could not get columns for table %s: %v", table, err)
			continue
		}

		var fieldValues []string
		var fieldParams []interface{}

		for colRows.Next() {
			var colName, dataType, isNullable string
			var colDefault, colKey sql.NullString
			if err := colRows.Scan(&colName, &dataType, &isNullable, &colDefault, &colKey); err != nil {
				continue
			}

			fieldId := fmt.Sprintf("fld_%s_%s", strings.ReplaceAll(table, "__", "_"), strings.ToLower(colName))
			fieldType := sqlTypeToFieldType(dataType)
			required := isNullable == "NO" && colName != "Id"
			requiredInt := 0
			if required {
				requiredInt = 1
			}

			fieldLabel := fieldNameToLabel(colName)
			fieldValues = append(fieldValues, "(?, ?, ?, ?, ?, ?, 0)")
			fieldParams = append(fieldParams, fieldId, objectId, colName, fieldLabel, fieldType, requiredInt)
		}
		colRows.Close()

		if len(fieldValues) == 0 {
			continue
		}

		insertSQL := fmt.Sprintf(
			"INSERT IGNORE INTO `%s` (`Id`, `ObjectId`, `ApiName`, `Label`, `Type`, `Required`, `IsDeleted`) VALUES %s",
			constants.TableField,
			strings.Join(fieldValues, ", "),
		)

		result, err := db.Exec(insertSQL, fieldParams...)
		if err != nil {
			log.Printf("   ⚠️  Failed to insert fields for %s: %v", table, err)
			continue
		}
		affected, _ := result.RowsAffected()
		if affected > 0 {
			log.Printf("   📋 Registered %d fields for %s", affected, table)
		}
	}

	return nil
}

// fieldNameToLabel converts a field API name to a human-readable label
// e.g. "ApprovedPrice__c" -> "Approved Price", "ParentId" -> "Parent"
func fieldNameToLabel(fieldName string) string {
	name := fieldName
	if strings.HasSuffix(name, "__c") {
		name = strings.TrimSuffix(name, "__c")
	}

	// Strip trailing "Id" for reference-style fields (e.g. "ParentId" -> "Parent").
	// Only for field names, not table names. Safe: "Valid"/"Rapid" end in lowercase "id".
	if len(name) > 4 && strings.HasSuffix(name, "Id") {
		name = strings.TrimSuffix(name, "Id")
	}

	name = splitCamelCase(name)
	name = strings.ReplaceAll(name, "_", " ")
	parts := strings.Fields(name)
	for i, part := range parts {
		if len(part) > 0 {
			parts[i] = strings.ToUpper(part[:1]) + part[1:]
		}
	}
	return strings.Join(parts, " ")
}

// tableNameToLabel converts a table name to a human-readable label
// e.g. "example_object" -> "Example Object", "custom_item__c" -> "Custom Item"
func tableNameToLabel(tableName string) string {
	// Remove __c suffix for custom objects
	name := tableName
	if strings.HasSuffix(name, "__c") {
		name = strings.TrimSuffix(name, "__c")
	}

	// NO "Id" stripping for table names — only fieldNameToLabel does that

	// Apply camelCase splitting (e.g. "FirstName" -> "First Name")
	name = splitCamelCase(name)

	// Split by underscore and capitalize each word
	parts := strings.Split(name, "_")
	for i, part := range parts {
		if len(part) > 0 {
			parts[i] = strings.ToUpper(part[:1]) + part[1:]
		}
	}

	return strings.Join(parts, " ")
}

// reconcileOrphanedFields finds _System_Field records whose ObjectId doesn't match
// any current _System_Object.Id, and re-links them based on the ApiName stored
// in the old _System_Object entry (which may still exist with a different Id due to
// bootstrap UPSERT).
func reconcileOrphanedFields(db *sql.DB) error {
	// Find orphan ObjectIds in _System_Field that don't match any _System_Object.Id
	// For each orphan, try to find the correct _System_Object by matching the table name
	query := fmt.Sprintf(`
		SELECT DISTINCT f.ObjectId
		FROM %s f
		LEFT JOIN %s o ON f.ObjectId = o.Id
		WHERE o.Id IS NULL
	`, "`"+constants.TableField+"`", "`"+constants.TableObject+"`")

	rows, err := db.Query(query)
	if err != nil {
		return fmt.Errorf("failed to find orphan ObjectIds: %w", err)
	}
	defer rows.Close()

	var orphanIds []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			continue
		}
		orphanIds = append(orphanIds, id)
	}
	if err := rows.Err(); err != nil {
		return err
	}

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

	log.Printf("🔗 Found %d orphaned ObjectIds in _System_Field, attempting to reconcile...", len(orphanIds))

	// For each orphan ObjectId, check if there's an _System_Object with matching ApiName
	// We need to find the ApiName associated with the orphan ObjectId by looking at
	// what table the fields belong to (check physical columns)
	reconciled := 0
	for _, orphanId := range orphanIds {
		// Get a sample field from this orphan group to identify the table
		var sampleApiName string
		err := db.QueryRow(
			fmt.Sprintf("SELECT `%s` FROM `%s` WHERE ObjectId = ? LIMIT 1", constants.FieldAPIName, constants.TableField),
			orphanId,
		).Scan(&sampleApiName)
		if err != nil {
			continue
		}

		// Try to find the table this field belongs to by checking INFORMATION_SCHEMA
		tableRows, err := db.Query(
			"SELECT DISTINCT TABLE_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = ? AND TABLE_NAME NOT LIKE '_System_%'",
			sampleApiName,
		)
		if err != nil {
			continue
		}

		var candidateTables []string
		for tableRows.Next() {
			var tName string
			if err := tableRows.Scan(&tName); err == nil {
				candidateTables = append(candidateTables, tName)
			}
		}
		tableRows.Close()

		// For each candidate, check if there's a matching _System_Object
		for _, candidate := range candidateTables {
			var newObjectId string
			err := db.QueryRow(
				fmt.Sprintf("SELECT `Id` FROM `%s` WHERE LOWER(`%s`) = LOWER(?)", constants.TableObject, constants.FieldAPIName),
				candidate,
			).Scan(&newObjectId)
			if err != nil {
				continue
			}

			// Found a match! Update all fields from orphanId to newObjectId
			result, err := db.Exec(
				fmt.Sprintf("UPDATE `%s` SET ObjectId = ? WHERE ObjectId = ?", constants.TableField),
				newObjectId, orphanId,
			)
			if err != nil {
				log.Printf("   ⚠️  Failed to update ObjectId %s → %s: %v", orphanId, newObjectId, err)
				continue
			}
			affected, _ := result.RowsAffected()
			if affected > 0 {
				log.Printf("   🔗 Reconciled %d fields: %s → %s (%s)", affected, orphanId, newObjectId, candidate)
				reconciled++
			}
			break // Found the match, stop looking
		}
	}

	if reconciled > 0 {
		log.Printf("✅ Reconciled %d orphaned field groups", reconciled)
	} else if len(orphanIds) > 0 {
		log.Printf("⚠️  Could not reconcile %d orphaned ObjectIds (fields may not match any physical table)", len(orphanIds))
	}

	return nil
}

// sqlTypeToFieldType maps SQL data types to CRM field types
func sqlTypeToFieldType(sqlType string) string {
	switch strings.ToUpper(sqlType) {
	case "VARCHAR", "CHAR", "TEXT", "MEDIUMTEXT", "LONGTEXT", "TINYTEXT":
		return "Text"
	case "INT", "BIGINT", "SMALLINT", "TINYINT", "MEDIUMINT":
		return "Number"
	case "DECIMAL", "DOUBLE", "FLOAT", "NUMERIC":
		return "Currency"
	case "DATETIME", "TIMESTAMP":
		return "DateTime"
	case "DATE":
		return "Date"
	case "TINYINT(1)", "BOOLEAN", "BOOL":
		return "Checkbox"
	case "JSON":
		return "TextArea"
	default:
		return "Text"
	}
}

// safeIndexName builds a length-safe single-column index name. Index names
// share TiDB's 64-char identifier limit, so a long table+column pair is hashed
// exactly like AddColumn hashes long column/FK names — otherwise CREATE INDEX
// trips error 1059 ("Identifier name … is too long") and, because the index
// runner only swallows duplicates, the index is silently skipped.
func safeIndexName(table, col string) string {
	raw := fmt.Sprintf("idx_%s_%s", strings.ToLower(table), strings.ToLower(col))
	return schema.GenerateSafeIdentifier(raw, "idx")
}

func safeCompositeIndexName(table string, cols ...string) string {
	parts := []string{"idx", strings.ToLower(table)}
	for _, col := range cols {
		parts = append(parts, strings.ToLower(col))
	}
	return schema.GenerateSafeIdentifier(strings.Join(parts, "_"), "idx")
}

// EnsureStandardIndexes creates standard indexes (IsDeleted, CreatedDate, LastModifiedDate,
// OwnerId) on all non-system physical tables that lack them. Discovered tables from data
// migration have no indexes by default — this fixes full-table-scan performance issues.
// Uses 2 bulk queries (SHOW TABLES + INFORMATION_SCHEMA.STATISTICS) instead of per-table checks.
func EnsureStandardIndexes(dbInstance *database.TiDBConnection) error {
	rawDB := dbInstance.DB()
	start := time.Now()
	log.Println("🔧 Ensuring standard indexes on business tables...")

	// Standard columns that should be indexed on every business table
	standardCols := []string{"IsDeleted", "CreatedDate", "LastModifiedDate", "OwnerId"}

	// Step 1: Get all physical tables
	physicalTables, err := getPhysicalTables(rawDB)
	if err != nil {
		return err
	}

	// Step 2: Bulk-load columns that LEAD an existing index.
	// SEQ_IN_INDEX = 1 restricts to the leading column of each index, matching
	// the runtime EnsureColumnIndex/checkColumnIndexed semantics: a column that
	// is only a trailing member of a composite index cannot serve a
	// `WHERE col = ?` lookup, so it does not count as "already indexed" for the
	// standard-column or relationship-FK passes below — a dedicated single-column
	// index is still created.
	existingIndexes := make(map[string]map[string]bool) // table → set of columns that lead an index
	idxRows, err := rawDB.Query(
		"SELECT TABLE_NAME, COLUMN_NAME FROM INFORMATION_SCHEMA.STATISTICS WHERE TABLE_SCHEMA = DATABASE() AND SEQ_IN_INDEX = 1",
	)
	if err != nil {
		log.Printf("   ⚠️  Cannot query index info: %v", err)
		return err
	}
	defer idxRows.Close()
	for idxRows.Next() {
		var tbl, col string
		if idxRows.Scan(&tbl, &col) == nil {
			if existingIndexes[tbl] == nil {
				existingIndexes[tbl] = make(map[string]bool)
			}
			existingIndexes[tbl][col] = true
		}
	}
	if err := idxRows.Err(); err != nil {
		return err
	}

	// Step 3: Bulk-load existing columns
	existingCols := make(map[string]map[string]bool)
	colRows, err := rawDB.Query(
		"SELECT TABLE_NAME, COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE()",
	)
	if err != nil {
		return err
	}
	defer colRows.Close()
	for colRows.Next() {
		var tbl, col string
		if colRows.Scan(&tbl, &col) == nil {
			if existingCols[tbl] == nil {
				existingCols[tbl] = make(map[string]bool)
			}
			existingCols[tbl][col] = true
		}
	}
	if err := colRows.Err(); err != nil {
		return err
	}

	// Step 4: Collect all missing indexes.
	var jobs []standardIndexJob
	for _, table := range physicalTables {
		lower := strings.ToLower(table)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}

		cols := existingCols[table]
		idxs := existingIndexes[table]
		if cols == nil {
			continue
		}

		for _, col := range standardCols {
			if !cols[col] {
				continue
			}
			if idxs != nil && idxs[col] {
				continue
			}
			jobs = append(jobs, standardIndexJob{
				table:     table,
				indexName: safeIndexName(table, col),
				cols:      []string{col},
			})
		}
	}

	// Step 4b: Also index relationship (lookup / master-detail) FK columns.
	// Related-list counts, lookups, joins and RLS subqueries all filter on
	// these; without an index they full-scan the child table (issue #1407).
	// The set of FK columns is derived from field metadata (Type), so this stays
	// object-agnostic — no SObject-name branch. Failure to read metadata is
	// non-fatal: standard-column indexing still proceeds.
	relCols, relErr := collectRelationshipFKColumns(rawDB)
	if relErr != nil {
		log.Printf("   ⚠️  Cannot derive relationship FK columns (skipping FK index pass): %v", relErr)
	}
	// relCols is keyed by metadata-derived table names (_System_Object.ApiName or
	// ReadSource.PhysicalTable), whose casing can differ from the live catalog
	// (e.g. lower_case_table_names, or a discovery-registered lowercase table vs
	// PascalCase metadata). Resolve each to the real catalog table name — the keys
	// of existingCols, sourced from INFORMATION_SCHEMA — before lookup, mirroring
	// how the standardCols pass uses getPhysicalTables(). Otherwise a casing
	// divergence makes existingCols[table] nil and silently skips the backfill.
	catalogByLower := make(map[string]string, len(existingCols))
	for t := range existingCols {
		catalogByLower[strings.ToLower(t)] = t
	}
	for table, cols := range relCols {
		realTable, ok := catalogByLower[strings.ToLower(table)]
		if !ok {
			continue
		}
		lower := strings.ToLower(realTable)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}
		existing := existingCols[realTable]
		idxs := existingIndexes[realTable]
		if existing == nil {
			continue
		}
		for col := range cols {
			if !existing[col] {
				continue
			}
			if idxs != nil && idxs[col] {
				continue
			}
			jobs = append(jobs, standardIndexJob{
				table:     realTable,
				indexName: safeIndexName(realTable, col),
				cols:      []string{col},
			})
		}
	}

	// Step 4c: Also index IsNameField columns so that prefix-LIKE suggestions
	// queries (`col LIKE 'term%'`) can use a B-tree index rather than full-scanning
	// the table. Formula-computed and deleted/system fields are excluded — they
	// have no physical column value to index. Failure is non-fatal, same as 4b.
	nameCols, nameErr := collectNameFieldColumns(rawDB)
	if nameErr != nil {
		log.Printf("   ⚠️  Cannot derive name-field columns (skipping name-field index pass): %v", nameErr)
	}
	nameJobCount := 0
	for table, cols := range nameCols {
		realTable, ok := catalogByLower[strings.ToLower(table)]
		if !ok {
			continue
		}
		lower := strings.ToLower(realTable)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}
		existing := existingCols[realTable]
		idxs := existingIndexes[realTable]
		if existing == nil {
			continue
		}
		for col := range cols {
			if !existing[col] {
				continue
			}
			if idxs != nil && idxs[col] {
				continue
			}
			jobs = append(jobs, standardIndexJob{
				table:     realTable,
				indexName: safeIndexName(realTable, col),
				cols:      []string{col},
			})
			nameJobCount++
		}
	}
	log.Printf("   📑 Name-field index jobs added: %d", nameJobCount)

	// Step 4d: Also index direct physical columns referenced by active
	// criteria-based sharing rules. Query-time RLS compiles these criteria
	// into SQL predicates; without an index, large-scope users can still
	// full-scan the base table even after the owner/group predicate is fixed.
	criteriaCols, criteriaErr := collectSharingRuleCriteriaColumns(rawDB)
	if criteriaErr != nil {
		log.Printf("   ⚠️  Cannot derive sharing-rule criteria columns (skipping criteria index pass): %v", criteriaErr)
	}
	criteriaJobCount := 0
	for table, cols := range criteriaCols {
		realTable, ok := catalogByLower[strings.ToLower(table)]
		if !ok {
			continue
		}
		lower := strings.ToLower(realTable)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}
		existing := existingCols[realTable]
		idxs := existingIndexes[realTable]
		if existing == nil {
			continue
		}
		for col := range cols {
			if !existing[col] {
				continue
			}
			if idxs != nil && idxs[col] {
				continue
			}
			jobs = append(jobs, standardIndexJob{
				table:     realTable,
				indexName: safeIndexName(realTable, col),
				cols:      []string{col},
			})
			criteriaJobCount++
		}
	}
	log.Printf("   📑 Sharing-rule criteria index jobs added: %d", criteriaJobCount)

	// Step 4e: Also index email columns. Exact-email lookups are a common read
	// path for login/seed/search flows; without a single-column index they
	// degrade to table scans on large objects like Lead.
	emailCols, emailErr := collectEmailFieldColumns(rawDB)
	if emailErr != nil {
		log.Printf("   ⚠️  Cannot derive email-field columns (skipping email-field index pass): %v", emailErr)
	}
	emailJobCount := 0
	for table, cols := range emailCols {
		realTable, ok := catalogByLower[strings.ToLower(table)]
		if !ok {
			continue
		}
		lower := strings.ToLower(realTable)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}
		existing := existingCols[realTable]
		if existing == nil {
			continue
		}
		for col := range cols {
			if !existing[col] {
				continue
			}
			jobCols := []string{col}
			indexName := safeIndexName(realTable, col)
			if existing["IsDeleted"] {
				jobCols = []string{"IsDeleted", col}
				indexName = safeCompositeIndexName(realTable, "IsDeleted", col)
			}
			jobs = append(jobs, standardIndexJob{
				table:     realTable,
				indexName: indexName,
				cols:      jobCols,
			})
			emailJobCount++
		}
	}
	log.Printf("   📧 Email-field index jobs added: %d", emailJobCount)

	// Step 4f: Reconcile the physical unique index for every active, persisted
	// field marked IsUnique in the metadata SSoT. Pack application normally
	// creates this constraint before publishing IsUnique=true; this pass closes
	// the gap for warm environments where metadata already converged but the
	// physical constraint was skipped. It deliberately uses the existing unique
	// index primitive so a non-unique leading index cannot mask a missing unique
	// constraint.
	uniqueCols, uniqueErr := collectUniqueFieldColumns(rawDB)
	if uniqueErr != nil {
		return fmt.Errorf("derive unique-field index requirements: %w", uniqueErr)
	}
	uniqueJobCount := 0
	for table, cols := range uniqueCols {
		realTable, ok := catalogByLower[strings.ToLower(table)]
		if !ok {
			continue
		}
		lower := strings.ToLower(realTable)
		if strings.HasPrefix(lower, "_system_") ||
			strings.HasPrefix(lower, "_fivetran_") ||
			strings.HasPrefix(lower, "v_") {
			continue
		}
		existing := existingCols[realTable]
		if existing == nil {
			continue
		}
		for col := range cols {
			if !existing[col] {
				continue
			}
			jobs = append(jobs, standardIndexJob{
				table:     realTable,
				indexName: schema.GenerateUniqueIndexName(realTable, col),
				cols:      []string{col},
				unique:    true,
			})
			uniqueJobCount++
		}
	}
	log.Printf("   🔑 Unique-field index jobs added: %d", uniqueJobCount)

	if len(jobs) == 0 {
		log.Printf("✅ Standard indexes ensured: 0 created (%v)", time.Since(start).Round(time.Millisecond))
		return nil
	}

	workers := ensureIndexesWorkerCount()
	log.Printf("   📑 Creating %d indexes (parallel, %d workers)...", len(jobs), workers)

	created, err := executeStandardIndexJobs(rawDB, jobs, workers)
	if err != nil {
		return err
	}

	log.Printf("✅ Standard and search indexes ensured: %d created (%v)", created, time.Since(start).Round(time.Millisecond))
	return nil
}

// collectRelationshipFKColumns returns, per physical table, the set of FK
// columns that back a non-system relationship (lookup / master-detail) field.
// It reads the metadata SSoT (_System_Field joined to its owning object) and
// resolves each object's physical table through the canonical ReadSource
// parser, so it matches the physical-table keys used elsewhere in this routine.
func collectRelationshipFKColumns(db *sql.DB) (map[string]map[string]bool, error) {
	// objectId -> physical table name
	objTable := make(map[string]string)
	objRows, err := db.Query(fmt.Sprintf("SELECT `%s`, `%s`, `%s` FROM `_System_Object`",
		constants.ColSysObject_ID, constants.ColSysObject_APIName, constants.FieldSysObject_ReadSource))
	if err != nil {
		return nil, err
	}
	defer objRows.Close()
	for objRows.Next() {
		var id, apiName string
		var readSource sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSource); err != nil {
			return nil, err
		}
		table := apiName
		if readSource.Valid && strings.TrimSpace(readSource.String) != "" {
			var cfg models.ObjectReadSourceConfig
			if json.Unmarshal([]byte(readSource.String), &cfg) == nil && strings.TrimSpace(cfg.PhysicalTable) != "" {
				table = cfg.PhysicalTable
			}
		}
		objTable[id] = table
	}
	if err := objRows.Err(); err != nil {
		return nil, err
	}

	out := make(map[string]map[string]bool)
	fieldRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, COALESCE(NULLIF(`%s`,''), `%s`) FROM `_System_Field` "+
			"WHERE (`%s` IN ('reference','masterdetail') OR `%s` = 1) "+
			"AND COALESCE(`%s`,0) = 0 AND COALESCE(`%s`,0) = 0",
		constants.ColSysField_ObjectID, constants.ColSysField_PhysicalName, constants.ColSysField_APIName,
		constants.ColSysField_Type, constants.ColSysField_IsMasterDetail,
		constants.ColSysField_IsSystem, constants.ColSysField_IsDeleted,
	))
	if err != nil {
		return nil, err
	}
	defer fieldRows.Close()
	for fieldRows.Next() {
		var objectID, col string
		if err := fieldRows.Scan(&objectID, &col); err != nil {
			return nil, err
		}
		table := objTable[objectID]
		if table == "" || col == "" {
			continue
		}
		if out[table] == nil {
			out[table] = make(map[string]bool)
		}
		out[table][col] = true
	}
	if err := fieldRows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

// collectNameFieldColumns returns, per physical table, the set of columns that
// are marked IsNameField=1 and have an active physical column (not a formula,
// not deleted, not a system field). These columns are indexed to make prefix-
// LIKE suggestions queries (`col LIKE 'term%'`) use the B-tree index instead
// of a full table scan (issue #1597).
func collectNameFieldColumns(db *sql.DB) (map[string]map[string]bool, error) {
	// objectId -> physical table name (same query pattern as collectRelationshipFKColumns)
	objTable := make(map[string]string)
	objRows, err := db.Query(fmt.Sprintf("SELECT `%s`, `%s`, `%s` FROM `%s`",
		constants.ColSysObject_ID, constants.ColSysObject_APIName, constants.FieldSysObject_ReadSource,
		constants.TableObject))
	if err != nil {
		return nil, err
	}
	defer objRows.Close()
	for objRows.Next() {
		var id, apiName string
		var readSource sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSource); err != nil {
			return nil, err
		}
		table := apiName
		if readSource.Valid && strings.TrimSpace(readSource.String) != "" {
			var cfg models.ObjectReadSourceConfig
			if json.Unmarshal([]byte(readSource.String), &cfg) == nil && strings.TrimSpace(cfg.PhysicalTable) != "" {
				table = cfg.PhysicalTable
			}
		}
		objTable[id] = table
	}
	if err := objRows.Err(); err != nil {
		return nil, err
	}

	out := make(map[string]map[string]bool)
	fieldRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, COALESCE(NULLIF(`%s`,''), `%s`) FROM `%s` "+
			"WHERE `%s` = 1 "+
			"AND COALESCE(`%s`,'') != 'active' "+
			"AND COALESCE(`%s`,0) = 0 "+
			"AND COALESCE(`%s`,0) = 0",
		constants.ColSysField_ObjectID,
		constants.ColSysField_PhysicalName, constants.ColSysField_APIName,
		constants.TableField,
		constants.ColSysField_IsNameField,
		constants.ColSysField_FormulaMode,
		constants.ColSysField_IsDeleted,
		constants.ColSysField_IsSystem,
	))
	if err != nil {
		return nil, err
	}
	defer fieldRows.Close()
	for fieldRows.Next() {
		var objectID, col string
		if err := fieldRows.Scan(&objectID, &col); err != nil {
			return nil, err
		}
		table := objTable[objectID]
		if table == "" || col == "" {
			continue
		}
		if out[table] == nil {
			out[table] = make(map[string]bool)
		}
		out[table][col] = true
	}
	if err := fieldRows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

// collectSharingRuleCriteriaColumns returns, per physical table, the set of
// direct physical columns referenced by active criteria-based sharing rules.
// Only direct current-object fields are included; relationship traversals,
// globals, and active-formula/view-only fields are excluded.
func collectSharingRuleCriteriaColumns(db *sql.DB) (map[string]map[string]bool, error) {
	objTable := make(map[string]string)
	objIDByAPIName := make(map[string]string)
	objRows, err := db.Query(fmt.Sprintf("SELECT `%s`, `%s`, `%s` FROM `%s`",
		constants.ColSysObject_ID, constants.ColSysObject_APIName, constants.FieldSysObject_ReadSource,
		constants.TableObject))
	if err != nil {
		return nil, err
	}
	defer objRows.Close()
	for objRows.Next() {
		var id, apiName string
		var readSource sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSource); err != nil {
			return nil, err
		}
		table := apiName
		if readSource.Valid && strings.TrimSpace(readSource.String) != "" {
			var cfg models.ObjectReadSourceConfig
			if json.Unmarshal([]byte(readSource.String), &cfg) == nil && strings.TrimSpace(cfg.PhysicalTable) != "" {
				table = cfg.PhysicalTable
			}
		}
		objTable[id] = table
		objIDByAPIName[strings.ToLower(apiName)] = id
	}
	if err := objRows.Err(); err != nil {
		return nil, err
	}

	fieldColsByObject := make(map[string]map[string]string)
	fieldRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, `%s`, COALESCE(NULLIF(`%s`,''), `%s`) FROM `%s` "+
			"WHERE COALESCE(`%s`,'') != 'active' "+
			"AND COALESCE(`%s`,0) = 0 "+
			"AND COALESCE(`%s`,0) = 0",
		constants.ColSysField_ObjectID,
		constants.ColSysField_APIName,
		constants.ColSysField_PhysicalName, constants.ColSysField_APIName,
		constants.TableField,
		constants.ColSysField_FormulaMode,
		constants.ColSysField_IsDeleted,
		constants.ColSysField_IsSystem,
	))
	if err != nil {
		return nil, err
	}
	defer fieldRows.Close()
	for fieldRows.Next() {
		var objectID, apiName, physicalCol string
		if err := fieldRows.Scan(&objectID, &apiName, &physicalCol); err != nil {
			return nil, err
		}
		if objectID == "" || apiName == "" || physicalCol == "" {
			continue
		}
		if fieldColsByObject[objectID] == nil {
			fieldColsByObject[objectID] = make(map[string]string)
		}
		fieldColsByObject[objectID][strings.ToLower(apiName)] = physicalCol
		fieldColsByObject[objectID][strings.ToLower(physicalCol)] = physicalCol
	}
	if err := fieldRows.Err(); err != nil {
		return nil, err
	}

	out := make(map[string]map[string]bool)
	ruleRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, `%s` FROM `%s` "+
			"WHERE COALESCE(`%s`,0) = 0 "+
			"AND COALESCE(NULLIF(TRIM(`%s`),''),'') != ''",
		constants.FieldObjectAPIName,
		constants.FieldSharingRule_Criteria,
		constants.TableSharingRule,
		constants.FieldIsDeleted,
		constants.FieldSharingRule_Criteria,
	))
	if err != nil {
		return nil, err
	}
	defer ruleRows.Close()
	for ruleRows.Next() {
		var objectAPIName, criteria string
		if err := ruleRows.Scan(&objectAPIName, &criteria); err != nil {
			return nil, err
		}
		objectID := objIDByAPIName[strings.ToLower(strings.TrimSpace(objectAPIName))]
		table := objTable[objectID]
		fieldCols := fieldColsByObject[objectID]
		if objectID == "" || table == "" || len(fieldCols) == 0 {
			continue
		}
		refs, err := formula.CollectCurrentObjectFieldReferences(criteria)
		if err != nil {
			continue
		}
		for _, ref := range refs {
			col := fieldCols[strings.ToLower(ref)]
			if col == "" {
				continue
			}
			if out[table] == nil {
				out[table] = make(map[string]bool)
			}
			out[table][col] = true
		}
	}
	if err := ruleRows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

// collectUniqueFieldColumns returns, per physical table, active persisted
// fields whose metadata declares IsUnique=true. The metadata row is the source
// of truth; EnsureStandardIndexes uses this only to reconcile a missing
// physical unique constraint on warm environments.
func collectUniqueFieldColumns(db *sql.DB) (map[string]map[string]bool, error) {
	objTable := make(map[string]string)
	objRows, err := db.Query(fmt.Sprintf("SELECT `%s`, `%s`, `%s` FROM `%s`",
		constants.ColSysObject_ID, constants.ColSysObject_APIName, constants.FieldSysObject_ReadSource,
		constants.TableObject))
	if err != nil {
		return nil, err
	}
	defer objRows.Close()
	for objRows.Next() {
		var id, apiName string
		var readSource sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSource); err != nil {
			return nil, err
		}
		table := apiName
		if readSource.Valid && strings.TrimSpace(readSource.String) != "" {
			var cfg models.ObjectReadSourceConfig
			if json.Unmarshal([]byte(readSource.String), &cfg) == nil && strings.TrimSpace(cfg.PhysicalTable) != "" {
				table = cfg.PhysicalTable
			}
		}
		objTable[id] = table
	}
	if err := objRows.Err(); err != nil {
		return nil, err
	}

	out := make(map[string]map[string]bool)
	fieldRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, COALESCE(NULLIF(`%s`,''), `%s`), COALESCE(`%s`,''), COALESCE(`%s`,''), COALESCE(`%s`,''), COALESCE(`%s`,'') FROM `%s` "+
			"WHERE COALESCE(`%s`,0) = 1 "+
			"AND COALESCE(`%s`,0) = 0 "+
			"AND COALESCE(`%s`,0) = 0",
		constants.ColSysField_ObjectID,
		constants.ColSysField_PhysicalName, constants.ColSysField_APIName,
		constants.ColSysField_FormulaMode,
		constants.FieldSysField_FormulaStorageMode,
		constants.FieldSysField_Formula,
		constants.FieldSysField_ProjectionConfig,
		constants.TableField,
		constants.FieldSysField_IsUnique,
		constants.ColSysField_IsDeleted,
		constants.ColSysField_IsSystem,
	))
	if err != nil {
		return nil, err
	}
	defer fieldRows.Close()
	for fieldRows.Next() {
		var objectID, col, formulaMode, formulaStorageMode, formulaExpr, projectionConfig string
		if err := fieldRows.Scan(&objectID, &col, &formulaMode, &formulaStorageMode, &formulaExpr, &projectionConfig); err != nil {
			return nil, err
		}
		if strings.TrimSpace(formulaMode) != "" || strings.TrimSpace(formulaStorageMode) != "" ||
			strings.TrimSpace(formulaExpr) != "" || strings.TrimSpace(projectionConfig) != "" {
			resolved, err := formulastorage.Resolve(formulastorage.Spec{
				Formula:             formulaExpr,
				FormulaMode:         formulaMode,
				FormulaStorageMode:  formulaStorageMode,
				HasFormula:          strings.TrimSpace(formulaMode) != "" || strings.TrimSpace(formulaExpr) != "",
				HasProjectionConfig: strings.TrimSpace(projectionConfig) != "",
			})
			if err != nil {
				return nil, fmt.Errorf("resolve formula storage for unique field %s: %w", col, err)
			}
			if !resolved.IsWriteMaterialized() {
				continue
			}
		}
		table := objTable[objectID]
		if table == "" || col == "" {
			continue
		}
		if out[table] == nil {
			out[table] = make(map[string]bool)
		}
		out[table][col] = true
	}
	if err := fieldRows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

// collectEmailFieldColumns returns, per physical table, the set of active
// physical columns whose metadata type is email. These columns are indexed so
// exact-email lookups can use a B-tree instead of full-scanning large tables.
func collectEmailFieldColumns(db *sql.DB) (map[string]map[string]bool, error) {
	objTable := make(map[string]string)
	objRows, err := db.Query(fmt.Sprintf("SELECT `%s`, `%s`, `%s` FROM `%s`",
		constants.ColSysObject_ID, constants.ColSysObject_APIName, constants.FieldSysObject_ReadSource,
		constants.TableObject))
	if err != nil {
		return nil, err
	}
	defer objRows.Close()
	for objRows.Next() {
		var id, apiName string
		var readSource sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSource); err != nil {
			return nil, err
		}
		table := apiName
		if readSource.Valid && strings.TrimSpace(readSource.String) != "" {
			var cfg models.ObjectReadSourceConfig
			if json.Unmarshal([]byte(readSource.String), &cfg) == nil && strings.TrimSpace(cfg.PhysicalTable) != "" {
				table = cfg.PhysicalTable
			}
		}
		objTable[id] = table
	}
	if err := objRows.Err(); err != nil {
		return nil, err
	}

	out := make(map[string]map[string]bool)
	fieldRows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, COALESCE(NULLIF(`%s`,''), `%s`) FROM `%s` "+
			"WHERE `%s` = '%s' "+
			"AND COALESCE(`%s`,'') != 'active' "+
			"AND COALESCE(`%s`,0) = 0 "+
			"AND COALESCE(`%s`,0) = 0",
		constants.ColSysField_ObjectID,
		constants.ColSysField_PhysicalName, constants.ColSysField_APIName,
		constants.TableField,
		constants.ColSysField_Type, constants.FieldTypeEmail,
		constants.ColSysField_FormulaMode,
		constants.ColSysField_IsDeleted,
		constants.ColSysField_IsSystem,
	))
	if err != nil {
		return nil, err
	}
	defer fieldRows.Close()
	for fieldRows.Next() {
		var objectID, col string
		if err := fieldRows.Scan(&objectID, &col); err != nil {
			return nil, err
		}
		table := objTable[objectID]
		if table == "" || col == "" {
			continue
		}
		if out[table] == nil {
			out[table] = make(map[string]bool)
		}
		out[table][col] = true
	}
	if err := fieldRows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

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

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

// reconcilePlatformDiscoveredFieldOverrides corrects platform-owned field
// metadata where the discover path conflicts with engine write paths. App-pack
// object/field facts belong in pack payload and are not repaired here.
func reconcilePlatformDiscoveredFieldOverrides(db *sql.DB) {
	type fieldOverride struct {
		object    string
		field     string
		setClause string
		// When true, the UPDATE additionally guards with `f.IsCustomized = 0
		// OR f.IsCustomized IS NULL` so an admin who changed the value via
		// the field-edit UI (which sets `IsCustomized=1`) is not silently
		// overwritten on every startup.
		//
		// Inference-correction entries enable the guard because the override
		// only exists to repair a generic builder default; explicit admin
		// intent should win.
		preserveAdminCustomization bool
	}

	overrides := []fieldOverride{
		{object: "User", field: "Password", setClause: "f.Required = 0", preserveAdminCustomization: true},
		{object: "User", field: "FailedLoginAttempts", setClause: "f.Required = 0", preserveAdminCustomization: true},
		// Issue #974: older live _System_Field rows may still carry Required=1
		// even after User.FirstName became nullable in system_tables.json.
		//
		// Unlike label/help/default payload on a system field, Required is
		// not admin-authorable here; leaving this behind IsCustomized would
		// strand warm-upgraded orgs that ever customized some other editable
		// column on the row.
		{object: "User", field: "FirstName", setClause: "f.Required = 0"},
	}

	var totalFixed int64
	for _, ov := range overrides {
		where := "WHERE o.ApiName = ? AND f.ApiName = ?"
		if ov.preserveAdminCustomization {
			where += " AND (f.IsCustomized = 0 OR f.IsCustomized IS NULL)"
		}
		query := fmt.Sprintf(
			"UPDATE `%s` f JOIN `%s` o ON o.Id = f.ObjectId SET %s %s",
			constants.TableField, constants.TableObject, ov.setClause, where,
		)
		result, err := db.Exec(query, ov.object, ov.field)
		if err != nil {
			log.Printf("   ⚠️  Failed to fix %s.%s metadata: %v", ov.object, ov.field, err)
			continue
		}
		if affected, _ := result.RowsAffected(); affected > 0 {
			totalFixed += affected
		}
	}
	if totalFixed > 0 {
		log.Printf("   🔧 Corrected %d platform discovered field metadata overrides", totalFixed)
	}
}

// BackfillFieldCapabilities enforces platform-universal non-writeability
// bounds on _System_Field.Createable/Updateable. It intentionally does not
// turn false values back to true: application packs and tenant overlays may
// declare normal fields read-only, and that metadata is the live SSoT.
// Idempotent — safe to run on every startup.
// This covers fields that were inserted before the Createable/Updateable columns existed,
// or fields inserted via raw SQL (discover path) that bypassed ComputeIntrinsic.
//
// Note: the SELECT and minimal FieldMetadata MUST include every field that
// ComputeIntrinsic reads. Otherwise universal platform bounds become dead code
// during backfill.
func BackfillFieldCapabilities(dbConn *database.TiDBConnection) error {
	db := dbConn.DB()

	// Read all fields with their object API name, then use the shared
	// platform-intrinsic function only as a lower bound. CRM/application
	// capabilities are stored on _System_Field by their owning pack payload.
	// Source / IsCompound are no longer stored; ComputeIntrinsic Layer 1d/1e
	// consult fieldprovenance.Classify* derived from L1 manifests (RFC v1.6).
	rows, err := db.Query(
		fmt.Sprintf("SELECT f.Id, f.ApiName, f.Type, f.Formula, f.Createable, f.Updateable, o.ApiName AS ObjApiName "+
			"FROM `%s` f JOIN `%s` o ON o.Id = f.ObjectId WHERE f.IsDeleted = 0",
			constants.TableField, constants.TableObject))
	if err != nil {
		return fmt.Errorf("backfill: failed to query fields: %w", err)
	}
	defer rows.Close()

	type fieldRow struct {
		id, apiName, objAPIName string
		fieldType               string
		formula                 sql.NullString
		curCreate, curUpdate    bool
	}
	var toUpdate []fieldRow

	for rows.Next() {
		var r fieldRow
		if err := rows.Scan(&r.id, &r.apiName, &r.fieldType, &r.formula, &r.curCreate, &r.curUpdate, &r.objAPIName); err != nil {
			return fmt.Errorf("backfill: scan field row: %w", err)
		}
		// Minimal FieldMetadata for ComputeIntrinsic. Layer 1d/1e delegate to
		// fieldprovenance.Classify* using the (object, field) pair — no
		// stored Source / IsCompound struct fields required.
		fm := &models.FieldMetadata{APIName: r.apiName, Type: models.FieldType(r.fieldType)}
		if r.formula.Valid && r.formula.String != "" {
			fm.Formula = &r.formula.String
		}
		boundC, boundU := fieldcapabilities.ComputeIntrinsic(r.objAPIName, fm)
		wantC := r.curCreate && boundC
		wantU := r.curUpdate && boundU
		if wantC != r.curCreate || wantU != r.curUpdate {
			r.curCreate = wantC
			r.curUpdate = wantU
			toUpdate = append(toUpdate, r)
		}
	}

	if err := rows.Err(); err != nil {
		return fmt.Errorf("backfill: iterate field rows: %w", err)
	}

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

	// Batch update in chunks. Propagate UPDATE errors: a swallowed failure here
	// would leave capability bounds unconverged while ApplyAll still stamps the
	// completed marker, so restart would skip the repair and serve stale
	// Createable/Updateable bits (#1590 marker-honesty).
	for _, r := range toUpdate {
		if _, err := db.Exec(fmt.Sprintf("UPDATE `%s` SET Createable = ?, Updateable = ? WHERE Id = ?",
			constants.TableField), r.curCreate, r.curUpdate, r.id); err != nil {
			return fmt.Errorf("backfill: update field %s capabilities: %w", r.id, err)
		}
	}
	log.Printf("   🔧 Backfilled Createable/Updateable for %d fields", len(toUpdate))
	return nil
}