package persistence

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"time"

	pkgerrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	domainrls "github.com/nexuscrm/kernel/pkg/rls"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/rs/zerolog/log"
)

// QueryRepository handles complex read operations (filtering, searching, analytics).
// Method implementations are split across domain-specific files:
//   - query_analytics.go — RunAnalytics, applyAnalyticsFilters, applyAnalyticsOperation, executeAndFormatAnalytics
//   - query_helpers.go   — shared helpers (resolvePhysicalName, newFieldMapper, schemaHasIsDeleted, isValidFieldName, etc.)
type QueryRepository struct {
	db           *sql.DB
	viewResolver ports.ViewResolver
}

// NewQueryRepository creates a new QueryRepository
func NewQueryRepository(db *sql.DB) *QueryRepository {
	return &QueryRepository{db: db}
}

// SetViewResolver injects the ViewResolver for formula view routing.
// When set, read queries will be routed to v_{object} views when available.
func (r *QueryRepository) SetViewResolver(vr ports.ViewResolver) {
	r.viewResolver = vr
}

// resolveReadTable returns the canonical read target for shared query readers.
func (r *QueryRepository) resolveReadTable(ctx context.Context, schema *models.ObjectMetadata) (string, error) {
	return resolveQueryReadTarget(ctx, schema, r.viewResolver)
}

// GetExecutor returns the DB connection or active transaction from context
func (r *QueryRepository) GetExecutor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

// Find executes a structured query request.
func (r *QueryRepository) Find(ctx context.Context, tableSchema *models.ObjectMetadata, req models.QueryRequest, rlsResult *domainrls.Result, visibleFields []string) (*models.QueryResult, error) {
	return r.find(ctx, tableSchema, req, rlsResult, visibleFields, nil)
}

// FindWithCorrelatedMemberships executes a structured row query with
// service-validated correlated membership predicates.
func (r *QueryRepository) FindWithCorrelatedMemberships(ctx context.Context, tableSchema *models.ObjectMetadata, req models.QueryRequest, rlsResult *domainrls.Result, visibleFields []string, correlatedMemberships []ports.QueryCorrelatedMembershipPlan) (*models.QueryResult, error) {
	return r.find(ctx, tableSchema, req, rlsResult, visibleFields, correlatedMemberships)
}

