package persistence

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	pkgerrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// SchemaProvider is a narrow interface for looking up object metadata by API name.
// MetadataService satisfies this interface, allowing RecordRepository to delegate
// schema lookups to the single authoritative metadata cache instead of maintaining
// its own duplicate cache.
type SchemaProvider interface {
	GetSchema(ctx context.Context, apiName string) *models.ObjectMetadata
}

// RecordRepository handles dynamic CRUD operations for any object.
// Method implementations are split across domain-specific files:
//   - record_bulk.go        — BulkInsert, bulkInsertRaw, writeEscapedValue
//   - record_recycle_bin.go — FindRecycleBinItemsByUserPage, FindAllRecycleBinItems(Page), FindRecycleBinEntry
type RecordRepository struct {
	db                             *sql.DB
	schemaProvider                 SchemaProvider
	viewResolver                   ports.ViewResolver
	openIndependentObjectIDProbeDB func() (*sql.DB, error)
}

const independentObjectIDProbeBudget = 250 * time.Millisecond

// objectIDProbeConcurrency bounds the internal fan-out used only for
// non-transactional reverse record-ID lookup. Runtime admission remains the
// request-level concurrency owner; this limit prevents one accepted untyped
// batch from turning a table-per-object scan into unbounded pool pressure.
const objectIDProbeConcurrency = 8

// objectTypeCandidateProbeChunkSize bounds the number of physical read tables
// composed into one non-transactional candidate query. Candidate discovery
// must inspect every table to preserve collision detection, but it does not
// need one driver round trip per table. Sixteen keeps SQL/parameter size
// bounded while reducing a large Catalog to a small number of statements.
const objectTypeCandidateProbeChunkSize = 16

var errIndependentObjectIDProbeUnavailable = errors.New("independent object identity probe database is unavailable")

// RecordRepositoryOption configures an optional repository capability.
type RecordRepositoryOption func(*RecordRepository)

// WithIndependentObjectIDProbeDB provides a non-blocking opener for a newly
// constructed empty sql.DB pool that uses the same database authority as the
// primary pool. The repository owns and closes each returned pool.
func WithIndependentObjectIDProbeDB(opener func() (*sql.DB, error)) RecordRepositoryOption {
	return func(repository *RecordRepository) {
		repository.openIndependentObjectIDProbeDB = opener
	}
}

// NewRecordRepository creates a new RecordRepository.
// Call SetSchemaProvider after construction to wire the MetadataService cache;
// without it, schema lookups fall back to direct database queries.
func NewRecordRepository(db *sql.DB, options ...RecordRepositoryOption) *RecordRepository {
	repository := &RecordRepository{db: db}
	for _, option := range options {
		if option != nil {
			option(repository)
		}
	}
	return repository
}

// SetSchemaProvider wires a SchemaProvider (typically MetadataService) so that
// schema lookups use the single authoritative metadata cache instead of
// hitting the database directly.
func (r *RecordRepository) SetSchemaProvider(sp SchemaProvider) {
	r.schemaProvider = sp
}

// SetViewResolver injects the ViewResolver so single-record reads route
// through the formula view (v_<object>) when the object has active formulas
// and a view exists. Without this, FindOne hits the physical table and
// formula columns come back as the stored snapshot (usually NULL).
func (r *RecordRepository) SetViewResolver(vr ports.ViewResolver) {
	r.viewResolver = vr
}

// physicalReadOnlyCtxKey marks a context whose single-record reads must target
// the base table, skipping the formula view. Callers set it only when they
// consume nothing but physical columns (existence checks, FK/relationship
// discovery, lock loads); formula and rollup-summary columns come back as the
// stored snapshot (usually NULL) under this flag, so consumers that read a
// computed value must NOT set it.
type physicalReadOnlyCtxKey struct{}

// WithPhysicalReadOnly returns a context that routes single-record reads
// (FindOne) to the base table instead of v_<object>. See physicalReadOnlyCtxKey
// for the safety contract. Used by the write-path validation/discovery/lock
// callers to avoid pulling 143-350 wide-view columns merely to check existence
// or read a foreign key.
func WithPhysicalReadOnly(ctx context.Context) context.Context {
	return context.WithValue(ctx, physicalReadOnlyCtxKey{}, true)
}

func physicalReadOnlyFromContext(ctx context.Context) bool {
	v, _ := ctx.Value(physicalReadOnlyCtxKey{}).(bool)
	return v
}

// resolveReadTable returns the table name to use for SELECT queries.
// When a ViewResolver is configured and the schema has active formulas,
// returns the view name (v_<object>); otherwise returns the raw table name
// passed in. A physical-read-only context forces the base table regardless of
// the view, for callers that consume only physical columns.
func (r *RecordRepository) resolveReadTable(ctx context.Context, tableName string, schema *models.ObjectMetadata) string {
	if r.viewResolver == nil || schema == nil {
		return tableName
	}
	if physicalReadOnlyFromContext(ctx) {
		return tableName
	}
	return r.viewResolver.ResolveTableName(ctx, schema)
}

// GetExecutor returns the transactional executor if present in context, or the DB connection
func (r *RecordRepository) GetExecutor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

// getSchema loads schema for aliasing. When a SchemaProvider is configured
// (the normal production path via MetadataService), it delegates to the
// provider's cached lookup. Otherwise it falls back to a direct DB query.
func (r *RecordRepository) getSchema(ctx context.Context, tableName string) (*models.ObjectMetadata, error) {
	// Prefer the injected provider (MetadataService) — single cache authority
	if r.schemaProvider != nil {
		return r.schemaProvider.GetSchema(ctx, tableName), nil
	}

	// Fallback: direct DB lookup (no local caching)
	metaRepo := NewMetadataRepository(r.db)
	return metaRepo.GetSchemaByAPIName(ctx, tableName)
}

// Exists checks if a record exists by ID
func (r *RecordRepository) Exists(ctx context.Context, tableName string, id string) (bool, error) {
	if !isValidFieldName(tableName) {
		return false, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)

	queryP := query.From(tableName).
		WithMetadata(schema).
		Select([]string{constants.FieldID}).
		Where(constants.FieldID, id).
		Limit(1).
		Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, queryP.SQL, queryP.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.Exists", tableName, tableName, queryP.SQL, len(queryP.Params), 1, 0, sqlStart, err)
		return false, err
	}
	defer rows.Close()

	exists := rows.Next()
	rowCount := 0
	if exists {
		rowCount = 1
	}
	logRepositorySQLTiming(ctx, "RecordRepository.Exists", tableName, tableName, queryP.SQL, len(queryP.Params), 1, rowCount, sqlStart, nil)
	return exists, nil
}

