package persistence

import (
	"fmt"
	"strings"

	"github.com/nexuscrm/shared/pkg/formulastorage"

	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// validFilterOps is the set of supported SQL comparison operators for criteria filters.
var validFilterOps = map[string]bool{
	"=": true, "!=": true, "<": true, ">": true, "<=": true, ">=": true,
	"LIKE": true, "NOT LIKE": true,
	"IN": true, "NOT IN": true,
}

// resolvePhysicalName maps a logical API name to its physical column name using the schema.
// Returns the original name if no physical mapping exists or schema is nil.
func resolvePhysicalName(schema *models.ObjectMetadata, apiName string) string {
	if schema == nil {
		return apiName
	}
	for _, f := range schema.Fields {
		if strings.EqualFold(f.APIName, apiName) {
			return actualColumnName(f.APIName, f.PhysicalName)
		}
	}
	return apiName
}

func actualColumnName(apiName, physicalName string) string {
	if physicalName == "" {
		return apiName
	}

	trimmed := strings.TrimSpace(physicalName)
	// Bootstrap metadata sometimes stores Salesforce-facing aliases as backticked values
	// while the actual DB column remains the logical API name.
	if strings.HasPrefix(trimmed, "`") && strings.HasSuffix(trimmed, "`") {
		return apiName
	}

	return trimmed
}

func newFieldMapperForTable(schema *models.ObjectMetadata, tableAlias string) func(string) string {
	return func(identifier string) string {
		if expr := virtualCompoundSQLExpression(schema, identifier, tableAlias); expr != "" {
			return expr
		}
		return resolvePhysicalName(schema, identifier)
	}
}

func readTargetFieldMapperForTable(schema *models.ObjectMetadata, tableAlias string) func(string) string {
	if !searchReadTargetExposesLogicalColumns(schema, tableAlias) {
		return newFieldMapperForTable(schema, tableAlias)
	}
	return func(identifier string) string {
		trimmed := strings.Trim(strings.TrimSpace(identifier), "`")
		if field := models.FindField(schema, trimmed); field != nil {
			return field.APIName
		}
		return trimmed
	}
}

func readTargetColumnName(schema *models.ObjectMetadata, tableAlias string, field string) string {
	if searchReadTargetExposesLogicalColumns(schema, tableAlias) {
		return projectedSelectColumnName(schema, field, true)
	}
	return resolvePhysicalName(schema, field)
}

func queryBuilderForReadTarget(schema *models.ObjectMetadata, tableName string) *query.Builder {
	builder := query.From(tableName)
	if !searchReadTargetExposesLogicalColumns(schema, tableName) {
		builder.WithMetadata(schema)
	}
	return builder
}

func virtualCompoundSQLExpression(schema *models.ObjectMetadata, apiName string, tableAlias string) string {
	field := models.FindField(schema, apiName)
	if field == nil || !field.IsVirtual || field.VirtualKind != constants.VirtualFieldCompound || len(field.ComposedFrom) == 0 {
		return ""
	}

	parts := make([]string, 0, len(field.ComposedFrom))
	for _, member := range field.ComposedFrom {
		if models.FindField(schema, member) == nil {
			continue
		}
		col := qualifiedPhysicalColumn(schema, member, tableAlias)
		parts = append(parts, fmt.Sprintf("NULLIF(TRIM(%s), '')", col))
	}
	if len(parts) == 0 {
		return ""
	}

	return fmt.Sprintf("NULLIF(TRIM(CONCAT_WS(' ', %s)), '')", strings.Join(parts, ", "))
}

func qualifiedPhysicalColumn(schema *models.ObjectMetadata, apiName string, tableAlias string) string {
	colName := resolvePhysicalName(schema, apiName)
	if tableAlias == "" {
		return fmt.Sprintf("`%s`", colName)
	}
	return fmt.Sprintf("`%s`.`%s`", tableAlias, colName)
}

type searchFieldExpression struct {
	SQL     string
	Columns []string
}

func searchFieldSQLExpression(schema *models.ObjectMetadata, apiName string, tableAlias string) (string, error) {
	expr, err := searchFieldReadExpression(schema, apiName, tableAlias)
	if err != nil {
		return "", err
	}
	return expr.SQL, nil
}

func searchFieldReadExpression(schema *models.ObjectMetadata, apiName string, tableAlias string) (searchFieldExpression, error) {
	return searchFieldReadExpressionForAPIName(schema, apiName, tableAlias, make(map[string]struct{}))
}

func searchFieldReadExpressionForAPIName(schema *models.ObjectMetadata, apiName string, tableAlias string, seen map[string]struct{}) (searchFieldExpression, error) {
	if !isValidFieldName(apiName) {
		return searchFieldExpression{}, fmt.Errorf("invalid field name")
	}
	field := models.FindField(schema, apiName)
	if field == nil {
		return searchFieldExpression{}, fmt.Errorf("field %s is not declared in metadata", apiName)
	}
	return searchFieldReadExpressionForField(schema, field, tableAlias, seen)
}

func searchFieldReadExpressionForField(schema *models.ObjectMetadata, field *models.FieldMetadata, tableAlias string, seen map[string]struct{}) (searchFieldExpression, error) {
	if field == nil {
		return searchFieldExpression{}, fmt.Errorf("field metadata is required")
	}
	key := strings.ToLower(field.APIName)
	if _, ok := seen[key]; ok {
		return searchFieldExpression{}, fmt.Errorf("compound field %s recursively references itself", field.APIName)
	}
	if field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound {
		return compoundSearchFieldReadExpression(schema, field, tableAlias, seen)
	}
	if field.IsVirtual {
		return searchFieldExpression{}, fmt.Errorf("field %s is virtual and cannot be searched directly", field.APIName)
	}
	if !formulastorage.FieldRequiresPhysicalColumn(field) && !searchReadTargetExposesLogicalColumns(schema, tableAlias) {
		return searchFieldExpression{}, fmt.Errorf("field %s is not available on the base read target", field.APIName)
	}
	return simpleSearchFieldReadExpression(schema, field, tableAlias), nil
}

func compoundSearchFieldReadExpression(schema *models.ObjectMetadata, field *models.FieldMetadata, tableAlias string, seen map[string]struct{}) (searchFieldExpression, error) {
	if len(field.ComposedFrom) == 0 {
		return searchFieldExpression{}, fmt.Errorf("compound field %s has no members", field.APIName)
	}
	key := strings.ToLower(field.APIName)
	seen[key] = struct{}{}
	defer delete(seen, key)

	parts := make([]string, 0, len(field.ComposedFrom))
	columns := make([]string, 0, len(field.ComposedFrom))
	for _, member := range field.ComposedFrom {
		memberExpr, err := searchFieldReadExpressionForAPIName(schema, member, tableAlias, seen)
		if err != nil {
			return searchFieldExpression{}, fmt.Errorf("compound field %s member %s is not searchable: %w", field.APIName, member, err)
		}
		parts = append(parts, fmt.Sprintf("NULLIF(TRIM(%s), '')", memberExpr.SQL))
		columns = append(columns, memberExpr.Columns...)
	}
	if len(parts) == 0 {
		return searchFieldExpression{}, fmt.Errorf("compound field %s has no searchable members", field.APIName)
	}

	return searchFieldExpression{
		SQL:     fmt.Sprintf("NULLIF(TRIM(CONCAT_WS(' ', %s)), '')", strings.Join(parts, ", ")),
		Columns: columns,
	}, nil
}

func simpleSearchFieldReadExpression(schema *models.ObjectMetadata, field *models.FieldMetadata, tableAlias string) searchFieldExpression {
	colName := field.APIName
	if !searchReadTargetExposesLogicalColumns(schema, tableAlias) {
		colName = actualColumnName(field.APIName, field.PhysicalName)
	}
	column := fmt.Sprintf("`%s`", colName)
	if tableAlias != "" {
		column = fmt.Sprintf("`%s`.`%s`", tableAlias, colName)
	}
	return searchFieldExpression{
		SQL:     column,
		Columns: searchFieldColumnKeys(tableAlias, colName),
	}
}

func searchReadTargetExposesLogicalColumns(schema *models.ObjectMetadata, tableAlias string) bool {
	if schema == nil || strings.TrimSpace(tableAlias) == "" {
		return false
	}
	baseReadTable, err := schema.ResolveReadSourceTable()
	if err != nil {
		return false
	}
	return !strings.EqualFold(tableAlias, baseReadTable)
}

func searchFieldColumnKeys(tableAlias string, colName string) []string {
	if strings.TrimSpace(colName) == "" {
		return nil
	}
	if strings.TrimSpace(tableAlias) != "" {
		return []string{normalizeSQLIdentifier(tableAlias + "." + colName)}
	}
	return []string{normalizeSQLIdentifier(colName)}
}

func projectedSelectColumnKeys(schema *models.ObjectMetadata, selectFields []string, tableAlias string) map[string]struct{} {
	if len(selectFields) == 1 && selectFields[0] == "*" {
		return nil
	}

	logicalReadTarget := searchReadTargetExposesLogicalColumns(schema, tableAlias)
	columns := make(map[string]struct{})
	emitted := make([]string, 0, len(selectFields)+1)
	for _, field := range selectFields {
		resolvedField := projectedSelectColumnName(schema, field, logicalReadTarget)
		col := resolvedField
		if !strings.Contains(resolvedField, ".") && !strings.Contains(resolvedField, "`") && resolvedField != "*" {
			if strings.TrimSpace(tableAlias) != "" {
				col = tableAlias + "." + resolvedField
			}
		}
		emitted = append(emitted, col)
		for _, key := range projectedColumnKeys(col) {
			columns[key] = struct{}{}
		}
	}

	idCol := projectedSelectColumnName(schema, constants.FieldID, logicalReadTarget)
	hasID := false
	for _, col := range emitted {
		if strings.Contains(col, idCol) || col == "*" {
			hasID = true
			break
		}
	}
	if !hasID {
		for _, key := range searchFieldColumnKeys(tableAlias, idCol) {
			columns[key] = struct{}{}
		}
	}

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

func projectedSelectColumnName(schema *models.ObjectMetadata, field string, logicalReadTarget bool) string {
	trimmed := strings.TrimSpace(field)
	if logicalReadTarget {
		if !strings.Contains(trimmed, ".") && !strings.Contains(trimmed, "`") {
			if meta := models.FindField(schema, trimmed); meta != nil {
				return meta.APIName
			}
		}
		return trimmed
	}
	return resolvePhysicalName(schema, trimmed)
}

func projectedColumnKeys(col string) []string {
	trimmed := strings.TrimSpace(col)
	if trimmed == "" || trimmed == "*" {
		return nil
	}
	return []string{normalizeSQLIdentifier(trimmed)}
}

func normalizeSQLIdentifier(identifier string) string {
	trimmed := strings.TrimSpace(identifier)
	trimmed = strings.Trim(trimmed, "`'\"")
	trimmed = strings.ReplaceAll(trimmed, "`", "")
	return strings.ToLower(trimmed)
}

func unknownColumnMatchesGeneratedSearchColumn(unknownColumn string, generatedColumns map[string]struct{}) bool {
	if len(generatedColumns) == 0 || strings.TrimSpace(unknownColumn) == "" {
		return false
	}
	_, ok := generatedColumns[normalizeSQLIdentifier(unknownColumn)]
	return ok
}

func unknownColumnMatchesProjectedSelectColumn(unknownColumn string, projectedColumns map[string]struct{}) bool {
	if len(projectedColumns) == 0 || strings.TrimSpace(unknownColumn) == "" {
		return false
	}
	_, ok := projectedColumns[normalizeSQLIdentifier(unknownColumn)]
	return ok
}

func searchFieldsWithoutGeneratedColumn(searchFields []string, unknownColumn string, generatedColumnFields map[string][]string) ([]string, bool) {
	if len(searchFields) == 0 || strings.TrimSpace(unknownColumn) == "" || len(generatedColumnFields) == 0 {
		return nil, false
	}
	fieldsToDrop := generatedColumnFields[normalizeSQLIdentifier(unknownColumn)]
	if len(fieldsToDrop) == 0 {
		return nil, false
	}

	drop := make(map[string]struct{}, len(fieldsToDrop))
	for _, field := range fieldsToDrop {
		drop[strings.ToLower(strings.TrimSpace(field))] = struct{}{}
	}

	next := make([]string, 0, len(searchFields))
	for _, field := range searchFields {
		if _, ok := drop[strings.ToLower(strings.TrimSpace(field))]; ok {
			continue
		}
		next = append(next, field)
	}
	return next, len(next) < len(searchFields)
}

func searchFieldUsesRetryableGeneratedRead(schema *models.ObjectMetadata, field *models.FieldMetadata) bool {
	return searchFieldUsesRetryableGeneratedReadForField(schema, field, make(map[string]struct{}))
}

func generatedSearchFieldsBlockedByCompoundTargets(schema *models.ObjectMetadata, searchFields []string) map[string]struct{} {
	blocked := make(map[string]struct{})
	for _, apiName := range searchFields {
		field := models.FindField(schema, apiName)
		if field == nil || !field.IsVirtual || field.VirtualKind != constants.VirtualFieldCompound {
			continue
		}
		if searchFieldUsesRetryableGeneratedRead(schema, field) {
			continue
		}
		collectRetryableGeneratedCompoundMembers(schema, field, blocked, make(map[string]struct{}))
	}
	if len(blocked) == 0 {
		return nil
	}
	return blocked
}

func collectRetryableGeneratedCompoundMembers(schema *models.ObjectMetadata, field *models.FieldMetadata, out map[string]struct{}, seen map[string]struct{}) {
	if field == nil {
		return
	}
	key := strings.ToLower(field.APIName)
	if _, ok := seen[key]; ok {
		return
	}
	seen[key] = struct{}{}
	defer delete(seen, key)

	if field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound {
		for _, member := range field.ComposedFrom {
			collectRetryableGeneratedCompoundMembers(schema, models.FindField(schema, member), out, seen)
		}
		return
	}
	if searchFieldUsesRetryableGeneratedReadForField(schema, field, seen) {
		out[key] = struct{}{}
	}
}

func searchFieldUsesRetryableGeneratedReadForField(schema *models.ObjectMetadata, field *models.FieldMetadata, seen map[string]struct{}) bool {
	if field == nil {
		return false
	}
	if field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound {
		key := strings.ToLower(field.APIName)
		if _, ok := seen[key]; ok {
			return false
		}
		seen[key] = struct{}{}
		defer delete(seen, key)

		if len(field.ComposedFrom) == 0 {
			return false
		}
		for _, member := range field.ComposedFrom {
			memberField := models.FindField(schema, member)
			if !searchFieldUsesRetryableGeneratedReadForField(schema, memberField, seen) {
				return false
			}
		}
		return true
	}
	resolved, err := formulastorage.Resolve(formulastorage.SpecFromField(field))
	if err != nil {
		return false
	}
	return resolved.RequiresFormulaView()
}

// prefixSearchExpressions returns the SQL expressions a prefix LIKE must anchor
// against for the /search/suggestions path, one per logical search target.
//
// /search/suggestions is contractually a NAME-prefix surface: a suggestion must
// match the COMPOSED display name, not an arbitrary member. The caller restricts
// searchFields to the SINGLE display name field the response projects as "Name"
// (suggestionSearchFields), so this helper receives just that one field —
// never bare compound members alongside their parent. For a virtual compound
// name field (e.g. Name composed of FirstName/LastName) it emits ONE anchored
// LIKE over the CONCAT_WS composed-name expression — the same
// NULLIF(TRIM(CONCAT_WS(' ', ...))) virtualCompoundSQLExpression builds — so the
// prefix matches the composed display name "Ada Lovelace", never a member: typing
// "Lov" must not match because the composed name does not start with "Lov".
//
// A LIKE over CONCAT_WS cannot use a B-tree index, so the compound case is NOT
// index-eligible — that is accepted: correctness of the composed-name prefix
// wins over the index, compound-name objects are rare, and the future name-index
// work does not cover compound names anyway. Ordinary single physical-column
// name fields keep the bare anchored `col LIKE 'term%'` (index-eligible).
//
// The expansion is driven entirely off compound metadata — there is no
// SObject-name branch.
func prefixSearchExpressions(schema *models.ObjectMetadata, searchFields []string, tableAlias string) ([]string, error) {
	exprs := make([]string, 0, len(searchFields))
	seen := make(map[string]struct{}, len(searchFields))
	for _, apiName := range searchFields {
		if !isValidFieldName(apiName) {
			return nil, fmt.Errorf("invalid field name")
		}
		field := models.FindField(schema, apiName)
		if field != nil && field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound && len(field.ComposedFrom) > 0 {
			// Anchor the composed-name expression, not the members.
			if expr := virtualCompoundSQLExpression(schema, apiName, tableAlias); expr != "" {
				if _, dup := seen[expr]; !dup {
					seen[expr] = struct{}{}
					exprs = append(exprs, expr)
				}
				continue
			}
			// Compound parent with no resolvable members: fall through to the
			// parent's own physical column rather than emitting nothing.
		}
		expr := qualifiedPhysicalColumn(schema, apiName, tableAlias)
		if _, dup := seen[expr]; !dup {
			seen[expr] = struct{}{}
			exprs = append(exprs, expr)
		}
	}
	return exprs, nil
}

// searchColumnsAllPhysical reports whether every column a search predicate
// references resolves to a real physical base-table column. The base-table fast
// path (ForceBaseTable) — for both the /search/ suggestions anchored prefix LIKE
// and the full substring `LOWER(col) LIKE '%term%'` — is only valid when this
// holds; an object whose name/search field is active-computed has no physical
// column on the base table and must be routed through the formula view instead,
// where the computed column exists. The compound branch mirrors the columns the
// emitted predicate references — a compound name's composed CONCAT_WS expression
// reads its physical member columns (off field.ComposedFrom) — so the gate and
// the emitted predicate stay in lock-step: if any member is view-only, the read
// is demoted to the view where that column exists.
// fieldHasBarePhysicalColumn reports whether a BARE `base`.`<field>` reference
// (no CONCAT_WS composition, no logical->physical alias mapping) resolves to a
// real column on the base table. This is stricter than
// FieldRequiresPhysicalColumn: a virtual-compound field (e.g. a Contact/User
// `Name` synthesized from FirstName/LastName) reports RequiresPhysicalColumn ==
// true, yet has NO own base-table column — the LIKE search path composes it via
// CONCAT_WS over physical members, but the parsed-term and ORDER BY paths emit a
// bare column reference that would be an unknown column on the base table. So a
// bare reference is only safe for a field that (a) is not a view-only formula
// and (b) is not a virtual-compound synthesized field. Mirrors the
// VirtualFieldCompound recognition in searchColumnsAllPhysical.
func fieldHasBarePhysicalColumn(field *models.FieldMetadata) bool {
	if field == nil {
		return false
	}
	if !formulastorage.FieldRequiresPhysicalColumn(field) {
		return false
	}
	if field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound {
		return false
	}
	return true
}

// searchTermFieldPhysical reports whether a substring-search term is safe to run
// against the base table with respect to the formula-query parser. The non-prefix
// search path feeds the term to query.ParseFormulaQuery: a `field op value` shape
// (e.g. "Score = 5") becomes `base`.`field` with the RAW api name and no
// logical->physical mapping. So the base-table fast path is only safe when either
// (a) the term is NOT a formula-query shape (it falls through to the LOWER(col)
// LIKE substring scan over the already-gated search fields), or (b) the parsed
// field has a bare physical base-table column whose PhysicalName is not aliased
// away from its api name. Anything else (view-only field, virtual-compound field
// such as logical Name, aliased PhysicalName, unknown field) would be an unknown
// column on the base table, so we demote to the view where ParseFormulaQuery's
// unmapped `view`.`field` exists.
func searchTermFieldPhysical(schema *models.ObjectMetadata, term string) bool {
	field, ok := query.FormulaQueryField(term)
	if !ok {
		return true
	}
	meta := models.FindField(schema, field)
	if !fieldHasBarePhysicalColumn(meta) {
		return false
	}
	// ParseFormulaQuery emits the raw api name; an aliased PhysicalName means that
	// api-named column does not exist on the base table.
	if meta.PhysicalName != "" && !strings.EqualFold(meta.PhysicalName, meta.APIName) {
		return false
	}
	return true
}

// searchSortFieldPhysical reports whether a caller-supplied ORDER BY field is
// safe on the base table. The query builder maps the sort field api name to its
// PhysicalName (resolveColumnName), so an aliased physical column is fine — but a
// field with no bare base-table column resolves to a column that exists only on
// the formula view: a view-only (active-computed) field, or a virtual-compound
// logical Name (e.g. Contact/User) that the view exposes but the base table
// synthesizes. Either is a syntactically valid identifier, so the builder emits
// it rather than dropping it, and the base-table read fails with an unknown
// column. An empty orderBy (default Id ordering) keeps the fast path. A field
// unknown to metadata is treated conservatively as NOT base-safe: only the view
// is guaranteed to expose logical Name and other display columns, so demoting
// matches the pre-existing view-backed read rather than risking a bare
// `base`.`<field>` that does not exist.
func searchSortFieldPhysical(schema *models.ObjectMetadata, orderBy string) bool {
	orderBy = strings.TrimSpace(orderBy)
	if orderBy == "" {
		return true
	}
	return fieldHasBarePhysicalColumn(models.FindField(schema, orderBy))
}

func searchColumnsAllPhysical(schema *models.ObjectMetadata, fields []string) bool {
	for _, apiName := range fields {
		field := models.FindField(schema, apiName)
		if field == nil {
			continue
		}
		if field.IsVirtual && field.VirtualKind == constants.VirtualFieldCompound && len(field.ComposedFrom) > 0 {
			hasResolvableMember := false
			for _, member := range field.ComposedFrom {
				memberField := models.FindField(schema, member)
				if memberField == nil {
					continue
				}
				hasResolvableMember = true
				if !formulastorage.FieldRequiresPhysicalColumn(memberField) {
					return false
				}
			}
			if hasResolvableMember {
				continue
			}
			// No resolvable members: prefixSearchExpressions falls back to the
			// parent's own physical column, so the parent itself must be physical.
		}
		if !formulastorage.FieldRequiresPhysicalColumn(field) {
			return false
		}
	}
	return true
}

// returnColumnsAllPhysical reports whether every projected return/select field
// resolves to a real physical base-table column. Unlike the suggestions prefix
// path (which returns only Id + Name), the full substring search projects the
// caller's requested return fields, so the base-table fast path is only safe
// when each of them physically exists on the base table — selecting a view-only
// formula column from the base table is illegal SQL and would error the read.
// A wildcard ("*") cannot be guaranteed physical-only and would silently drop
// formula columns the caller expects, so it always demotes to the view. This is
// intentionally stricter than searchColumnsAllPhysical: a virtual-compound Name
// can be searched through a CONCAT_WS expression over physical members, but a
// SELECT projection still emits the logical field itself unless the read target
// is the formula view. On the base table that bare Name column does not exist.
func returnColumnsAllPhysical(schema *models.ObjectMetadata, fields []string) bool {
	for _, apiName := range fields {
		if strings.TrimSpace(apiName) == "*" {
			return false
		}
		if !fieldHasBarePhysicalColumn(models.FindField(schema, apiName)) {
			return false
		}
	}
	return true
}

// schemaHasOpaquePredicateUnsafeField reports whether a pre-compiled predicate
// string (RLS criteria SQL or a filter expression) could reference a field using
// a logical/API name that is unsafe on the base table. Unlike builder-managed
// search/sort columns, those opaque SQL fragments are already compiled; this
// layer cannot rewrite them to PhysicalName or composed-member expressions.
// Demote if any field lacks a bare base column (view-only formula or virtual
// compound Name) or if a physical field is stored under an aliased PhysicalName.
func schemaHasOpaquePredicateUnsafeField(schema *models.ObjectMetadata) bool {
	if schema == nil {
		return false
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if !fieldHasBarePhysicalColumn(field) {
			return true
		}
		if field.PhysicalName != "" && !strings.EqualFold(field.PhysicalName, field.APIName) {
			return true
		}
	}
	return false
}

// prefixLikePattern builds the anchored LIKE parameter for prefix mode and
// escapes the LIKE wildcards (% and _) in the user term so a literal typed
// wildcard does not widen the match. The trailing % is the anchor.
func prefixLikePattern(term string) string {
	escaped := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(term)
	return escaped + "%"
}

// schemaHasIsDeleted checks if the schema has an IsDeleted field.
func schemaHasIsDeleted(schema *models.ObjectMetadata) bool {
	for _, f := range schema.Fields {
		if strings.EqualFold(f.APIName, constants.FieldIsDeleted) {
			return true
		}
	}
	return false
}

// isValidFieldName checks if a field name is safe (alphanumeric + underscore + dot)
func isValidFieldName(name string) bool {
	if len(name) == 0 {
		return false
	}
	for _, c := range name {
		if !((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '.') {
			return false
		}
	}
	return true
}

// replaceIdentifiers safely replaces identifiers in a SQL string using a map, ignoring content in quotes.
// It is case-insensitive for matching keys in the map.
func replaceIdentifiers(sql string, replacements map[string]string) string {
	var sb strings.Builder
	var tokenBuilder strings.Builder
	inQuote := false
	var quoteChar rune

	flushToken := func() {
		if tokenBuilder.Len() > 0 {
			token := tokenBuilder.String()
			// Check if token matches a replacement (case-insensitive)
			if replacement, ok := replacements[strings.ToLower(token)]; ok {
				sb.WriteString(replacement)
			} else {
				sb.WriteString(token)
			}
			tokenBuilder.Reset()
		}
	}

	for _, r := range sql {
		if inQuote {
			sb.WriteRune(r)
			if r == quoteChar {
				inQuote = false
			}
			continue
		}

		if r == '\'' || r == '"' || r == '`' {
			flushToken()
			inQuote = true
			quoteChar = r
			sb.WriteRune(r)
			continue
		}

		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' {
			tokenBuilder.WriteRune(r)
		} else {
			flushToken()
			sb.WriteRune(r)
		}
	}
	flushToken()

	return sb.String()
}

// isValidFilterOp checks if the given operator is in the valid set.
func isValidFilterOp(op string) bool {
	return validFilterOps[strings.ToUpper(op)]
}

// buildFieldReplacements builds a map of API name → qualified physical column for HAVING clause resolution.
func buildFieldReplacements(schema *models.ObjectMetadata, tableAlias string) map[string]string {
	replacements := make(map[string]string)
	if schema == nil {
		return replacements
	}
	if strings.TrimSpace(tableAlias) == "" {
		tableAlias = schema.APIName
	}
	for _, f := range schema.Fields {
		colName := actualColumnName(f.APIName, f.PhysicalName)
		if colName != f.APIName {
			replacements[strings.ToLower(f.APIName)] = fmt.Sprintf("`%s`.`%s`", tableAlias, colName)
		}
	}
	return replacements
}

// buildQualifiedFieldReplacements builds a map of both API names and physical column
// names to fully-qualified columns for contexts where additional joins can make
// unqualified filter fields ambiguous.
func buildQualifiedFieldReplacements(schema *models.ObjectMetadata, tableAlias string) map[string]string {
	replacements := make(map[string]string)
	if schema == nil || tableAlias == "" {
		return replacements
	}

	for _, f := range schema.Fields {
		colName := readTargetColumnName(schema, tableAlias, f.APIName)
		qualified := fmt.Sprintf("`%s`.`%s`", tableAlias, colName)
		replacements[strings.ToLower(f.APIName)] = qualified
		replacements[strings.ToLower(actualColumnName(f.APIName, f.PhysicalName))] = qualified
	}

	return replacements
}

func idSearchCondition(schema *models.ObjectMetadata, field *models.FieldMetadata, term string, tableAlias string) (string, any, bool) {
	if field == nil || !strings.EqualFold(field.APIName, constants.FieldID) {
		return "", nil, false
	}
	trimmed := strings.TrimSpace(term)
	if isCanonicalSalesforceID(trimmed) {
		return fmt.Sprintf("%s = ?", qualifiedPhysicalColumn(schema, field.APIName, tableAlias)), trimmed, true
	}
	if isSafeSalesforceIDPrefix(trimmed) {
		return fmt.Sprintf("%s LIKE ? ESCAPE '\\\\'", qualifiedPhysicalColumn(schema, field.APIName, tableAlias)), prefixLikePattern(trimmed), true
	}
	return "", nil, false
}

func isCanonicalSalesforceID(value string) bool {
	l := len(value)
	return (l == 15 || l == 18) && isSalesforceIDChars(value)
}

func isSafeSalesforceIDPrefix(value string) bool {
	l := len(value)
	return l >= 5 && l < 18 && isSalesforceIDChars(value)
}

func isSalesforceIDChars(value string) bool {
	if value == "" {
		return false
	}
	for _, c := range value {
		if !((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9')) {
			return false
		}
	}
	return true
}