func (r *QueryRepository) find(ctx context.Context, tableSchema *models.ObjectMetadata, req models.QueryRequest, rlsResult *domainrls.Result, visibleFields []string, correlatedMemberships []ports.QueryCorrelatedMembershipPlan) (*models.QueryResult, error) {
	preparationStarted := perftiming.StartObservation(ctx)
	selectedFields := visibleFields
	if len(req.SelectFields) > 0 {
		visible := make(map[string]bool, len(visibleFields))
		for _, f := range visibleFields {
			visible[f] = true
		}
		selectedFields = make([]string, 0, len(req.SelectFields)+4)
		added := make(map[string]bool)
		addField := func(field string) {
			if field == "" || added[field] || !visible[field] {
				return
			}
			selectedFields = append(selectedFields, field)
			added[field] = true
		}
		addField(constants.FieldID)
		for _, f := range req.SelectFields {
			if strings.HasSuffix(f, "_Name") {
				addField(strings.TrimSuffix(f, "_Name"))
				continue
			}
			addField(f)
		}
	}

	readRLSResult, err := r.prepareRLSForRead(ctx, rlsResult, req.ScanAll)
	if err != nil {
		return nil, err
	}
	readTable, err := resolveQueryReadTargetForRequestWithMemberships(ctx, tableSchema, r.viewResolver, req, req.SelectFields, selectedFields, readRLSResult, correlatedMemberships)
	if err != nil {
		return nil, err
	}
	if err := validateParentFilterReadTarget(tableSchema, readTable, req.ParentFilterPaths); err != nil {
		return nil, err
	}
	readRLSResult, err = prepareRLSForReadTarget(readRLSResult, tableSchema, readTable)
	if err != nil {
		return nil, err
	}

	// 1. Build Base Builder (Common for Data and Count)
	applyBaseLogic := func(b *query.Builder) error {
		return r.applyFindBaseLogic(ctx, b, tableSchema, readTable, req, correlatedMemberships)
	}

	// 2. Execute Data Query (route to formula view if available)
	// Apply RLS with the resolved physical table name so that manual-share
	// and team-member predicates reference the actual FROM target, not the
	// logical object name (which may differ when routed to a formula view).
	applyRLS := func(b *query.Builder) {
		if rlsWhere := domainrls.ToSQLFilter(readRLSResult, readTable); rlsWhere != "" {
			rlsWhere = qualifyParentFilterRootRLSFilter(rlsWhere, tableSchema, readTable, req.ParentFilterPaths)
			b.WhereRaw(rlsWhere, nil)
		}
	}

	builder := queryBuilderForReadTarget(tableSchema, readTable)
	builder.Select(selectedFields)
	if err := applyBaseLogic(builder); err != nil {
		return nil, err
	}
	applyRLS(builder)

	// Apply sorting — MRU JOIN takes precedence over standard field sort
	if req.MRUUserID != "" {
		builder.JoinWithParams("INNER JOIN", constants.TableUserRecentItem, "__mru",
			fmt.Sprintf("`__mru`.`RecordId` = `%s`.`Id` AND `__mru`.`UserId` = ? AND `__mru`.`ObjectApiName` = ?", readTable),
			[]interface{}{req.MRUUserID, req.ObjectAPIName})
		builder.OrderByRaw("`__mru`.`AccessedAt` DESC", nil)
	} else if len(req.OrderBy) > 0 {
		orderBySQL, err := parentFilterStructuredOrderBySQL(tableSchema, readTable, req.ParentFilterPaths, req.OrderBy, "OrderBy")
		if err != nil {
			return nil, err
		}
		if orderBySQL != "" {
			builder.OrderByRaw(orderBySQL, nil)
		}
	} else if req.SortField != "" {
		orderBySQL, err := parentFilterStructuredOrderBySQL(tableSchema, readTable, req.ParentFilterPaths, []models.SortCriterion{{Field: req.SortField, Direction: req.SortDirection}}, "SortField")
		if err != nil {
			return nil, err
		}
		if orderBySQL != "" {
			builder.OrderByRaw(orderBySQL, nil)
		}
	}

	// Apply limit & offset
	limit := req.Limit
	if limit <= 0 {
		limit = constants.DefaultSOSLLimit
	}
	// Fetch one extra to detect "Has Next"
	builder.Limit(limit + 1)

	if req.Offset > 0 {
		builder.Offset(req.Offset)
	}

	q := builder.Build()
	perftiming.Observe(ctx, "query_repository.preparation", preparationStarted)
	log.Debug().Str("sql", q.SQL).Any("params", q.Params).Msg("DEBUG_FIND_SQL")
	exec := r.GetExecutor(ctx)
	sqlStart := time.Now()
	sqlObservationStarted := perftiming.StartObservation(ctx)
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	perftiming.Observe(ctx, "query_repository.sql_execution", sqlObservationStarted)
	if err != nil {
		logRepositorySQLTiming(ctx, "QueryRepository.Find.data", tableSchema.APIName, readTable, q.SQL, len(q.Params), len(selectedFields), 0, sqlStart, err)
		if IsUnknownColumnError(err) {
			colName := extractUnknownColumnName(err)
			return nil, pkgerrors.NewColumnNotFoundError(colName)
		}
		return nil, pkgerrors.NewInternalError("query execution error", err)
	}
	defer rows.Close()

	scanStarted := perftiming.StartObservation(ctx)
	records, err := query.ScanRowsToSObjects(rows)
	perftiming.Observe(ctx, "query_repository.row_scan", scanStarted)
	logRepositorySQLTiming(ctx, "QueryRepository.Find.data", tableSchema.APIName, readTable, q.SQL, len(q.Params), len(selectedFields), len(records), sqlStart, err)
	if err != nil {
		return nil, err
	}

	// 3. Determine Pagination Status
	done := true
	if len(records) > limit {
		done = false
		records = records[:limit] // Trim extra record
	}

	// 4. Calculate Total Size
	// Optimization: if done=true and offset=0, total = len(records)
	totalSize := len(records)
	if req.SkipTotalSize {
		return &models.QueryResult{
			TotalSize:        totalSize,
			TotalSizeUnknown: !done,
			Done:             done,
			Records:          records,
		}, nil
	}
	if (!done || req.Offset > 0) && !(done && req.Offset == 0) {
		// Need count query
		countBuilder := queryBuilderForReadTarget(tableSchema, readTable).SkipID()
		countBuilder.AddSelectRaw(countSelectExpression(tableSchema, readTable))
		if err := applyBaseLogic(countBuilder); err != nil {
			return nil, err
		}
		applyRLS(countBuilder)
		qCount := countBuilder.Build()

		var count int
		sqlStart := time.Now()
		err := exec.QueryRowContext(ctx, qCount.SQL, qCount.Params...).Scan(&count)
		logRepositorySQLTiming(ctx, "QueryRepository.Find.count", tableSchema.APIName, readTable, qCount.SQL, len(qCount.Params), 1, count, sqlStart, err)
		if err != nil {
			return nil, fmt.Errorf("count query error: %w", err)
		}
		totalSize = count
	}

	return &models.QueryResult{
		TotalSize: totalSize,
		Done:      done,
		Records:   records,
	}, nil
}