// GetLock retrieves a record by ID and locks it (SELECT ... FOR UPDATE)
func (r *RecordRepository) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	// Must be in a transaction for locking to work effectively
	// We check context for transaction
	exec := r.GetExecutor(ctx)
	if _, ok := exec.(*sql.Tx); !ok {
		// Strict mode: Locking requires transaction
		return nil, pkgerrors.NewValidationError("transaction", fmt.Sprintf("transaction required for locking record %s", id))
	}

	schema, _ := r.getSchema(ctx, tableName)

	q := query.From(tableName).
		WithMetadata(schema).
		Select([]string{"*"}).
		Where(constants.FieldID, id).
		Limit(1).
		Build()

	// Inject FOR UPDATE
	q.SQL += " FOR UPDATE"

	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.GetLock", tableName, tableName, q.SQL, len(q.Params), 0, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.GetLock", tableName, tableName, q.SQL, len(q.Params), 0, len(results), sqlStart, err)
	if err != nil {
		return nil, err
	}

	if len(results) == 0 {
		return nil, nil
	}

	result := results[0]

	return r.normalizeRecord(result, schema), nil
}

// remapPhysicalToLogical converts physical column names back to logical API names
// It strictly enforces the casing defined in the Schema (PascalCase) regardless of DB driver output.
func (r *RecordRepository) remapPhysicalToLogical(record models.SObject, schema *models.ObjectMetadata) models.SObject {
	newRecord := make(models.SObject)

	// Build Case-Insensitive Mapping: { "lowercase_name" -> "PascalCaseName" }
	// Priority:
	// 1. PhysicalName (Alias)
	// 2. APIName (Standard)
	toLogical := make(map[string]string)

	for _, f := range schema.Fields {
		// Map APIName (lower) -> APIName (Pascal)
		toLogical[strings.ToLower(f.APIName)] = f.APIName

		// Map PhysicalName (lower) -> APIName (Pascal)
		if f.PhysicalName != "" {
			toLogical[strings.ToLower(f.PhysicalName)] = f.APIName
		}
	}

	for k, v := range record {
		lowerKey := strings.ToLower(k)

		// Check if matches a known field
		if logical, ok := toLogical[lowerKey]; ok {
			newRecord[logical] = v
		} else {
			// If not found in schema, keep original (e.g., aggregate results or system columns)
			// But check standard ID field specifically
			if lowerKey == "id" {
				newRecord[constants.FieldID] = v
			} else {
				newRecord[k] = v
			}
		}
	}
	return newRecord
}

// normalizeRecord converts storage rows into API-shaped records by remapping
// physical column names back to logical API names and hydrating field types.
func (r *RecordRepository) normalizeRecord(record models.SObject, schema *models.ObjectMetadata) models.SObject {
	if record == nil || schema == nil {
		return record
	}

	record = r.remapPhysicalToLogical(record, schema)
	return schemautil.FromStorageRecord(schema, record, nil)
}

func mapRecordMutationDuplicateKeyError(tableName string, err error) error {
	if err == nil || !IsDuplicateKeyError(err) {
		return err
	}
	keyName := ExtractDuplicateKeyName(err.Error())
	return fmt.Errorf("%w: %w", ports.ErrDatabaseDuplicateKey, NewDuplicateKeyError(tableName, keyName))
}

// Insert executes an INSERT statement
func (r *RecordRepository) Insert(ctx context.Context, tableName string, record models.SObject) error {
	if !isValidFieldName(tableName) {
		return pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	// Live schema resolves aliases; the embedded catalog owns authority policy.
	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectAuthorityOnlyFieldWrites(tableName, schema, record); err != nil {
		return err
	}
	exec := r.GetExecutor(ctx)
	if _, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema); err != nil {
		return pkgerrors.NewInternalError("inspect managed row mutation storage", err)
	}

	// Flatten JSON maps for driver compatibility
	flattenedRecord := make(models.SObject)
	for k, v := range record {
		if m, ok := v.(map[string]interface{}); ok {
			b, err := json.Marshal(m)
			if err != nil {
				flattenedRecord[k] = "{}"
			} else {
				flattenedRecord[k] = string(b)
			}
		} else if mp, ok := v.(schemautil.NormalizedMultiPicklistJSON); ok {
			flattenedRecord[k] = string(mp)
		} else {
			flattenedRecord[k] = v
		}
	}

	builder := query.Insert(tableName, flattenedRecord)
	if schema != nil {
		builder.WithMetadata(schema)
	}
	q := builder.Build()

	sqlStart := time.Now()
	_, err := exec.ExecContext(ctx, q.SQL, q.Params...)
	logRepositorySQLTiming(ctx, "RecordRepository.Insert", tableName, tableName, q.SQL, len(q.Params), len(flattenedRecord), 0, sqlStart, err)
	if err != nil {
		return mapRecordMutationDuplicateKeyError(tableName, err)
	}
	return nil
}

// Update executes an UPDATE statement
func (r *RecordRepository) Update(ctx context.Context, tableName string, id string, updates models.SObject) error {

	if !isValidFieldName(tableName) {
		return pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectAuthorityOnlyFieldWrites(tableName, schema, updates); err != nil {
		return err
	}
	if fieldMetadataDeletionAuthored(updates, schema) {
		if err := RejectGenericFieldMetadataRemoval(tableName, schema); err != nil {
			return err
		}
	}
	exec := r.GetExecutor(ctx)
	mutationProtection, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema)
	if err != nil {
		return pkgerrors.NewInternalError("inspect managed row mutation storage", err)
	}

	// Flatten JSON maps for driver compatibility
	flattenedUpdates := make(models.SObject)
	for k, v := range updates {
		if m, ok := v.(map[string]interface{}); ok {
			b, err := json.Marshal(m)
			if err != nil {
				flattenedUpdates[k] = "{}"
			} else {
				flattenedUpdates[k] = string(b)
			}
		} else if mp, ok := v.(schemautil.NormalizedMultiPicklistJSON); ok {
			flattenedUpdates[k] = string(mp)
		} else {
			flattenedUpdates[k] = v
		}
	}

	builder := query.Update(tableName).
		WithMetadata(schema).
		Set(flattenedUpdates).
		Where(constants.FieldID, id)
	if predicate := mutationProtection.mutablePredicate(); predicate != "" {
		builder.WhereRaw(predicate, nil)
	}

	// Build query
	q := builder.Build()

	// Execute
	sqlStart := time.Now()
	result, err := exec.ExecContext(ctx, q.SQL, q.Params...)
	logRepositorySQLTiming(ctx, "RecordRepository.Update", tableName, tableName, q.SQL, len(q.Params), len(flattenedUpdates), 0, sqlStart, err)
	if err != nil {
		return mapRecordMutationDuplicateKeyError(tableName, err)
	}
	return finishManagedRowMutation(ctx, exec, schema, id, "updated", mutationProtection, result)
}