// Search performs a text search on specific fields
func (r *QueryRepository) Search(ctx context.Context, schema *models.ObjectMetadata, term string, searchFields []string, selectFields []string, limit int, offset int, orderBy string, orderDir string, filterExpr string, rlsResult *domainrls.Result, opts ports.SearchOptions) ([]models.SObject, error) {
	funcStart := time.Now()
	defer func() {
		log.Debug().
			Dur("repo_latency_ms", time.Since(funcStart)).
			Str("table", schema.APIName).
			Int("term_length", len(term)).
			Int("limit", limit).
			Msg("[SEARCH_METRIC] QueryRepository.Search latency")
	}()

	// ForceBaseTable reads the object's base table directly instead of the
	// formula view, so a substring/prefix LIKE scan can use a TiFlash columnar
	// replica (and a future name-column index) and skip the fat view's display
	// joins. Both the /search/ suggestions prefix path and the
	// /parameterizedSearch + /search substring path request it; the default
	// (non-search) read path keeps the formula-view-resolved target unchanged.
	//
	// The base-table fast path is only valid when EVERY column an emitted
	// statement references — predicate AND projection — is a real physical
	// base-table column. View-only (active-computed formula) columns exist only
	// on the formula view, so reading them off the base table is illegal SQL or
	// silently wrong. Sources that can reference a view-only column:
	//   1. The search columns themselves (an active-computed search field has no
	//      physical column — a base-table LIKE over it is illegal SQL).
	//   2. The projected return columns (substring search SELECTs the caller's
	//      requested fields; a view-only one cannot be read from the base table).
	//      The prefix/suggestions path returns only Id + Name so never needs this.
	//   3. Query-level RLS predicates or a filter expression can reference an
	//      active-computed/view-only field or virtual compound field.
	// RLS source expressions and direct fields are mapped for the selected read
	// target, so physical aliases keep the base path. Opaque pre-compiled RLS or
	// filter fragments demote when the schema contains a field whose API name is
	// unsafe on the base table. ControlledByParent / owner sharing keeps the fast
	// path when its fields have physical storage. Metadata-driven, no SObject-name
	// branch.
	readTable, err := schema.ResolveReadSourceTable()
	if err != nil {
		return nil, pkgerrors.NewValidationError("readSource", err.Error())
	}
	forceBaseTable := opts.ForceBaseTable
	if forceBaseTable {
		if !searchColumnsAllPhysical(schema, searchFields) {
			forceBaseTable = false
		} else if !opts.Prefix && !returnColumnsAllPhysical(schema, selectFields) {
			// Substring path projects caller-requested return fields; a view-only
			// one is unreadable from the base table. (Prefix returns only Id+Name.)
			forceBaseTable = false
		} else if !opts.Prefix && !searchTermFieldPhysical(schema, term) {
			// Substring path runs the term through ParseFormulaQuery: a typed
			// `field op value` term emits `base`.`field` with the RAW api name —
			// no logical->physical mapping. A view-only field (no base column) or
			// a PhysicalName-aliased field would be an unknown column on the base
			// table, turning a previously view-backed search into a hard error.
			// Demote so the parsed predicate resolves where the column exists.
			// (Prefix never parses formula terms — see the opts.Prefix branch.)
			forceBaseTable = false
		} else if !searchSortFieldPhysical(schema, orderBy) {
			// A caller-supplied ORDER BY on a view-only field resolves to a column
			// absent from the base table (a valid identifier, so the builder emits
			// it rather than dropping it) → unknown column on the base table.
			// Demote to the view, where the sort column exists.
			forceBaseTable = false
		} else if filterExpr != "" && schemaHasOpaquePredicateUnsafeField(schema) {
			forceBaseTable = false
		} else if !rlsCurrentTableCanUseBaseTable(schema, readTable, rlsResult) {
			forceBaseTable = false
		}
	}
	if !forceBaseTable {
		readTable, err = r.resolveReadTable(ctx, schema)
		if err != nil {
			return nil, err
		}
	}
	readRLSResult, err := r.prepareRLSForRead(ctx, rlsResult, false)
	if err != nil {
		return nil, err
	}
	readRLSResult, err = prepareRLSForReadTarget(readRLSResult, schema, readTable)
	if err != nil {
		return nil, err
	}
	hasNonSearchWhereSource := false
	var sqlWhere string
	var filterArgs []interface{}
	if filterExpr != "" {
		sqlWhere, filterArgs, err = formula.ToSQLWithMapper(filterExpr, newFieldMapperForTable(schema, readTable))
		if err != nil {
			return nil, pkgerrors.NewValidationError("filterExpr", fmt.Sprintf("invalid filter expression: %v", err))
		}
		if sqlWhere != "" {
			hasNonSearchWhereSource = true
		}
	}
	rlsWhere := domainrls.ToSQLFilter(readRLSResult, readTable)
	if rlsWhere != "" {
		hasNonSearchWhereSource = true
	}

	projectedSelectColumns := projectedSelectColumnKeys(schema, selectFields, readTable)

	activeSearchFields := append([]string(nil), searchFields...)
	for {
		builder := query.From(readTable)
		if !searchReadTargetExposesLogicalColumns(schema, readTable) {
			builder.WithMetadata(schema)
		}
		builder.Select(selectFields).ExcludeDeleted()

		// Apply Sorting
		if orderBy != "" {
			builder.OrderBy(orderBy, orderDir)
		} else if !opts.Prefix {
			// Stable primary-key ordering for the full (non-prefix) search path.
			// With no ORDER BY, TiDB satisfies `LOWER(col) LIKE '%term%' LIMIT n`
			// via an early-stop index scan on the row store (TiKV): the non-sargable
			// LIKE is evaluated row-by-row until n matches accumulate, which is slow
			// on large objects. Ordering by the clustered primary key forces a global
			// TopN, so the optimizer costs the full scan and routes it to a TiFlash
			// columnar replica when one exists (and falls back to TiKV otherwise —
			// never worse than today). Id is a generic infrastructure column on every
			// object, so this stays engine-generic with no per-object branch, and it
			// also makes result/pagination order deterministic. The prefix/suggestion
			// path is intentionally excluded: it relies on an index-eligible anchored
			// `col LIKE 'term%'` and must stay on its fast index path.
			builder.OrderBy(constants.FieldID, "DESC")
		}

		// Apply Offset
		if offset > 0 {
			builder.Offset(offset)
		}

		// Apply Filter Expression
		if sqlWhere != "" {
			builder.WhereRaw(sqlWhere, filterArgs)
		}
		if rlsWhere != "" {
			builder.WhereRaw(rlsWhere, nil)
		}

		var generatedSearchColumns map[string]struct{}
		var generatedSearchColumnFields map[string][]string

		// Support for listing all records with "*" wildcard
		if term != "*" && term != "" {
			if opts.Prefix {
				// Prefix (autocomplete) mode: always a literal anchored prefix match.
				// Never run the formula-query parser here — a typed term containing
				// `=`/`<`/`>`/`!=`/`<>` (e.g. "ab=cd") is a literal suggestion prefix,
				// not an equality filter. The LIKE pattern escapes user wildcards and
				// carries an explicit ESCAPE '\' so escaping holds under
				// NO_BACKSLASH_ESCAPES sql_mode.
				//
				// Anchored prefix match against the NAME target(s): an ordinary physical
				// name column emits a bare, index-eligible `col LIKE 'term%'` (columns
				// collate utf8mb4_unicode_ci, so no LOWER() wrapper is needed and the
				// index is preserved); a virtual compound name field emits ONE anchored
				// LIKE over its composed-name CONCAT_WS expression so the prefix matches
				// the composed display name, never an arbitrary member (that case is not
				// index-eligible, which is accepted — see prefixSearchExpressions).
				exprs, err := prefixSearchExpressions(schema, activeSearchFields, readTable)
				if err != nil {
					return nil, pkgerrors.NewValidationError("searchField", "invalid search field name")
				}
				searchConditions := make([]string, 0, len(exprs))
				searchParams := make([]interface{}, 0, len(exprs))
				for _, expr := range exprs {
					searchConditions = append(searchConditions, fmt.Sprintf("%s LIKE ? ESCAPE '\\\\'", expr))
					searchParams = append(searchParams, prefixLikePattern(term))
				}
				builder.WhereRaw(fmt.Sprintf("(%s)", strings.Join(searchConditions, " OR ")), searchParams)
			} else if formulaCondition, formulaParams, ok := query.ParseFormulaQuery(term, readTable); ok {
				// Try to parse as formula expression (field op value)
				builder.WhereRaw(formulaCondition, formulaParams)
			} else {
				// Fallback to text search
				searchConditions := make([]string, 0)
				searchParams := make([]interface{}, 0)
				generatedSearchColumns = make(map[string]struct{})
				generatedSearchColumnFields = make(map[string][]string)
				// hasSubstringLike tracks whether this iteration emits at least one
				// non-sargable `LOWER(col) LIKE '%term%'` predicate (vs a sargable
				// `Id = ?` / `Id LIKE 'term%'` ID condition). The enforce_mpp hint is
				// applied only when true — see the WithOptimizerHint call below.
				hasSubstringLike := false
				blockedGeneratedSearchFields := generatedSearchFieldsBlockedByCompoundTargets(schema, activeSearchFields)
				for _, field := range activeSearchFields {
					fieldMeta := models.FindField(schema, field)
					if condition, param, ok := idSearchCondition(schema, fieldMeta, term, readTable); ok {
						searchConditions = append(searchConditions, condition)
						searchParams = append(searchParams, param)
						continue
					}
					if fieldMeta != nil && strings.EqualFold(fieldMeta.APIName, constants.FieldID) {
						continue
					}
					searchExpr, err := searchFieldReadExpression(schema, field, readTable)
					if err != nil {
						return nil, pkgerrors.NewValidationError("searchField", fmt.Sprintf("invalid search field name: %s", field))
					}
					fieldMeta = models.FindField(schema, field)
					_, blockedByCompoundTarget := blockedGeneratedSearchFields[strings.ToLower(fieldMeta.APIName)]
					if !blockedByCompoundTarget && searchFieldUsesRetryableGeneratedRead(schema, fieldMeta) {
						for _, col := range searchExpr.Columns {
							key := normalizeSQLIdentifier(col)
							generatedSearchColumns[key] = struct{}{}
							generatedSearchColumnFields[key] = append(generatedSearchColumnFields[key], field)
						}
					}
					searchConditions = append(searchConditions, fmt.Sprintf("LOWER(%s) LIKE ?", searchExpr.SQL))
					searchParams = append(searchParams, fmt.Sprintf("%%%s%%", strings.ToLower(term)))
					hasSubstringLike = true
				}
				if len(searchConditions) == 0 {
					return []models.SObject{}, nil
				}
				builder.WhereRaw(fmt.Sprintf("(%s)", strings.Join(searchConditions, " OR ")), searchParams)
				// Scope the TiFlash MPP hint to exactly this substring-search branch.
				// Only here do we emit a non-sargable `LOWER(col) LIKE '%term%'` over
				// possibly many columns. On a small replica-backed object the optimizer
				// misestimates that LIKE as low-selectivity and picks a TiKV clustered-PK
				// reverse keep-order scan, betting it can early-stop at LIMIT n. When the
				// term matches few rows that bet fails: it scans every row and evaluates
				// the view's formula columns per row — 20s on a 3.6k-row object measured
				// on staging. enforce_mpp removes the TiKV keep-order option so a
				// replica-backed object takes the TiFlash MPP plan (20s -> ~130ms);
				// objects without a replica silently fall back to TiKV (verified: no
				// error, no warning), never worse than today. The hint is deliberately
				// NOT applied to the `*` list-all, exact-ID, or parsed-formula shapes:
				// those are sargable (point-get / keep-order top-N / index range) and
				// must keep their existing row-store plans. SET_VAR is statement-scoped
				// and does not leak to the session.
				if hasSubstringLike {
					builder.WithOptimizerHint("/*+ SET_VAR(tidb_enforce_mpp=ON) */")
				}
			}
		}

		if limit <= 0 {
			limit = constants.DefaultSOSLLimit
		}
		builder.Limit(limit)

		q := builder.Build()
		log.Debug().Str("sql", q.SQL).Any("params", q.Params).Msg("DEBUG_SEARCH_SQL")

		sqlStart := time.Now()
		exec := r.GetExecutor(ctx)
		rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
		if err != nil {
			unknownColumn := extractUnknownColumnName(err)
			if !hasNonSearchWhereSource &&
				IsUnknownColumnError(err) &&
				unknownColumnMatchesGeneratedSearchColumn(unknownColumn, generatedSearchColumns) {
				if unknownColumnMatchesProjectedSelectColumn(unknownColumn, projectedSelectColumns) {
					log.Warn().
						Err(err).
						Str("object", schema.APIName).
						Str("unknown_column", unknownColumn).
						Msg("Generated search field is also projected and cannot be read")
					return []models.SObject{}, nil
				}
				if !isUnknownColumnInWhereClause(err) {
					return nil, pkgerrors.NewInternalError("search execution error", err)
				}
				if nextSearchFields, ok := searchFieldsWithoutGeneratedColumn(activeSearchFields, unknownColumn, generatedSearchColumnFields); ok {
					if len(nextSearchFields) > 0 {
						log.Warn().
							Err(err).
							Str("object", schema.APIName).
							Str("unknown_column", unknownColumn).
							Msg("Retrying search without unreadable generated search field")
						activeSearchFields = nextSearchFields
						continue
					}
					log.Warn().
						Err(err).
						Str("object", schema.APIName).
						Str("unknown_column", unknownColumn).
						Msg("Generated search field cannot be read and no fallback search fields remain")
					return []models.SObject{}, nil
				}
				return nil, pkgerrors.NewValidationError("searchField", "generated search field cannot be read from the selected search source")
			}
			return nil, pkgerrors.NewInternalError("search execution error", err)
		}
		defer rows.Close()
		log.Debug().
			Dur("sql_exec_ms", time.Since(sqlStart)).
			Str("table", schema.APIName).
			Int("term_length", len(term)).
			Msg("[SEARCH_METRIC] SQL execution time")

		results, err := query.ScanRowsToSObjects(rows)
		if err != nil {
			return nil, err
		}
		log.Debug().Int("count", len(results)).Str("table", schema.APIName).Msg("DEBUG_SEARCH_RESULTS_COUNT")
		return results, nil
	}
}

// ExecuteRawSQL executes a raw SQL string (Validated by Service Layer)
func (r *QueryRepository) ExecuteRawSQL(ctx context.Context, sql string, params []interface{}) ([]models.SObject, error) {
	exec := r.GetExecutor(ctx)
	rows, err := exec.QueryContext(ctx, sql, params...)
	if err != nil {
		return nil, pkgerrors.NewInternalError("raw query error", err)
	}
	defer rows.Close()

	return query.ScanRowsToSObjects(rows)
}

// GetLookupNames retrieves specific ID and Name fields for a list of IDs
// idField: the field to use for ID matching (determined by caller from metadata)
// nameField: the field to use for display name
func (r *QueryRepository) GetLookupNames(ctx context.Context, tableSchema *models.ObjectMetadata, ids []string, idField string, nameField string) ([]models.SObject, error) {
	if len(ids) == 0 {
		return []models.SObject{}, nil
	}

	// Use Builder for schema-aware query construction
	if !isValidFieldName(idField) {
		return nil, pkgerrors.NewValidationError("idField", fmt.Sprintf("invalid id field name: %s", idField))
	}
	if !isValidFieldName(nameField) {
		return nil, pkgerrors.NewValidationError("nameField", fmt.Sprintf("invalid name field name: %s", nameField))
	}

	readTable, err := r.resolveReadTable(ctx, tableSchema)
	if err != nil {
		return nil, err
	}
	builder := query.From(readTable).WithMetadata(tableSchema)
	builder.Select([]string{idField, nameField})

	// Convert string IDs to interface{} for WhereIn
	vals := make([]interface{}, len(ids))
	for i, v := range ids {
		vals[i] = v
	}
	builder.WhereIn(idField, vals)
	if tableSchema != nil && schemaHasIsDeleted(tableSchema) {
		builder.ExcludeDeleted()
	}

	q := builder.Build()

	exec := r.GetExecutor(ctx)
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		return nil, pkgerrors.NewInternalError("lookup hydration error", err)
	}
	defer rows.Close()

	return query.ScanRowsToSObjects(rows)
}