// Delete executes a DELETE statement
func (r *RecordRepository) Delete(ctx context.Context, tableName string, id string) error {
	if !isValidFieldName(tableName) {
		return pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectGenericFieldMetadataRemoval(tableName, schema); err != nil {
		return err
	}
	if err := RejectGenericProtectedCascadeParentRemoval(tableName, schema); err != nil {
		return err
	}
	exec := r.GetExecutor(ctx)
	mutationProtection, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema)
	if err != nil {
		return pkgerrors.NewInternalError("inspect managed row mutation storage", err)
	}

	builder := query.Delete(tableName).
		WithMetadata(schema).
		Where(constants.FieldID, id)
	if predicate := mutationProtection.mutablePredicate(); predicate != "" {
		builder.WhereRaw(predicate, nil)
	}
	q := builder.Build()

	sqlStart := time.Now()
	result, err := exec.ExecContext(ctx, q.SQL, q.Params...)
	logRepositorySQLTiming(ctx, "RecordRepository.Delete", tableName, tableName, q.SQL, len(q.Params), 1, 0, sqlStart, err)
	if err != nil {
		return err
	}
	return finishManagedRowMutation(ctx, exec, schema, id, "deleted", mutationProtection, result)
}

// GetChildren retrieves active child records for cascade delete
func (r *RecordRepository) GetChildren(ctx context.Context, childTable string, foreignKey string, parentID string) ([]models.SObject, error) {
	if !isValidFieldName(childTable) {
		return nil, pkgerrors.NewValidationError("childTable", fmt.Sprintf("invalid child table name: %s", childTable))
	}
	if !isValidFieldName(foreignKey) {
		return nil, pkgerrors.NewValidationError("foreignKey", fmt.Sprintf("invalid foreign key field name: %s", foreignKey))
	}

	schema, _ := r.getSchema(ctx, childTable)

	b := query.From(childTable).
		WithMetadata(schema).
		Select([]string{constants.FieldID}).
		Where(foreignKey, parentID)
	if schema != nil && schemaHasIsDeleted(schema) {
		b = b.ExcludeDeleted()
	}
	q := b.Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.GetChildren", childTable, childTable, q.SQL, len(q.Params), 1, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.GetChildren", childTable, childTable, q.SQL, len(q.Params), 1, len(results), sqlStart, err)
	return results, err
}

// GetDeletedChildren retrieves soft-deleted child records for recycle-bin purge traversal.
func (r *RecordRepository) GetDeletedChildren(ctx context.Context, childTable string, foreignKey string, parentID string) ([]models.SObject, error) {
	if !isValidFieldName(childTable) {
		return nil, pkgerrors.NewValidationError("childTable", fmt.Sprintf("invalid child table name: %s", childTable))
	}
	if !isValidFieldName(foreignKey) {
		return nil, pkgerrors.NewValidationError("foreignKey", fmt.Sprintf("invalid foreign key field name: %s", foreignKey))
	}

	schema, _ := r.getSchema(ctx, childTable)
	if schema == nil || !schemaHasIsDeleted(schema) {
		return nil, nil
	}

	q := query.From(childTable).
		WithMetadata(schema).
		Select([]string{constants.FieldID}).
		Where(foreignKey, parentID).
		OnlyDeleted().
		Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.GetDeletedChildren", childTable, childTable, q.SQL, len(q.Params), 1, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.GetDeletedChildren", childTable, childTable, q.SQL, len(q.Params), 1, len(results), sqlStart, err)
	return results, err
}

// ExistsByField checks if any record exists matching a specific field value
func (r *RecordRepository) ExistsByField(ctx context.Context, tableName string, fieldName string, value interface{}) (bool, error) {
	if !isValidFieldName(tableName) {
		return false, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	if !isValidFieldName(fieldName) {
		return false, pkgerrors.NewValidationError("fieldName", fmt.Sprintf("invalid field name: %s", fieldName))
	}

	schema, _ := r.getSchema(ctx, tableName)

	b := query.From(tableName).
		WithMetadata(schema).
		Select([]string{constants.FieldID}).
		Where(fieldName, value)
	if schema != nil && schemaHasIsDeleted(schema) {
		b = b.ExcludeDeleted()
	}
	q := b.Limit(1).Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.ExistsByField", tableName, tableName, q.SQL, len(q.Params), 1, 0, sqlStart, err)
		return false, err
	}
	defer rows.Close()

	exists := rows.Next()
	rowCount := 0
	if exists {
		rowCount = 1
	}
	logRepositorySQLTiming(ctx, "RecordRepository.ExistsByField", tableName, tableName, q.SQL, len(q.Params), 1, rowCount, sqlStart, nil)
	return exists, nil
}

// FindOneByField returns a single record matching a specific field value
func (r *RecordRepository) FindOneByField(ctx context.Context, tableName string, fieldName string, value interface{}) (models.SObject, error) {
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	if !isValidFieldName(fieldName) {
		return nil, pkgerrors.NewValidationError("fieldName", fmt.Sprintf("invalid field name: %s", fieldName))
	}

	schema, _ := r.getSchema(ctx, tableName)

	b := query.From(tableName).
		WithMetadata(schema).
		Select([]string{"*"}).
		Where(fieldName, value)
	if schema != nil && schemaHasIsDeleted(schema) {
		b = b.ExcludeDeleted()
	}
	q := b.Limit(1).Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.FindOneByField", tableName, tableName, q.SQL, len(q.Params), 0, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.FindOneByField", tableName, tableName, q.SQL, len(q.Params), 0, len(results), sqlStart, err)
	if err != nil {
		return nil, err
	}
	if len(results) == 0 {
		return nil, nil
	}
	return r.normalizeRecord(results[0], schema), nil
}