// GetRecordsByIDs retrieves selected fields for a list of IDs.
func (r *QueryRepository) GetRecordsByIDs(ctx context.Context, tableSchema *models.ObjectMetadata, ids []string, fields []string) ([]models.SObject, error) {
	if len(ids) == 0 {
		return []models.SObject{}, nil
	}

	if len(fields) == 0 {
		fields = []string{constants.FieldID}
	}

	hasID := false
	for _, field := range fields {
		if !isValidFieldName(field) {
			return nil, fmt.Errorf("invalid field name: %s", field)
		}
		if strings.EqualFold(field, constants.FieldID) {
			hasID = true
		}
	}
	if !hasID {
		fields = append([]string{constants.FieldID}, fields...)
	}

	readTable, err := r.resolveReadTable(ctx, tableSchema)
	if err != nil {
		return nil, err
	}
	builder := query.From(readTable).WithMetadata(tableSchema)
	builder.Select(fields)
	if tableSchema != nil && schemaHasIsDeleted(tableSchema) {
		builder.ExcludeDeleted()
	}

	vals := make([]interface{}, len(ids))
	for i, v := range ids {
		vals[i] = v
	}
	builder.WhereIn(constants.FieldID, vals)

	q := builder.Build()
	exec := r.GetExecutor(ctx)
	rows, err := exec.QueryContext(ctx, q.SQL, q.Params...)
	if err != nil {
		return nil, fmt.Errorf("get records by ids error: %w", err)
	}
	defer rows.Close()

	return query.ScanRowsToSObjects(rows)
}