// FindOne returns a single record by ID, excluding deleted
func (r *RecordRepository) FindOne(ctx context.Context, tableName string, id string) (models.SObject, error) {
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)
	readTable := r.resolveReadTable(ctx, tableName, schema)

	b := query.From(readTable).
		WithMetadata(schema).
		Select([]string{"*"}).
		Where(constants.FieldID, id)
	if schema != nil && schemaHasIsDeleted(schema) {
		b = b.ExcludeDeleted()
	}
	q := b.Limit(1).Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.FindOne", tableName, readTable, q.SQL, len(q.Params), 0, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.FindOne", tableName, readTable, q.SQL, len(q.Params), 0, len(results), sqlStart, err)
	if err != nil {
		return nil, err
	}
	if len(results) == 0 {
		return nil, nil
	}
	return r.normalizeRecord(results[0], schema), nil
}

// FindMany returns multiple records by IDs, excluding deleted. Returns a map of ID -> record.
func (r *RecordRepository) FindMany(ctx context.Context, tableName string, ids []string) (map[string]models.SObject, error) {
	if len(ids) == 0 {
		return map[string]models.SObject{}, nil
	}
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)

	vals := make([]interface{}, len(ids))
	for i, v := range ids {
		vals[i] = v
	}

	readTable := r.resolveReadTable(ctx, tableName, schema)

	b := query.From(readTable).
		WithMetadata(schema).
		Select([]string{"*"}).
		WhereIn(constants.FieldID, vals)
	if schema != nil && schemaHasIsDeleted(schema) {
		b = b.ExcludeDeleted()
	}
	q := b.Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.FindMany", tableName, readTable, q.SQL, len(q.Params), 0, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.FindMany", tableName, readTable, q.SQL, len(q.Params), 0, len(results), sqlStart, err)
	if err != nil {
		return nil, err
	}

	resultMap := make(map[string]models.SObject, len(results))
	for _, rec := range results {
		rec = r.normalizeRecord(rec, schema)
		if id, ok := rec[constants.FieldID].(string); ok {
			resultMap[id] = rec
		}
	}
	return resultMap, nil
}

// FindAny returns a record by ID, INCLUDING deleted ones
func (r *RecordRepository) FindAny(ctx context.Context, tableName string, id string) (models.SObject, error) {
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)

	q := query.From(tableName).
		WithMetadata(schema).
		Select([]string{"*"}).
		Where(constants.FieldID, id).
		Limit(1).
		Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.FindAny", tableName, tableName, q.SQL, len(q.Params), 0, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	logRepositorySQLTiming(ctx, "RecordRepository.FindAny", tableName, tableName, q.SQL, len(q.Params), 0, len(results), sqlStart, err)
	if err != nil {
		return nil, err
	}
	if len(results) == 0 {
		return nil, nil
	}
	return r.normalizeRecord(results[0], schema), nil
}

// PhysicalDelete permanently removes a record
func (r *RecordRepository) PhysicalDelete(ctx context.Context, tableName string, id string) error {
	if !isValidFieldName(tableName) {
		return pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectGenericFieldMetadataRemoval(tableName, schema); err != nil {
		return err
	}
	if err := RejectGenericProtectedCascadeParentRemoval(tableName, schema); err != nil {
		return err
	}
	exec := r.GetExecutor(ctx)
	mutationProtection, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema)
	if err != nil {
		return pkgerrors.NewInternalError("inspect managed row mutation storage", err)
	}

	builder := query.Delete(tableName).
		WithMetadata(schema).
		Where(constants.FieldID, id)
	if predicate := mutationProtection.mutablePredicate(); predicate != "" {
		builder.WhereRaw(predicate, nil)
	}
	q := builder.Build()

	sqlStart := time.Now()
	result, err := exec.ExecContext(ctx, q.SQL, q.Params...)
	logRepositorySQLTiming(ctx, "RecordRepository.PhysicalDelete", tableName, tableName, q.SQL, len(q.Params), 1, 0, sqlStart, err)
	if err != nil {
		return err
	}
	return finishManagedRowMutation(ctx, exec, schema, id, "physically deleted", mutationProtection, result)
}

// DeleteByField deletes records matching a specific field value
func (r *RecordRepository) DeleteByField(ctx context.Context, tableName string, fieldName string, value interface{}) error {
	if !isValidFieldName(tableName) {
		return pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}
	if !isValidFieldName(fieldName) {
		return pkgerrors.NewValidationError("fieldName", fmt.Sprintf("invalid field name: %s", fieldName))
	}

	schema, _ := r.getSchema(ctx, tableName)
	if err := RejectGenericFieldMetadataRemoval(tableName, schema); err != nil {
		return err
	}
	if err := RejectGenericProtectedCascadeParentRemoval(tableName, schema); err != nil {
		return err
	}
	exec := r.GetExecutor(ctx)
	mutationProtection, err := managedRowMutationProtectionFor(ctx, exec, tableName, schema)
	if err != nil {
		return pkgerrors.NewInternalError("inspect managed row mutation storage", err)
	}
	if err := rejectManagedRowMutationBulkDelete(tableName, mutationProtection); err != nil {
		return err
	}

	q := query.Delete(tableName).
		WithMetadata(schema).
		Where(fieldName, value).
		Build()

	sqlStart := time.Now()
	_, err = exec.ExecContext(ctx, q.SQL, q.Params...)
	logRepositorySQLTiming(ctx, "RecordRepository.DeleteByField", tableName, tableName, q.SQL, len(q.Params), 1, 0, sqlStart, err)
	return err
}

// CheckUniqueness checks if a value exists for a field, excluding a specific ID
func (r *RecordRepository) CheckUniqueness(ctx context.Context, schema *models.ObjectMetadata, fieldName string, value interface{}, excludeID string) (bool, error) {
	builder := query.From(schema.APIName).
		WithMetadata(schema).
		Select([]string{constants.FieldID}).
		Where(fieldName, value).
		Limit(1)

	if excludeID != "" {
		// STRICT ALIASING: Use correct logical field name. Builder now handles negative conditions specifically.
		builder.WhereNot(constants.FieldID, excludeID)
	}

	q := builder.Build()

	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.CheckUniqueness", schema.APIName, schema.APIName, q.SQL, len(q.Params), 1, 0, sqlStart, err)
		return false, err
	}
	defer rows.Close()

	exists := rows.Next()
	rowCount := 0
	if exists {
		rowCount = 1
	}
	logRepositorySQLTiming(ctx, "RecordRepository.CheckUniqueness", schema.APIName, schema.APIName, q.SQL, len(q.Params), 1, rowCount, sqlStart, nil)
	return exists, nil
}

// CheckQueueSupport checks if a queue supports a specific object type
func (r *RecordRepository) CheckQueueSupport(ctx context.Context, queueID string, objectType string) (bool, error) {
	exec := r.GetExecutor(ctx)
	var count int
	checkQuery := fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE %s = ? AND %s = ?",
		constants.TableQueueSobject,
		constants.FieldQueueSobject_QueueID,
		constants.FieldQueueSobject_SobjectType,
	)
	sqlStart := time.Now()
	err := exec.QueryRowContext(ctx, checkQuery, queueID, objectType).Scan(&count)
	logRepositorySQLTiming(ctx, "RecordRepository.CheckQueueSupport", constants.TableQueueSobject, constants.TableQueueSobject, checkQuery, 2, 1, count, sqlStart, err)
	if err != nil {
		return false, err
	}
	return count > 0, nil
}

// GetRecordType looks up the object type for a given record ID by enumerating
// live objects and probing each for the id. Schema drift (a live _System_Object
// row whose physical table is missing, or an Id-less system table) is treated as
// a non-match, so a genuinely absent id returns NotFound and only genuine infra
// failures fail closed.
//
// This is a best-effort reverse lookup: a 1146 here is ambiguous — it can be a
// transient stale-InfoSchema miss on a freshly-created object on multi-node
// TiDB, or a pre-existing orphaned metadata row (a live _System_Object row with
// no physical table). Each non-transactional 1146 gets one immediate probe on
// a genuinely fresh connection. A second 1146 is still treated as schema drift,
// so historical orphans remain a non-match instead of poisoning unrelated lookups.
func (r *RecordRepository) GetRecordType(ctx context.Context, id string) (string, error) {
	idTypeMap, err := r.FindObjectTypesForIDs(ctx, []string{id}, nil)
	if err != nil {
		return "", err
	}
	if objectType, ok := idTypeMap[id]; ok {
		return objectType, nil
	}
	return "", pkgerrors.NewNotFoundError("record", id)
}

// liveObject pairs a live object's logical ApiName with the physical table its
// reads route to. readTable equals apiName unless the object declares a
// ReadSource alias (e.g. RecordType → _System_RecordType), in which case NO
// physical table named apiName exists and probing it would raise a spurious
// 1146. Identity probes must target readTable, exactly as every other read path
// resolves it via ObjectMetadata.ResolveReadSourceTable().
type liveObject struct {
	apiName   string
	readTable string
}

type objectTypeCandidateProbeTarget struct {
	readTable string
	apiNames  []string
}

func objectTypeCandidateProbeTargets(objects []liveObject) []objectTypeCandidateProbeTarget {
	targets := make([]objectTypeCandidateProbeTarget, 0, len(objects))
	indexByReadTable := make(map[string]int, len(objects))
	for _, object := range objects {
		index, exists := indexByReadTable[object.readTable]
		if !exists {
			indexByReadTable[object.readTable] = len(targets)
			targets = append(targets, objectTypeCandidateProbeTarget{
				readTable: object.readTable,
				apiNames:  []string{object.apiName},
			})
			continue
		}
		targets[index].apiNames = append(targets[index].apiNames, object.apiName)
	}
	for i := range targets {
		sort.Strings(targets[i].apiNames)
	}
	return targets
}

func (r *RecordRepository) listLiveObjects(ctx context.Context) ([]liveObject, error) {
	exec := r.GetExecutor(ctx)

	// Query each record-shaped live object's ApiName plus its ReadSource so
	// identity probes hit the physical read table, not the logical name (which
	// need not be a table). _System_Field is the Catalog authority for whether a
	// logical object participates in generic record identity: a live-listed
	// infrastructure/operations table without a live Id field is not a generic
	// record object and must not enter the physical probe fan-out. Physical drift
	// remains guarded by the probe path below.
	metaQuery := fmt.Sprintf("SELECT o.%s, o.%s FROM %s o WHERE o.%s != 1 AND EXISTS (SELECT 1 FROM %s f WHERE f.%s = o.%s AND f.%s = '%s' AND COALESCE(f.%s, 0) = 0)",
		constants.FieldAPIName,
		constants.FieldSysObject_ReadSource,
		constants.TableObject,
		constants.FieldIsDeleted,
		constants.TableField,
		constants.FieldObjectID,
		constants.FieldID,
		constants.FieldAPIName,
		constants.FieldID,
		constants.FieldIsDeleted,
	)

	sqlStart := time.Now()
	rows, err := exec.QueryContext(ctx, metaQuery)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.listLiveObjects", constants.TableObject, constants.TableObject, metaQuery, 0, 1, 0, sqlStart, err)
		return nil, pkgerrors.NewInternalError("failed to query object metadata", err)
	}
	defer rows.Close()

	var objects []liveObject
	for rows.Next() {
		var name string
		var readSource sql.NullString
		if err := rows.Scan(&name, &readSource); err != nil {
			continue
		}
		objects = append(objects, liveObject{apiName: name, readTable: resolveLiveReadTable(name, readSource)})
	}
	if err := rows.Err(); err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.listLiveObjects", constants.TableObject, constants.TableObject, metaQuery, 0, 1, len(objects), sqlStart, err)
		return nil, pkgerrors.NewInternalError("failed to scan object metadata", err)
	}
	logRepositorySQLTiming(ctx, "RecordRepository.listLiveObjects", constants.TableObject, constants.TableObject, metaQuery, 0, 1, len(objects), sqlStart, nil)
	return objects, nil
}

// resolveLiveReadTable returns the physical table an object's reads route to.
// A ReadSource alias points to its declared physicalTable; everything else (no
// alias, or a malformed/unreadable alias we cannot trust) falls back to the
// ApiName, matching the pre-existing enumeration behavior.
func resolveLiveReadTable(apiName string, readSource sql.NullString) string {
	if !readSource.Valid || strings.TrimSpace(readSource.String) == "" {
		return apiName
	}
	cfg, err := unmarshalObjectReadSourceConfig(readSource.String)
	if err != nil || cfg == nil || strings.TrimSpace(cfg.PhysicalTable) == "" {
		return apiName
	}
	return cfg.PhysicalTable
}

type contextQueryer interface {
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
}

func (r *RecordRepository) findObjectIDs(ctx context.Context, tableName string, ids []string) ([]string, error) {
	foundIDs, err := r.findObjectIDsWithQueryer(ctx, r.GetExecutor(ctx), tableName, ids)
	if !IsTableNotExistsError(err) {
		return foundIDs, err
	}

	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		// The transaction owns its session. Never escape it after a 1146; the
		// outer resolver keeps the established transaction-local drift behavior.
		return foundIDs, err
	}

	if r.openIndependentObjectIDProbeDB == nil {
		return nil, errIndependentObjectIDProbeUnavailable
	}
	return r.reprobeObjectIDsOnIndependentDB(ctx, tableName, ids)
}

func (r *RecordRepository) reprobeObjectIDsOnIndependentDB(ctx context.Context, tableName string, ids []string) ([]string, error) {
	independentDB, err := r.openIndependentObjectIDProbeDB()
	if err != nil {
		return nil, fmt.Errorf("open independent object identity probe database: %w", err)
	}
	if independentDB == nil {
		return nil, errIndependentObjectIDProbeUnavailable
	}
	defer independentDB.Close()

	reprobeCtx, cancel := context.WithTimeout(ctx, independentObjectIDProbeBudget)
	defer cancel()
	return r.findObjectIDsWithQueryer(reprobeCtx, independentDB, tableName, ids)
}

func (r *RecordRepository) findObjectIDsWithQueryer(ctx context.Context, queryer contextQueryer, tableName string, ids []string) ([]string, error) {
	if !isValidFieldName(tableName) {
		return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
	}

	vals := make([]interface{}, len(ids))
	exactIDs := make(map[string]struct{}, len(ids))
	for i, id := range ids {
		vals[i] = id
		exactIDs[id] = struct{}{}
	}

	var sb strings.Builder
	sb.WriteString("SELECT `")
	sb.WriteString(constants.FieldID)
	sb.WriteString("` FROM `")
	sb.WriteString(tableName)
	sb.WriteString("` WHERE `")
	sb.WriteString(constants.FieldID)
	sb.WriteString("` IN (")
	for i := range ids {
		if i > 0 {
			sb.WriteString(",")
		}
		sb.WriteString("?")
	}
	sb.WriteString(")")

	sqlText := sb.String()
	sqlStart := time.Now()
	rows, err := queryer.QueryContext(ctx, sqlText, vals...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.findObjectIDs", tableName, tableName, sqlText, len(vals), 1, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	foundIDs := make([]string, 0)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		// TiDB/MySQL coerces a string predicate to the physical column type.
		// A Salesforce ID such as "001..." therefore matches a BIGINT Id=1
		// (notably _System_AuditOutbox) even though the opaque identities are
		// different. Preserve the indexed SQL probe, then reject every coerced
		// row by comparing the stored identity to the caller's exact input.
		if _, exact := exactIDs[id]; !exact {
			continue
		}
		foundIDs = append(foundIDs, id)
	}
	if err := rows.Err(); err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.findObjectIDs", tableName, tableName, sqlText, len(vals), 1, len(foundIDs), sqlStart, err)
		return nil, err
	}
	logRepositorySQLTiming(ctx, "RecordRepository.findObjectIDs", tableName, tableName, sqlText, len(vals), 1, len(foundIDs), sqlStart, nil)
	return foundIDs, nil
}

// FindObjectTypesForIDs finds object types for a batch of record IDs using the
// same live object-identity source and probe semantics as GetRecordType.
//
// The tableNames argument is ignored for compatibility with older call sites:
// object identity must come from live _System_Object enumeration, not a cached
// metadata snapshot that can fail open to an empty list.
//
// Each object is probed against its RESOLVED physical read table (ReadSource
// aware), so an aliased object (e.g. RecordType -> _System_RecordType) is probed
// on its real table rather than a logical name that is not a table. Schema drift
// (missing table 1146 after one independent-connection reprobe / unknown column
// 1054) is a non-match and is skipped; only genuine infra errors fail the batch
// closed.
func (r *RecordRepository) FindObjectTypesForIDs(ctx context.Context, ids []string, _ []string) (map[string]string, error) {
	result := make(map[string]string)
	if len(ids) == 0 {
		return result, nil
	}

	uniqueIDs := make([]string, 0, len(ids))
	seenIDs := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		if _, seen := seenIDs[id]; seen {
			continue
		}
		seenIDs[id] = struct{}{}
		uniqueIDs = append(uniqueIDs, id)
	}

	liveObjects, err := r.listLiveObjects(ctx)
	if err != nil {
		return nil, err
	}
	if len(liveObjects) == 0 {
		return result, nil
	}

	workers := r.objectIDProbeWorkers(ctx, len(liveObjects))
	if workers == 1 {
		return r.findObjectTypesForIDsSequential(ctx, uniqueIDs, liveObjects)
	}
	return r.findObjectTypesForIDsConcurrent(ctx, uniqueIDs, liveObjects, workers)
}

// FindObjectTypeCandidatesForIDs returns every logical object whose physical
// read table currently contains an opaque ID. It is a snapshot lookup for
// detecting existing cross-object collisions; it does not claim to serialize
// concurrent inserts into tables where that ID is absent.
func (r *RecordRepository) FindObjectTypeCandidatesForIDs(ctx context.Context, ids []string) (map[string][]string, error) {
	result := make(map[string][]string)
	uniqueIDs := make([]string, 0, len(ids))
	seenIDs := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, seen := seenIDs[id]; seen {
			continue
		}
		seenIDs[id] = struct{}{}
		uniqueIDs = append(uniqueIDs, id)
	}
	if len(uniqueIDs) == 0 {
		return result, nil
	}

	liveObjects, err := r.listLiveObjects(ctx)
	if err != nil {
		return nil, err
	}
	targets := objectTypeCandidateProbeTargets(liveObjects)
	if len(targets) == 0 {
		return result, nil
	}
	workers := r.objectIDProbeWorkers(ctx, len(targets))
	if workers == 1 {
		return r.findObjectTypeCandidatesForIDsSequential(ctx, uniqueIDs, targets)
	}
	return r.findObjectTypeCandidatesForIDsConcurrent(ctx, uniqueIDs, targets, workers)
}

func appendObjectTypeCandidates(result map[string][]string, target objectTypeCandidateProbeTarget, foundIDs []string) {
	for _, id := range foundIDs {
		result[id] = append(result[id], target.apiNames...)
	}
}

func sortObjectTypeCandidates(result map[string][]string) {
	for id := range result {
		sort.Strings(result[id])
	}
}

func objectTypeCandidateProbeError(err error) error {
	return pkgerrors.NewInternalError("failed to resolve object type candidates for ids: one or more probes errored", err)
}

type objectTypeCandidateChunkResult struct {
	index int
	found map[int][]string
	err   error
}

func (r *RecordRepository) findObjectTypeCandidateChunk(
	ctx context.Context,
	ids []string,
	targets []objectTypeCandidateProbeTarget,
	start int,
	end int,
) (map[int][]string, error) {
	if start < 0 || end > len(targets) || start >= end {
		return nil, pkgerrors.NewValidationError("targets", "invalid object identity candidate chunk")
	}
	args := make([]interface{}, 0, (end-start)*len(ids))
	var statement strings.Builder
	for index := start; index < end; index++ {
		tableName := targets[index].readTable
		if !isValidFieldName(tableName) {
			return nil, pkgerrors.NewValidationError("tableName", fmt.Sprintf("invalid table name: %s", tableName))
		}
		if index > start {
			statement.WriteString(" UNION ALL ")
		}
		statement.WriteString("SELECT `")
		statement.WriteString(constants.FieldID)
		statement.WriteString("`, ")
		statement.WriteString(strconv.Itoa(index))
		statement.WriteString(" AS `__target_index` FROM `")
		statement.WriteString(tableName)
		statement.WriteString("` WHERE `")
		statement.WriteString(constants.FieldID)
		statement.WriteString("` IN (")
		for idIndex, id := range ids {
			if idIndex > 0 {
				statement.WriteString(",")
			}
			statement.WriteString("?")
			args = append(args, id)
		}
		statement.WriteString(")")
	}

	sqlText := statement.String()
	sqlStart := time.Now()
	rows, err := r.GetExecutor(ctx).QueryContext(ctx, sqlText, args...)
	if err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.findObjectTypeCandidateChunk", "", "", sqlText, len(args), end-start, 0, sqlStart, err)
		return nil, err
	}
	defer rows.Close()

	found := make(map[int][]string)
	exactIDs := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		exactIDs[id] = struct{}{}
	}
	rowCount := 0
	for rows.Next() {
		var id string
		var targetIndex int
		if err := rows.Scan(&id, &targetIndex); err != nil {
			return nil, err
		}
		if targetIndex < start || targetIndex >= end {
			return nil, errors.New("object identity candidate query returned an invalid target index")
		}
		if _, exact := exactIDs[id]; !exact {
			continue
		}
		found[targetIndex] = append(found[targetIndex], id)
		rowCount++
	}
	if err := rows.Err(); err != nil {
		logRepositorySQLTiming(ctx, "RecordRepository.findObjectTypeCandidateChunk", "", "", sqlText, len(args), end-start, rowCount, sqlStart, err)
		return nil, err
	}
	logRepositorySQLTiming(ctx, "RecordRepository.findObjectTypeCandidateChunk", "", "", sqlText, len(args), end-start, rowCount, sqlStart, nil)
	return found, nil
}

func (r *RecordRepository) findObjectTypeCandidateChunkAfterDrift(
	ctx context.Context,
	ids []string,
	targets []objectTypeCandidateProbeTarget,
	start int,
	end int,
) (map[int][]string, error) {
	found := make(map[int][]string)
	for index := start; index < end; index++ {
		foundIDs, err := r.findObjectIDs(ctx, targets[index].readTable, ids)
		if err != nil {
			if IsSchemaDrift(err) {
				continue
			}
			return nil, err
		}
		found[index] = foundIDs
	}
	return found, nil
}

func (r *RecordRepository) findObjectTypeCandidatesForIDsSequential(ctx context.Context, ids []string, targets []objectTypeCandidateProbeTarget) (map[string][]string, error) {
	result := make(map[string][]string)
	var lastProbeErr error
	for _, target := range targets {
		foundIDs, err := r.findObjectIDs(ctx, target.readTable, ids)
		if err != nil {
			if !IsSchemaDrift(err) {
				lastProbeErr = err
			}
			continue
		}
		appendObjectTypeCandidates(result, target, foundIDs)
	}
	if lastProbeErr != nil {
		return nil, objectTypeCandidateProbeError(lastProbeErr)
	}
	sortObjectTypeCandidates(result)
	return result, nil
}

func (r *RecordRepository) findObjectTypeCandidatesForIDsConcurrent(ctx context.Context, ids []string, targets []objectTypeCandidateProbeTarget, workers int) (map[string][]string, error) {
	probeCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	chunkCount := (len(targets) + objectTypeCandidateProbeChunkSize - 1) / objectTypeCandidateProbeChunkSize
	jobs := make(chan int, chunkCount)
	results := make(chan objectTypeCandidateChunkResult, chunkCount)
	for chunk := 0; chunk < chunkCount; chunk++ {
		jobs <- chunk
	}
	close(jobs)
	if workers > chunkCount {
		workers = chunkCount
	}

	var group sync.WaitGroup
	group.Add(workers)
	for worker := 0; worker < workers; worker++ {
		go func() {
			defer group.Done()
			for chunk := range jobs {
				start := chunk * objectTypeCandidateProbeChunkSize
				end := start + objectTypeCandidateProbeChunkSize
				if end > len(targets) {
					end = len(targets)
				}
				found, err := r.findObjectTypeCandidateChunk(probeCtx, ids, targets, start, end)
				if err != nil && IsSchemaDrift(err) {
					found, err = r.findObjectTypeCandidateChunkAfterDrift(probeCtx, ids, targets, start, end)
				}
				select {
				case results <- objectTypeCandidateChunkResult{index: chunk, found: found, err: err}:
				case <-probeCtx.Done():
					return
				}
			}
		}()
	}

	result := make(map[string][]string)
	probeErrors := make([]error, chunkCount)
	for completed := 0; completed < chunkCount; completed++ {
		select {
		case <-ctx.Done():
			cancel()
			group.Wait()
			return nil, objectTypeCandidateProbeError(ctx.Err())
		case probe := <-results:
			if probe.err != nil {
				probeErrors[probe.index] = probe.err
				continue
			}
			for index, foundIDs := range probe.found {
				appendObjectTypeCandidates(result, targets[index], foundIDs)
			}
		}
	}
	group.Wait()
	for _, err := range probeErrors {
		if err != nil {
			return nil, objectTypeCandidateProbeError(err)
		}
	}
	sortObjectTypeCandidates(result)
	return result, nil
}

func (r *RecordRepository) objectIDProbeWorkers(ctx context.Context, objectCount int) int {
	if objectCount <= 1 || r == nil || r.db == nil {
		return 1
	}
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		// database/sql transactions own one connection and do not support
		// concurrent statements. Preserve the established transaction-local
		// scan exactly.
		return 1
	}
	workers := objectIDProbeConcurrency
	if workers > objectCount {
		workers = objectCount
	}
	maxOpen := r.db.Stats().MaxOpenConnections
	if maxOpen <= 1 {
		// Only an explicitly bounded multi-connection pool opts into fan-out.
		// This keeps mocks, embedded callers, and unconfigured sql.DB owners on
		// the established serial path instead of assuming spare capacity.
		return 1
	}
	if workers > maxOpen {
		workers = maxOpen
	}
	if workers < 1 {
		return 1
	}
	return workers
}

func (r *RecordRepository) findObjectTypesForIDsSequential(ctx context.Context, uniqueIDs []string, liveObjects []liveObject) (map[string]string, error) {
	result := make(map[string]string)

	// Probe-failure handling mirrors the single-ID GetRecordType contract:
	//
	//  1. Schema drift (missing table 1146 after one independent-connection
	//     reprobe, unknown column 1054) means this probe cannot return a match, so
	//     it is skipped. Drift is a real, reachable state here, not a bug: a live
	//     _System_Object row can lack its physical table (1146) and a few Id-less
	//     _System_* tables (e.g. _System_Config) stay live-listed (1054).
	//     DropTable is now atomic (#3484) so it no longer creates the
	//     tableless-object orphan, but historical orphans may persist until
	//     reconciled. A fresh object's table can also 1146 transiently on one
	//     multi-node TiDB session (stale InfoSchema), so 1146 is immediately
	//     reprobed once through a distinct session before being classified as
	//     drift. No delay or unbounded retry is involved.
	//
	//  2. Infra failure (deadlock, connection loss, timeout) means an unresolved
	//     ID might have lived in the failed table, so unresolved batches fail
	//     closed instead of degrading to "not found".
	var probeErred bool
	var lastProbeErr error
	for _, obj := range liveObjects {
		// No static prefix filter: _System_* objects are a mixed bag —
		// many (e.g. _System_Object, _System_RecordType) have a standard
		// Id column and MUST be probed so Composite Collections can resolve
		// them, while a few lack Id. Schema drift is absorbed below.
		foundIDs, err := r.findObjectIDs(ctx, obj.readTable, uniqueIDs)
		if err != nil {
			if IsSchemaDrift(err) {
				continue
			}
			probeErred = true
			lastProbeErr = err
			continue
		}
		// Report the LOGICAL object type even when reads route through an alias
		// physical table.
		for _, foundID := range foundIDs {
			result[foundID] = obj.apiName
		}
		if len(result) == len(uniqueIDs) {
			return result, nil
		}
	}

	if probeErred {
		return nil, pkgerrors.NewInternalError(
			"failed to resolve object types for ids: one or more probes errored",
			lastProbeErr,
		)
	}
	return result, nil
}

type objectIDProbeResult struct {
	index    int
	foundIDs []string
	err      error
}

// findObjectTypesForIDsConcurrent overlaps independent table probes while
// applying their results strictly in live-object enumeration order. That keeps
// the existing first-complete/overwrite semantics even if the same legacy ID
// exists in more than one table; concurrency changes latency, not authority.
func (r *RecordRepository) findObjectTypesForIDsConcurrent(ctx context.Context, uniqueIDs []string, liveObjects []liveObject, workers int) (map[string]string, error) {
	probeCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	jobs := make(chan int)
	results := make(chan objectIDProbeResult, workers)
	var group sync.WaitGroup
	group.Add(workers)
	for worker := 0; worker < workers; worker++ {
		go func() {
			defer group.Done()
			for index := range jobs {
				foundIDs, err := r.findObjectIDs(probeCtx, liveObjects[index].readTable, uniqueIDs)
				select {
				case results <- objectIDProbeResult{index: index, foundIDs: foundIDs, err: err}:
				case <-probeCtx.Done():
					return
				}
			}
		}()
	}

	for index := 0; index < workers; index++ {
		jobs <- index
	}
	nextProbe := workers
	nextApply := 0
	pending := make(map[int]objectIDProbeResult, workers)
	result := make(map[string]string)
	var probeErred bool
	var lastProbeErr error

	stop := func() {
		cancel()
		close(jobs)
		group.Wait()
	}
	for nextApply < len(liveObjects) {
		select {
		case <-ctx.Done():
			probeErr := ctx.Err()
			stop()
			return nil, pkgerrors.NewInternalError(
				"failed to resolve object types for ids: one or more probes errored",
				probeErr,
			)
		case probe := <-results:
			pending[probe.index] = probe
		}

		for {
			probe, ok := pending[nextApply]
			if !ok {
				break
			}
			delete(pending, nextApply)
			object := liveObjects[nextApply]
			nextApply++
			if probe.err != nil {
				if !IsSchemaDrift(probe.err) {
					probeErred = true
					lastProbeErr = probe.err
				}
			} else {
				for _, foundID := range probe.foundIDs {
					result[foundID] = object.apiName
				}
				if len(result) == len(uniqueIDs) {
					stop()
					return result, nil
				}
			}
			if nextProbe < len(liveObjects) {
				jobs <- nextProbe
				nextProbe++
			}
		}
	}
	stop()
	if probeErred {
		return nil, pkgerrors.NewInternalError(
			"failed to resolve object types for ids: one or more probes errored",
			lastProbeErr,
		)
	}
	return result, nil
}