ResolveSpec/pkg/restheadspec/headers.go

package restheadspec

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"reflect"
	"regexp"
	"sort"
	"strconv"
	"strings"

	"github.com/bitechdev/ResolveSpec/pkg/common"
	"github.com/bitechdev/ResolveSpec/pkg/logger"
	"github.com/bitechdev/ResolveSpec/pkg/reflection"
)

// ExtendedRequestOptions extends common.RequestOptions with additional features
type ExtendedRequestOptions struct {
	common.RequestOptions

	// Field selection
	CleanJSON bool

	// Advanced filtering
	SearchColumns  []string
	CustomSQLWhere string
	CustomSQLOr    string

	// Joins
	Expand        []ExpandOption
	CustomSQLJoin []string // Custom SQL JOIN clauses
	JoinAliases   []string // Extracted table aliases from CustomSQLJoin for validation

	// Advanced features
	AdvancedSQL map[string]string // Column -> SQL expression
	ComputedQL  map[string]string // Column -> CQL expression
	Distinct    bool
	SkipCount   bool
	SkipCache   bool
	PKRow       *string

	// Response format
	ResponseFormat string // "simple", "detail", "syncfusion"

	// Single record normalization - convert single-element arrays to objects
	SingleRecordAsObject bool

	// Transaction
	AtomicTransaction bool

	// X-Files configuration - comprehensive query options as a single JSON object
	XFiles        *XFiles
	XFilesPresent bool // Flag to indicate if X-Files header was provided
}

// ExpandOption represents a relation expansion configuration
type ExpandOption struct {
	Relation string
	Columns  []string
	Where    string
	Sort     string
}

// decodeHeaderValue decodes base64 encoded header values
// Supports ZIP_ and __ prefixes for base64 encoding
func decodeHeaderValue(value string) string {
	str, err := DecodeParam(value)
	if err != nil {
		return value
	}
	return str
}

// DecodeParam - Decodes parameter string and returns unencoded string
func DecodeParam(pStr string) (string, error) {
	var code = pStr
	if strings.HasPrefix(pStr, "ZIP_") {
		code = strings.ReplaceAll(pStr, "ZIP_", "")
		code = strings.ReplaceAll(code, "\n", "")
		code = strings.ReplaceAll(code, "\r", "")
		code = strings.ReplaceAll(code, " ", "")
		strDat, err := base64.StdEncoding.DecodeString(code)
		if err != nil {
			return code, fmt.Errorf("failed to read parameter base64: %v", err)
		} else {
			code = string(strDat)
		}
	} else if strings.HasPrefix(pStr, "__") {
		code = strings.ReplaceAll(pStr, "__", "")
		code = strings.ReplaceAll(code, "\n", "")
		code = strings.ReplaceAll(code, "\r", "")
		code = strings.ReplaceAll(code, " ", "")

		strDat, err := base64.StdEncoding.DecodeString(code)
		if err != nil {
			return code, fmt.Errorf("failed to read parameter base64: %v", err)
		} else {
			code = string(strDat)
		}
	}

	if strings.HasPrefix(code, "ZIP_") || strings.HasPrefix(code, "__") {
		code, _ = DecodeParam(code)
	}

	return code, nil
}

// parseOptionsFromHeaders parses all request options from HTTP headers
// If model is provided, it will resolve table names to field names in preload/expand options
func (h *Handler) parseOptionsFromHeaders(r common.Request, model interface{}) ExtendedRequestOptions {
	options := ExtendedRequestOptions{
		RequestOptions: common.RequestOptions{
			Filters: make([]common.FilterOption, 0),
			Sort:    make([]common.SortOption, 0),
			Preload: make([]common.PreloadOption, 0),
		},
		AdvancedSQL:          make(map[string]string),
		ComputedQL:           make(map[string]string),
		Expand:               make([]ExpandOption, 0),
		CustomSQLJoin:        make([]string, 0),
		ResponseFormat:       "simple", // Default response format
		SingleRecordAsObject: true,     // Default: normalize single-element arrays to objects
	}

	// Get all headers
	headers := r.AllHeaders()

	// Get all query parameters
	queryParams := r.AllQueryParams()

	// Merge headers and query parameters - query parameters take precedence
	// This allows the same parameters to be specified in either headers or query string
	// Normalize keys to lowercase to ensure query params properly override headers
	combinedParams := make(map[string]string)
	for key, value := range headers {
		combinedParams[strings.ToLower(key)] = value
	}
	for key, value := range queryParams {
		combinedParams[strings.ToLower(key)] = value
	}

	sortedKeys := make([]string, 0, len(combinedParams))
	for key := range combinedParams {
		sortedKeys = append(sortedKeys, key)
	}
	sort.Slice(sortedKeys, func(i, j int) bool {
		if sortedKeys[i] != sortedKeys[j] {
			return sortedKeys[i] < sortedKeys[j]
		}
		return combinedParams[sortedKeys[i]] < combinedParams[sortedKeys[j]]
	})

	// Process each parameter (from both headers and query params)
	// Note: keys are already normalized to lowercase in combinedParams
	for _, key := range sortedKeys {
		value := combinedParams[key]
		// Decode value if it's base64 encoded
		decodedValue := decodeHeaderValue(value)

		// Parse based on parameter prefix/name
		switch {
		// Field Selection
		case strings.HasPrefix(key, "x-select-fields"):
			h.parseSelectFields(&options, decodedValue)
		case strings.HasPrefix(key, "x-not-select-fields"):
			h.parseNotSelectFields(&options, decodedValue)
		case strings.HasPrefix(key, "x-clean-json"):
			options.CleanJSON = strings.EqualFold(decodedValue, "true")

		// Filtering & Search
		case strings.HasPrefix(key, "x-fieldfilter-"):
			h.parseFieldFilter(&options, key, decodedValue)
		case strings.HasPrefix(key, "x-searchfilter-"):
			h.parseSearchFilter(&options, key, decodedValue)
		case strings.HasPrefix(key, "x-searchop-"):
			h.parseSearchOp(&options, key, decodedValue, "AND")
		case strings.HasPrefix(key, "x-searchor-"):
			h.parseSearchOp(&options, key, decodedValue, "OR")
		case strings.HasPrefix(key, "x-searchand-"):
			h.parseSearchOp(&options, key, decodedValue, "AND")
		case strings.HasPrefix(key, "x-searchcols"):
			options.SearchColumns = h.parseCommaSeparated(decodedValue)
		case strings.HasPrefix(key, "x-spatialfilter-"):
			h.parseGeoFilter(&options, key, "x-spatialfilter-", decodedValue)
		case strings.HasPrefix(key, "x-vectorfilter-"):
			h.parseGeoFilter(&options, key, "x-vectorfilter-", decodedValue)

		// pgvector KNN search
		case key == "x-vector-search-vector":
			h.ensureVectorSearch(&options).Vector = parseFloat32List(decodedValue)
		case key == "x-vector-search-as":
			h.ensureVectorSearch(&options).As = decodedValue
		case key == "x-vector-search-dir":
			h.ensureVectorSearch(&options).Direction = decodedValue
		case strings.HasPrefix(key, "x-vector-search-"):
			vs := h.ensureVectorSearch(&options)
			vs.Column = strings.TrimPrefix(key, "x-vector-search-")
			vs.Metric = decodedValue
		case strings.HasPrefix(key, "x-custom-sql-w"):
			if options.CustomSQLWhere != "" {
				options.CustomSQLWhere = fmt.Sprintf("%s AND (%s)", options.CustomSQLWhere, decodedValue)
			} else {
				options.CustomSQLWhere = decodedValue
			}
		case strings.HasPrefix(key, "x-custom-sql-or"):
			if options.CustomSQLOr != "" {
				options.CustomSQLOr = fmt.Sprintf("%s OR (%s)", options.CustomSQLOr, decodedValue)
			} else {
				options.CustomSQLOr = decodedValue
			}

		// Joins & Relations
		case strings.HasPrefix(key, "x-preload"):
			if strings.HasSuffix(key, "-where") {
				continue
			}
			whereClaude := combinedParams[fmt.Sprintf("%s-where", key)]
			h.parsePreload(&options, decodedValue, decodeHeaderValue(whereClaude))

		case strings.HasPrefix(key, "x-expand"):
			h.parseExpand(&options, decodedValue)
		case strings.HasPrefix(key, "x-custom-sql-join"):
			h.parseCustomSQLJoin(&options, decodedValue)

		// Sorting & Pagination
		case strings.HasPrefix(key, "x-sort"):
			h.parseSorting(&options, decodedValue)
		// Special cases for older clients using sort(a,b,-c) syntax
		case strings.HasPrefix(key, "sort(") && strings.Contains(key, ")"):
			sortValue := key[strings.Index(key, "(")+1 : strings.Index(key, ")")]
			h.parseSorting(&options, sortValue)
		case strings.HasPrefix(key, "x-limit"):
			if limit, err := strconv.Atoi(decodedValue); err == nil {
				options.Limit = &limit
			}
		// Special cases for older clients using limit(n) syntax
		case strings.HasPrefix(key, "limit(") && strings.Contains(key, ")"):
			limitValue := key[strings.Index(key, "(")+1 : strings.Index(key, ")")]
			limitValueParts := strings.Split(limitValue, ",")

			if len(limitValueParts) > 1 {
				if limit, err := strconv.Atoi(limitValueParts[0]); err == nil {
					options.Limit = &limit
				}
				if offset, err := strconv.Atoi(limitValueParts[1]); err == nil {
					options.Offset = &offset
				}

			} else {
				if limit, err := strconv.Atoi(limitValueParts[0]); err == nil {
					options.Limit = &limit
				}
			}

		case strings.HasPrefix(key, "x-offset"):
			if offset, err := strconv.Atoi(decodedValue); err == nil {
				options.Offset = &offset
			}

		case strings.HasPrefix(key, "x-cursor-forward"):
			options.CursorForward = decodedValue
		case strings.HasPrefix(key, "x-cursor-backward"):
			options.CursorBackward = decodedValue

		// Advanced Features
		case strings.HasPrefix(key, "x-advsql-"):
			colName := strings.TrimPrefix(key, "x-advsql-")
			options.AdvancedSQL[colName] = decodedValue
		case strings.HasPrefix(key, "x-cql-sel-"):
			colName := strings.TrimPrefix(key, "x-cql-sel-")
			options.ComputedQL[colName] = decodedValue

		case strings.HasPrefix(key, "x-distinct"):
			options.Distinct = strings.EqualFold(decodedValue, "true")
		case strings.HasPrefix(key, "x-skipcount"):
			options.SkipCount = strings.EqualFold(decodedValue, "true")
		case strings.HasPrefix(key, "x-skipcache"):
			options.SkipCache = strings.EqualFold(decodedValue, "true")
		case strings.HasPrefix(key, "x-fetch-rownumber"):
			options.FetchRowNumber = &decodedValue
		case strings.HasPrefix(key, "x-pkrow"):
			options.PKRow = &decodedValue

		// Response Format
		case strings.HasPrefix(key, "x-simpleapi"):
			options.ResponseFormat = "simple"
		case strings.HasPrefix(key, "x-detailapi"):
			options.ResponseFormat = "detail"
		case strings.HasPrefix(key, "x-syncfusion"):
			options.ResponseFormat = "syncfusion"
		case strings.HasPrefix(key, "x-single-record-as-object"):
			// Parse as boolean - "false" disables, "true" enables (default is true)
			if strings.EqualFold(decodedValue, "false") {
				options.SingleRecordAsObject = false
			} else if strings.EqualFold(decodedValue, "true") {
				options.SingleRecordAsObject = true
			}

		// Transaction Control
		case strings.HasPrefix(key, "x-transaction-atomic"):
			options.AtomicTransaction = strings.EqualFold(decodedValue, "true")

		// X-Files - comprehensive JSON configuration
		case strings.HasPrefix(key, "x-files"):
			h.parseXFiles(&options, decodedValue)
		}
	}

	// Resolve relation names (convert table names/prefixes to actual model field names) if model is provided.
	// This runs for both regular headers and X-Files, because XFile prefixes don't always match model
	// field names (e.g., prefix "HUB" vs field "HUB_RID_HUB"). RelatedKey/ForeignKey are used to
	// disambiguate when multiple fields point to the same related type.
	if model != nil {
		h.resolveRelationNamesInOptions(&options, model)
	}

	// Always sort according to the primary key if no sorting is specified
	if len(options.Sort) == 0 {
		pkName := reflection.GetPrimaryKeyName(model)
		options.Sort = []common.SortOption{{Column: pkName, Direction: "ASC"}}
	}

	return options
}

// ensureVectorSearch returns the options' VectorSearchOption, allocating it on
// first use.
func (h *Handler) ensureVectorSearch(options *ExtendedRequestOptions) *common.VectorSearchOption {
	if options.VectorSearch == nil {
		options.VectorSearch = &common.VectorSearchOption{}
	}
	return options.VectorSearch
}

// parseFloat32List parses a JSON array ("[1,2,3]") or comma-separated list into
// a []float32.
func parseFloat32List(value string) []float32 {
	value = strings.TrimSpace(value)
	if value == "" {
		return nil
	}
	var raw []float64
	if err := json.Unmarshal([]byte(value), &raw); err == nil {
		out := make([]float32, len(raw))
		for i, f := range raw {
			out[i] = float32(f)
		}
		return out
	}
	parts := strings.Split(strings.Trim(value, "[]"), ",")
	out := make([]float32, 0, len(parts))
	for _, p := range parts {
		f, err := strconv.ParseFloat(strings.TrimSpace(p), 32)
		if err != nil {
			return nil
		}
		out = append(out, float32(f))
	}
	return out
}

// parseGeoFilter parses an x-spatialfilter-<col> / x-vectorfilter-<col> header.
// The value is a JSON object: {"op":"st_dwithin","geom":...,"distance":...} or
// {"op":"st_intersects","value":<geojson>}. An optional "logic":"or" controls
// how the filter combines with the previous one.
func (h *Handler) parseGeoFilter(options *ExtendedRequestOptions, key, prefix, value string) {
	col := strings.TrimPrefix(key, prefix)
	if col == "" || strings.TrimSpace(value) == "" {
		return
	}
	var raw map[string]interface{}
	if err := json.Unmarshal([]byte(value), &raw); err != nil {
		logger.Warn("Invalid %s%s filter JSON: %v", prefix, col, err)
		return
	}
	op, _ := raw["op"].(string)
	if op == "" {
		logger.Warn("%s%s filter missing \"op\"", prefix, col)
		return
	}
	logicOp := "AND"
	if lo, ok := raw["logic"].(string); ok && strings.EqualFold(lo, "or") {
		logicOp = "OR"
	}

	var fv interface{}
	if v, ok := raw["value"]; ok {
		fv = v
	} else {
		delete(raw, "op")
		delete(raw, "logic")
		fv = raw
	}

	options.Filters = append(options.Filters, common.FilterOption{
		Column:        col,
		Operator:      op,
		Value:         fv,
		LogicOperator: logicOp,
	})
}

// parseSelectFields parses x-select-fields header
func (h *Handler) parseSelectFields(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}
	options.Columns = h.parseCommaSeparated(value)
	if len(options.Columns) > 1 {
		options.CleanJSON = true
	}
}

// parseNotSelectFields parses x-not-select-fields header
func (h *Handler) parseNotSelectFields(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}
	options.OmitColumns = h.parseCommaSeparated(value)
	if len(options.OmitColumns) > 1 {
		options.CleanJSON = true
	}
}

// parseFieldFilter parses x-fieldfilter-{colname} header (exact match)
func (h *Handler) parseFieldFilter(options *ExtendedRequestOptions, headerKey, value string) {
	colName := strings.TrimPrefix(headerKey, "x-fieldfilter-")
	options.Filters = append(options.Filters, common.FilterOption{
		Column:        colName,
		Operator:      "eq",
		Value:         value,
		LogicOperator: "AND", // Default to AND
		NullAsZero:    true,
	})
}

// parseSearchFilter parses x-searchfilter-{colname} header (ILIKE search)
func (h *Handler) parseSearchFilter(options *ExtendedRequestOptions, headerKey, value string) {
	colName := strings.TrimPrefix(headerKey, "x-searchfilter-")
	// Use ILIKE for fuzzy search
	options.Filters = append(options.Filters, common.FilterOption{
		Column:        colName,
		Operator:      "ilike",
		Value:         "%" + value + "%",
		LogicOperator: "AND", // Default to AND
	})
}

// parseSearchOp parses x-searchop-{operator}-{colname} and x-searchor-{operator}-{colname}
func (h *Handler) parseSearchOp(options *ExtendedRequestOptions, headerKey, value, logicOp string) {
	// Extract operator and column name
	// Format: x-searchop-{operator}-{colname} or x-searchor-{operator}-{colname}
	var prefix string
	if logicOp == "OR" {
		prefix = "x-searchor-"
	} else {
		prefix = "x-searchop-"
		if strings.HasPrefix(headerKey, "x-searchand-") {
			prefix = "x-searchand-"
		}
	}

	rest := strings.TrimPrefix(headerKey, prefix)
	parts := strings.SplitN(rest, "-", 2)
	if len(parts) != 2 {
		logger.Warn("Invalid search operator header format: %s", headerKey)
		return
	}

	operator := parts[0]
	colName := parts[1]

	if strings.HasPrefix(colName, "cql") {
		// Computed column - Will not filter on it
		logger.Warn("Search operators on computed columns are not supported: %s", colName)
		return
	}

	// Map operator names to filter operators
	filterOp := h.mapSearchOperator(colName, operator, value)

	// Set the logic operator (AND or OR)
	filterOp.LogicOperator = logicOp

	options.Filters = append(options.Filters, filterOp)

	logger.Debug("%s logic filter: %s %s %v", logicOp, colName, filterOp.Operator, filterOp.Value)
}

// mapSearchOperator maps search operator names to filter operators
func (h *Handler) mapSearchOperator(colName, operator, value string) common.FilterOption {
	operator = strings.ToLower(operator)

	switch operator {
	case "contains", "contain", "like":
		return common.FilterOption{Column: colName, Operator: "ilike", Value: "%" + value + "%"}
	case "notcontains", "notcontain", "notlike":
		return common.FilterOption{Column: colName, Operator: "not_ilike", Value: "%" + value + "%"}
	case "beginswith", "startswith":
		return common.FilterOption{Column: colName, Operator: "ilike", Value: value + "%"}
	case "endswith":
		return common.FilterOption{Column: colName, Operator: "ilike", Value: "%" + value}
	case "equals", "eq", "=":
		return common.FilterOption{Column: colName, Operator: "eq", Value: value}
	case "notequals", "neq", "ne", "!=", "<>":
		return common.FilterOption{Column: colName, Operator: "neq", Value: value}
	case "greaterthan", "gt", ">":
		return common.FilterOption{Column: colName, Operator: "gt", Value: value}
	case "lessthan", "lt", "<":
		return common.FilterOption{Column: colName, Operator: "lt", Value: value}
	case "greaterthanorequal", "gte", "ge", ">=":
		return common.FilterOption{Column: colName, Operator: "gte", Value: value}
	case "lessthanorequal", "lte", "le", "<=":
		return common.FilterOption{Column: colName, Operator: "lte", Value: value}
	case "between":
		// Parse between values (format: "value1,value2")
		// Between is exclusive (> value1 AND < value2)
		parts := strings.Split(value, ",")
		if len(parts) == 2 {
			return common.FilterOption{Column: colName, Operator: "between", Value: parts}
		}
		return common.FilterOption{Column: colName, Operator: "eq", Value: value}
	case "betweeninclusive":
		// Parse between values (format: "value1,value2")
		// Between inclusive is >= value1 AND <= value2
		parts := strings.Split(value, ",")
		if len(parts) == 2 {
			return common.FilterOption{Column: colName, Operator: "between_inclusive", Value: parts}
		}
		return common.FilterOption{Column: colName, Operator: "eq", Value: value}
	case "in":
		// Parse IN values (format: "value1,value2,value3")
		values := strings.Split(value, ",")
		return common.FilterOption{Column: colName, Operator: "in", Value: values}
	case "empty", "isnull", "null":
		// Check for NULL or empty string
		return common.FilterOption{Column: colName, Operator: "is_null", Value: nil}
	case "notempty", "isnotnull", "notnull":
		// Check for NOT NULL
		return common.FilterOption{Column: colName, Operator: "is_not_null", Value: nil}
	default:
		logger.Warn("Unknown search operator: %s, defaulting to equals", operator)
		return common.FilterOption{Column: colName, Operator: "eq", Value: value}
	}
}

// parsePreload parses x-preload header
// Format: RelationName:field1,field2 or RelationName or multiple separated by |
func (h *Handler) parsePreload(options *ExtendedRequestOptions, values ...string) {
	if len(values) == 0 {
		return
	}
	value := values[0]
	whereClause := ""
	if len(values) > 1 {
		whereClause = values[1]
	}
	if value == "" {
		return
	}

	// Split by | for multiple preloads
	preloads := strings.Split(value, "|")
	for _, preloadStr := range preloads {
		preloadStr = strings.TrimSpace(preloadStr)
		if preloadStr == "" {
			continue
		}

		// Parse relation:columns format
		parts := strings.SplitN(preloadStr, ":", 2)
		preload := common.PreloadOption{
			Relation: strings.TrimSpace(parts[0]),
			Where:    whereClause,
		}

		if len(parts) == 2 {
			// Parse columns
			preload.Columns = h.parseCommaSeparated(parts[1])
		}

		options.Preload = append(options.Preload, preload)
	}
}

// parseExpand parses x-expand header (LEFT JOIN expansion)
// Format: RelationName:field1,field2 or RelationName or multiple separated by |
func (h *Handler) parseExpand(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}

	// Split by | for multiple expands
	expands := strings.Split(value, "|")
	for _, expandStr := range expands {
		expandStr = strings.TrimSpace(expandStr)
		if expandStr == "" {
			continue
		}

		// Parse relation:columns format
		parts := strings.SplitN(expandStr, ":", 2)
		expand := ExpandOption{
			Relation: strings.TrimSpace(parts[0]),
		}

		if len(parts) == 2 {
			// Parse columns
			expand.Columns = h.parseCommaSeparated(parts[1])
		}

		options.Expand = append(options.Expand, expand)
	}
}

// reMultiJoinBoundary finds the start of each individual JOIN clause within a string that
// may contain multiple consecutive JOIN clauses (e.g., "INNER JOIN ... LEFT OUTER JOIN ...").
var reMultiJoinBoundary = regexp.MustCompile(`(?i)(?:inner|left(?:\s+outer)?|right(?:\s+outer)?|full(?:\s+outer)?|cross)\s+join\b`)

// splitJoinClauses splits a SQL string that may contain multiple JOIN clauses into
// individual clauses. A plain pipe-separated segment may itself contain several JOINs;
// this function splits them so each gets its own alias entry.
func splitJoinClauses(joinStr string) []string {
	indices := reMultiJoinBoundary.FindAllStringIndex(joinStr, -1)
	if len(indices) <= 1 {
		return []string{strings.TrimSpace(joinStr)}
	}
	parts := make([]string, 0, len(indices))
	for i, idx := range indices {
		end := len(joinStr)
		if i+1 < len(indices) {
			end = indices[i+1][0]
		}
		if part := strings.TrimSpace(joinStr[idx[0]:end]); part != "" {
			parts = append(parts, part)
		}
	}
	return parts
}

// parseCustomSQLJoin parses x-custom-sql-join header
// Format: Single JOIN clause or multiple JOIN clauses separated by |
// Example: "LEFT JOIN departments d ON d.id = employees.department_id"
// Example: "LEFT JOIN departments d ON d.id = e.dept_id | INNER JOIN roles r ON r.id = e.role_id"
func (h *Handler) parseCustomSQLJoin(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}

	// Split by | for multiple joins
	joins := strings.Split(value, "|")
	for _, joinStr := range joins {
		joinStr = strings.TrimSpace(joinStr)
		if joinStr == "" {
			continue
		}

		// Basic validation: should contain "JOIN" keyword
		upperJoin := strings.ToUpper(joinStr)
		if !strings.Contains(upperJoin, "JOIN") {
			logger.Warn("Invalid custom SQL join (missing JOIN keyword): %s", joinStr)
			continue
		}

		// Sanitize the join clause using common.SanitizeWhereClause
		// Note: This is basic sanitization - in production you may want stricter validation
		sanitizedJoin := common.SanitizeWhereClause(joinStr, "", nil)
		if sanitizedJoin == "" {
			logger.Warn("Custom SQL join failed sanitization: %s", joinStr)
			continue
		}

		// Split into individual JOIN clauses so each clause gets its own alias entry.
		// CustomSQLJoin and JoinAliases are kept parallel (one entry per individual clause).
		for _, clause := range splitJoinClauses(sanitizedJoin) {
			alias := extractJoinAlias(clause)
			// Keep arrays parallel; use empty string when alias cannot be extracted.
			options.JoinAliases = append(options.JoinAliases, alias)
			options.RequestOptions.JoinAliases = append(options.RequestOptions.JoinAliases, alias)
			if alias != "" {
				logger.Debug("Extracted join alias: %s", alias)
			}
			logger.Debug("Adding custom SQL join: %s", clause)
			options.CustomSQLJoin = append(options.CustomSQLJoin, clause)
		}
	}
}

// extractJoinAlias extracts the table alias from a JOIN clause
// Examples:
//   - "LEFT JOIN departments d ON ..." -> "d"
//   - "INNER JOIN users AS u ON ..." -> "u"
//   - "JOIN roles r ON ..." -> "r"
//   - "INNER JOIN LATERAL (...) fn ON true" -> "fn"
func extractJoinAlias(joinClause string) string {
	upperJoin := strings.ToUpper(joinClause)

	// Find the "JOIN" keyword position
	joinIdx := strings.Index(upperJoin, "JOIN")
	if joinIdx == -1 {
		return ""
	}

	// Lateral joins: alias is the word after the closing ) and before ON
	if strings.Contains(upperJoin, "LATERAL") {
		lastClose := strings.LastIndex(joinClause, ")")
		if lastClose != -1 {
			words := strings.Fields(joinClause[lastClose+1:])
			// words should be like ["fn", "on", "true"] or ["on", "true"]
			if len(words) >= 1 && !strings.EqualFold(words[0], "on") {
				return words[0]
			}
		}
		return ""
	}

	// Regular joins: find the "ON" keyword position (first occurrence)
	onIdx := strings.Index(upperJoin, " ON ")
	if onIdx == -1 {
		return ""
	}

	// Extract the part between JOIN and ON
	betweenJoinAndOn := strings.TrimSpace(joinClause[joinIdx+4 : onIdx])

	// Split by spaces to get words
	words := strings.Fields(betweenJoinAndOn)
	if len(words) == 0 {
		return ""
	}

	// If there's an AS keyword, the alias is after it
	for i, word := range words {
		if strings.EqualFold(word, "AS") && i+1 < len(words) {
			return words[i+1]
		}
	}

	// Otherwise, the alias is the last word (if there are 2+ words)
	// Format: "table_name alias" or just "table_name"
	if len(words) >= 2 {
		return words[len(words)-1]
	}

	// Only one word means it's just the table name, no alias
	return ""
}

// parseSorting parses x-sort header
// Format: +field1,-field2,field3 (+ for ASC, - for DESC, default ASC)
func (h *Handler) parseSorting(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}

	sortFields := h.parseCommaSeparated(value)
	for _, field := range sortFields {
		field = strings.TrimSpace(field)
		if field == "" {
			continue
		}

		direction := "ASC"
		colName := field

		switch {
		case strings.HasPrefix(field, "-"):
			direction = "DESC"
			colName = strings.TrimPrefix(field, "-")
		case strings.HasPrefix(field, "+"):
			direction = "ASC"
			colName = strings.TrimPrefix(field, "+")
		case strings.HasSuffix(field, " desc"):
			direction = "DESC"
			colName = strings.TrimSuffix(field, "desc")
		case strings.HasSuffix(field, " asc"):
			direction = "ASC"
			colName = strings.TrimSuffix(field, "asc")
		}

		options.Sort = append(options.Sort, common.SortOption{
			Column:    strings.Trim(colName, " "),
			Direction: direction,
		})
	}
}

// parseCommaSeparated parses comma-separated values and trims whitespace
// It respects bracket nesting and only splits on commas outside of parentheses
func (h *Handler) parseCommaSeparated(value string) []string {
	if value == "" {
		return nil
	}

	result := make([]string, 0)
	var current strings.Builder
	nestingLevel := 0

	for _, char := range value {
		switch char {
		case '(':
			nestingLevel++
			current.WriteRune(char)
		case ')':
			nestingLevel--
			current.WriteRune(char)
		case ',':
			if nestingLevel == 0 {
				// We're outside all brackets, so split here
				part := strings.TrimSpace(current.String())
				if part != "" {
					result = append(result, part)
				}
				current.Reset()
			} else {
				// Inside brackets, keep the comma
				current.WriteRune(char)
			}
		default:
			current.WriteRune(char)
		}
	}

	// Add the last part
	part := strings.TrimSpace(current.String())
	if part != "" {
		result = append(result, part)
	}

	return result
}

// parseXFiles parses x-files header containing comprehensive JSON configuration
// and populates ExtendedRequestOptions fields from it
func (h *Handler) parseXFiles(options *ExtendedRequestOptions, value string) {
	if value == "" {
		return
	}

	var xfiles XFiles
	if err := json.Unmarshal([]byte(value), &xfiles); err != nil {
		logger.Warn("Failed to parse x-files header: %v", err)
		return
	}

	logger.Debug("Parsed x-files configuration for table: %s", xfiles.TableName)

	// Store the original XFiles for reference
	options.XFiles = &xfiles
	options.XFilesPresent = true // Mark that X-Files header was provided

	// Map XFiles fields to ExtendedRequestOptions

	// Column selection
	if len(xfiles.Columns) > 0 {
		options.Columns = append(options.Columns, xfiles.Columns...)
		logger.Debug("X-Files: Added columns: %v", xfiles.Columns)
	}

	// Omit columns
	if len(xfiles.OmitColumns) > 0 {
		options.OmitColumns = append(options.OmitColumns, xfiles.OmitColumns...)
		logger.Debug("X-Files: Added omit columns: %v", xfiles.OmitColumns)
	}

	// Computed columns (CQL) -> ComputedQL
	if len(xfiles.CQLColumns) > 0 {
		if options.ComputedQL == nil {
			options.ComputedQL = make(map[string]string)
		}
		for i, cqlExpr := range xfiles.CQLColumns {
			colName := fmt.Sprintf("cql%d", i+1)
			options.ComputedQL[colName] = cqlExpr
			logger.Debug("X-Files: Added computed column %s: %s", colName, cqlExpr)
		}
	}

	// Sorting
	if len(xfiles.Sort) > 0 {
		for _, sortField := range xfiles.Sort {
			direction := "ASC"
			colName := sortField

			// Handle direction prefixes
			if strings.HasPrefix(sortField, "-") {
				direction = "DESC"
				colName = strings.TrimPrefix(sortField, "-")
			} else if strings.HasPrefix(sortField, "+") {
				colName = strings.TrimPrefix(sortField, "+")
			}

			// Handle DESC suffix
			if strings.HasSuffix(strings.ToLower(colName), " desc") {
				direction = "DESC"
				colName = strings.TrimSuffix(strings.ToLower(colName), " desc")
			} else if strings.HasSuffix(strings.ToLower(colName), " asc") {
				colName = strings.TrimSuffix(strings.ToLower(colName), " asc")
			}

			options.Sort = append(options.Sort, common.SortOption{
				Column:    strings.TrimSpace(colName),
				Direction: direction,
			})
		}
		logger.Debug("X-Files: Added %d sort options", len(xfiles.Sort))
	}

	// Filter fields
	if len(xfiles.FilterFields) > 0 {
		for _, filterField := range xfiles.FilterFields {
			options.Filters = append(options.Filters, common.FilterOption{
				Column:        filterField.Field,
				Operator:      filterField.Operator,
				Value:         filterField.Value,
				LogicOperator: "AND", // Default to AND
			})
		}
		logger.Debug("X-Files: Added %d filter fields", len(xfiles.FilterFields))
	}

	// SQL AND conditions -> CustomSQLWhere
	if len(xfiles.SqlAnd) > 0 {
		if options.CustomSQLWhere != "" {
			options.CustomSQLWhere += " AND "
		}
		options.CustomSQLWhere += "(" + strings.Join(xfiles.SqlAnd, " AND ") + ")"
		logger.Debug("X-Files: Added SQL AND conditions")
	}

	// SQL OR conditions -> CustomSQLOr
	if len(xfiles.SqlOr) > 0 {
		if options.CustomSQLOr != "" {
			options.CustomSQLOr += " OR "
		}
		options.CustomSQLOr += "(" + strings.Join(xfiles.SqlOr, " OR ") + ")"
		logger.Debug("X-Files: Added SQL OR conditions")
	}

	// Pagination - Limit
	if limitStr := xfiles.Limit.String(); limitStr != "" && limitStr != "0" {
		if limitVal, err := xfiles.Limit.Int64(); err == nil && limitVal > 0 {
			limit := int(limitVal)
			options.Limit = &limit
			logger.Debug("X-Files: Set limit: %d", limit)
		}
	}

	// Pagination - Offset
	if offsetStr := xfiles.Offset.String(); offsetStr != "" && offsetStr != "0" {
		if offsetVal, err := xfiles.Offset.Int64(); err == nil && offsetVal > 0 {
			offset := int(offsetVal)
			options.Offset = &offset
			logger.Debug("X-Files: Set offset: %d", offset)
		}
	}

	// Cursor pagination
	if xfiles.CursorForward != "" {
		options.CursorForward = xfiles.CursorForward
		logger.Debug("X-Files: Set cursor forward")
	}
	if xfiles.CursorBackward != "" {
		options.CursorBackward = xfiles.CursorBackward
		logger.Debug("X-Files: Set cursor backward")
	}

	// Flags
	if xfiles.Skipcount {
		options.SkipCount = true
		logger.Debug("X-Files: Set skip count")
	}

	// Process ParentTables and ChildTables recursively
	h.processXFilesRelations(&xfiles, options, "")
}

// processXFilesRelations processes ParentTables and ChildTables from XFiles
// and adds them as Preload options recursively
func (h *Handler) processXFilesRelations(xfiles *XFiles, options *ExtendedRequestOptions, basePath string) {
	if xfiles == nil {
		return
	}

	// Process ParentTables
	if len(xfiles.ParentTables) > 0 {
		logger.Debug("X-Files: Processing %d parent tables", len(xfiles.ParentTables))
		for _, parentTable := range xfiles.ParentTables {
			h.addXFilesPreload(parentTable, options, basePath)
		}
	}

	// Process ChildTables
	if len(xfiles.ChildTables) > 0 {
		logger.Debug("X-Files: Processing %d child tables", len(xfiles.ChildTables))
		for _, childTable := range xfiles.ChildTables {
			h.addXFilesPreload(childTable, options, basePath)
		}
	}
}

// resolveRelationNamesInOptions resolves all table names to field names in preload options
// This is called internally by parseOptionsFromHeaders when a model is provided
func (h *Handler) resolveRelationNamesInOptions(options *ExtendedRequestOptions, model interface{}) {
	if options == nil || model == nil {
		return
	}

	// Resolve relation names in all preload options
	for i := range options.Preload {
		preload := &options.Preload[i]

		// Split the relation path (e.g., "parent.child.grandchild")
		parts := strings.Split(preload.Relation, ".")
		resolvedParts := make([]string, 0, len(parts))

		// Resolve each part of the path
		currentModel := model
		for partIdx, part := range parts {
			isLast := partIdx == len(parts)-1
			var resolvedPart string
			if isLast {
				// For the final part, use join-key-aware resolution to disambiguate when
				// multiple fields point to the same type (e.g., HUB_RID_HUB vs HUB_RID_ASSIGNEDTO).
				// RelatedKey = parent's local column linking to child; ForeignKey = local column linking to parent.
				localKey := preload.RelatedKey
				if localKey == "" {
					localKey = preload.ForeignKey
				}
				resolvedPart = h.resolveRelationNameWithJoinKey(currentModel, part, localKey)
			} else {
				resolvedPart = h.resolveRelationName(currentModel, part)
			}
			resolvedParts = append(resolvedParts, resolvedPart)

			// Try to get the model type for the next level
			// This allows nested resolution
			if nextModel := reflection.GetRelationModel(currentModel, resolvedPart); nextModel != nil {
				currentModel = nextModel
			}
		}

		// Update the relation path with resolved names
		resolvedPath := strings.Join(resolvedParts, ".")
		if resolvedPath != preload.Relation {
			logger.Debug("Resolved relation path '%s' -> '%s'", preload.Relation, resolvedPath)
			preload.Relation = resolvedPath
		}
	}

	// Resolve relation names in expand options
	for i := range options.Expand {
		expand := &options.Expand[i]
		resolved := h.resolveRelationName(model, expand.Relation)
		if resolved != expand.Relation {
			logger.Debug("Resolved expand relation '%s' -> '%s'", expand.Relation, resolved)
			expand.Relation = resolved
		}
	}
}

// resolveRelationName resolves a relation name or table name to the actual field name in the model
// If the input is already a field name, it returns it as-is
// If the input is a table name, it looks up the corresponding relation field
func (h *Handler) resolveRelationName(model interface{}, nameOrTable string) string {
	if model == nil || nameOrTable == "" {
		return nameOrTable
	}

	modelType := reflect.TypeOf(model)
	if modelType == nil {
		return nameOrTable
	}

	// Dereference pointer if needed
	if modelType.Kind() == reflect.Pointer {
		modelType = modelType.Elem()
	}

	// Check again after dereferencing
	if modelType == nil {
		return nameOrTable
	}

	// Ensure it's a struct
	if modelType.Kind() != reflect.Struct {
		return nameOrTable
	}

	// First, check if the input matches a field name directly
	for i := 0; i < modelType.NumField(); i++ {
		field := modelType.Field(i)
		if field.Name == nameOrTable {
			// It's already a field name
			// logger.Debug("Input '%s' is a field name", nameOrTable)
			return nameOrTable
		}
	}

	// If not found as a field name, try to look it up as a table name
	normalizedInput := strings.ToLower(strings.ReplaceAll(nameOrTable, "_", ""))

	for i := 0; i < modelType.NumField(); i++ {
		field := modelType.Field(i)
		fieldType := field.Type

		// Check if it's a slice or pointer to a struct
		var targetType reflect.Type
		if fieldType.Kind() == reflect.Slice {
			targetType = fieldType.Elem()
		} else if fieldType.Kind() == reflect.Pointer {
			targetType = fieldType.Elem()
		}

		if targetType != nil {
			// Dereference pointer if the slice contains pointers
			if targetType.Kind() == reflect.Pointer {
				targetType = targetType.Elem()
			}

			// Check if it's a struct type
			if targetType.Kind() == reflect.Struct {
				// Get the type name and normalize it
				typeName := targetType.Name()

				// Extract the table name from type name
				// Patterns: ModelCoreMastertaskitem -> mastertaskitem
				//           ModelMastertaskitem -> mastertaskitem
				normalizedTypeName := strings.ToLower(typeName)

				// Remove common prefixes like "model", "modelcore", etc.
				normalizedTypeName = strings.TrimPrefix(normalizedTypeName, "modelcore")
				normalizedTypeName = strings.TrimPrefix(normalizedTypeName, "model")

				// Compare normalized names
				if normalizedTypeName == normalizedInput {
					logger.Debug("Resolved table name '%s' to field '%s' (type: %s)", nameOrTable, field.Name, typeName)
					return field.Name
				}
			}
		}
	}

	// If no match found, return the original input
	logger.Debug("No field found for '%s', using as-is", nameOrTable)
	return nameOrTable
}

// resolveRelationNameWithJoinKey resolves a relation name like resolveRelationName, but when
// multiple fields point to the same related type, uses localKey to pick the one whose bun join
// tag starts with "join:localKey=". Falls back to resolveRelationName if no key match is found.
func (h *Handler) resolveRelationNameWithJoinKey(model interface{}, nameOrTable string, localKey string) string {
	if localKey == "" {
		return h.resolveRelationName(model, nameOrTable)
	}

	modelType := reflect.TypeOf(model)
	if modelType == nil {
		return nameOrTable
	}
	if modelType.Kind() == reflect.Pointer {
		modelType = modelType.Elem()
	}
	if modelType == nil || modelType.Kind() != reflect.Struct {
		return nameOrTable
	}

	// If it's already a direct field name, return as-is (no ambiguity).
	for i := 0; i < modelType.NumField(); i++ {
		if modelType.Field(i).Name == nameOrTable {
			return nameOrTable
		}
	}

	normalizedInput := strings.ToLower(strings.ReplaceAll(nameOrTable, "_", ""))
	localKeyLower := strings.ToLower(localKey)

	// Find all fields whose related type matches nameOrTable, then pick the one
	// whose bun join tag local key matches localKey.
	var fallbackField string
	for i := 0; i < modelType.NumField(); i++ {
		field := modelType.Field(i)
		fieldType := field.Type

		var targetType reflect.Type
		if fieldType.Kind() == reflect.Slice {
			targetType = fieldType.Elem()
		} else if fieldType.Kind() == reflect.Pointer {
			targetType = fieldType.Elem()
		}
		if targetType != nil && targetType.Kind() == reflect.Pointer {
			targetType = targetType.Elem()
		}
		if targetType == nil || targetType.Kind() != reflect.Struct {
			continue
		}

		normalizedTypeName := strings.ToLower(targetType.Name())
		normalizedTypeName = strings.TrimPrefix(normalizedTypeName, "modelcore")
		normalizedTypeName = strings.TrimPrefix(normalizedTypeName, "model")
		if normalizedTypeName != normalizedInput {
			continue
		}

		// Type name matches; record as fallback.
		if fallbackField == "" {
			fallbackField = field.Name
		}

		// Check bun join tag: "join:localKey=foreignKey"
		bunTag := field.Tag.Get("bun")
		for _, tagPart := range strings.Split(bunTag, ",") {
			tagPart = strings.TrimSpace(tagPart)
			if !strings.HasPrefix(tagPart, "join:") {
				continue
			}
			joinSpec := strings.TrimPrefix(tagPart, "join:")
			// joinSpec can be "col1=col2" or "col1=col2 col3=col4" (multi-col joins)
			joinCols := strings.Fields(joinSpec)
			if len(joinCols) == 0 {
				joinCols = []string{joinSpec}
			}
			for _, joinCol := range joinCols {
				eqIdx := strings.Index(joinCol, "=")
				if eqIdx < 0 {
					continue
				}
				joinLocalKey := strings.ToLower(joinCol[:eqIdx])
				if joinLocalKey == localKeyLower {
					logger.Debug("Resolved '%s' (localKey: %s) -> field '%s'", nameOrTable, localKey, field.Name)
					return field.Name
				}
			}
		}
	}

	if fallbackField != "" {
		logger.Debug("No join key match for '%s' (localKey: %s), using first type match: '%s'", nameOrTable, localKey, fallbackField)
		return fallbackField
	}
	return h.resolveRelationName(model, nameOrTable)
}

// addXFilesPreload converts an XFiles relation into a PreloadOption
// and recursively processes its children
func (h *Handler) addXFilesPreload(xfile *XFiles, options *ExtendedRequestOptions, basePath string) {
	if xfile == nil || xfile.TableName == "" {
		return
	}

	// Use the Prefix (e.g., "MAL") as the relation name, which matches the Go struct field name
	// Fall back to TableName if Prefix is not specified
	relationName := xfile.Prefix
	if relationName == "" {
		relationName = xfile.TableName
	}

	// SPECIAL CASE: For recursive child tables, generate FK-based relation name
	// Example: If prefix is "MAL" and relatedkey is "rid_parentmastertaskitem",
	// the actual struct field is "MAL_RID_PARENTMASTERTASKITEM", not "MAL"
	if xfile.Recursive && xfile.RelatedKey != "" && basePath != "" {
		// Check if this is a self-referencing recursive relation (same table as parent)
		// by comparing the last part of basePath with the current prefix
		basePathParts := strings.Split(basePath, ".")
		lastPrefix := basePathParts[len(basePathParts)-1]

		if lastPrefix == relationName {
			// This is a recursive self-reference, use FK-based name
			fkUpper := strings.ToUpper(xfile.RelatedKey)
			relationName = relationName + "_" + fkUpper
			logger.Debug("X-Files: Generated FK-based relation name for recursive table: %s", relationName)
		}
	}

	relationPath := relationName
	if basePath != "" {
		relationPath = basePath + "." + relationName
	}

	logger.Debug("X-Files: Adding preload for relation: %s", relationPath)

	// Create PreloadOption from XFiles configuration
	preloadOpt := common.PreloadOption{
		Relation:    relationPath,
		TableName:   xfile.TableName, // Store the actual database table name for WHERE clause processing
		Columns:     xfile.Columns,
		OmitColumns: xfile.OmitColumns,
	}

	// Add sorting if specified
	if len(xfile.Sort) > 0 {
		preloadOpt.Sort = make([]common.SortOption, 0, len(xfile.Sort))
		for _, sortField := range xfile.Sort {
			direction := "ASC"
			colName := sortField

			// Handle direction prefixes
			if strings.HasPrefix(sortField, "-") {
				direction = "DESC"
				colName = strings.TrimPrefix(sortField, "-")
			} else if strings.HasPrefix(sortField, "+") {
				colName = strings.TrimPrefix(sortField, "+")
			}

			preloadOpt.Sort = append(preloadOpt.Sort, common.SortOption{
				Column:    strings.TrimSpace(colName),
				Direction: direction,
			})
		}
	}

	// Add filters if specified
	if len(xfile.FilterFields) > 0 {
		preloadOpt.Filters = make([]common.FilterOption, 0, len(xfile.FilterFields))
		for _, filterField := range xfile.FilterFields {
			preloadOpt.Filters = append(preloadOpt.Filters, common.FilterOption{
				Column:        filterField.Field,
				Operator:      filterField.Operator,
				Value:         filterField.Value,
				LogicOperator: "AND",
			})
		}
	}

	// Transfer SqlJoins from XFiles to PreloadOption first, so aliases are available for WHERE sanitization
	if len(xfile.SqlJoins) > 0 {
		preloadOpt.SqlJoins = make([]string, 0, len(xfile.SqlJoins))
		preloadOpt.JoinAliases = make([]string, 0, len(xfile.SqlJoins))

		for _, joinClause := range xfile.SqlJoins {
			// Sanitize the join clause
			sanitizedJoin := common.SanitizeWhereClause(joinClause, "", nil)
			if sanitizedJoin == "" {
				logger.Warn("X-Files: SqlJoin failed sanitization for %s: %s", relationPath, joinClause)
				continue
			}

			preloadOpt.SqlJoins = append(preloadOpt.SqlJoins, sanitizedJoin)

			// Extract join alias for validation
			alias := extractJoinAlias(sanitizedJoin)
			if alias != "" {
				preloadOpt.JoinAliases = append(preloadOpt.JoinAliases, alias)
				logger.Debug("X-Files: Extracted join alias for %s: %s", relationPath, alias)
			}
		}

		logger.Debug("X-Files: Added %d SQL joins to preload %s", len(preloadOpt.SqlJoins), relationPath)
	}

	// Add WHERE clause if SQL conditions specified
	// SqlJoins must be processed first so join aliases are known and not incorrectly replaced
	whereConditions := make([]string, 0)
	if len(xfile.SqlAnd) > 0 {
		var sqlAndOpts *common.RequestOptions
		if len(preloadOpt.JoinAliases) > 0 {
			sqlAndOpts = &common.RequestOptions{JoinAliases: preloadOpt.JoinAliases}
		}
		for _, sqlCond := range xfile.SqlAnd {
			sanitizedCond := common.SanitizeWhereClause(sqlCond, xfile.TableName, sqlAndOpts)
			if sanitizedCond != "" {
				whereConditions = append(whereConditions, sanitizedCond)
			}
		}
	}
	if len(whereConditions) > 0 {
		preloadOpt.Where = strings.Join(whereConditions, " AND ")
	}

	// Add limit if specified
	if limitStr := xfile.Limit.String(); limitStr != "" && limitStr != "0" {
		if limitVal, err := xfile.Limit.Int64(); err == nil && limitVal > 0 {
			limit := int(limitVal)
			preloadOpt.Limit = &limit
		}
	}

	// Add computed columns (CQL) -> ComputedQL
	if len(xfile.CQLColumns) > 0 {
		preloadOpt.ComputedQL = make(map[string]string)
		for i, cqlExpr := range xfile.CQLColumns {
			colName := fmt.Sprintf("cql%d", i+1)
			preloadOpt.ComputedQL[colName] = cqlExpr
			logger.Debug("X-Files: Added computed column %s to preload %s: %s", colName, relationPath, cqlExpr)
		}
	}

	// Set recursive flag
	preloadOpt.Recursive = xfile.Recursive

	// Extract relationship keys for proper foreign key filtering
	if xfile.PrimaryKey != "" {
		preloadOpt.PrimaryKey = xfile.PrimaryKey
		logger.Debug("X-Files: Set primary key for %s: %s", relationPath, xfile.PrimaryKey)
	}
	if xfile.RelatedKey != "" {
		preloadOpt.RelatedKey = xfile.RelatedKey
		logger.Debug("X-Files: Set related key for %s: %s", relationPath, xfile.RelatedKey)
	}
	if xfile.ForeignKey != "" {
		preloadOpt.ForeignKey = xfile.ForeignKey
		logger.Debug("X-Files: Set foreign key for %s: %s", relationPath, xfile.ForeignKey)
	}

	// Check if this table has a recursive child - if so, mark THIS preload as recursive
	// and store the recursive child's RelatedKey for recursion generation
	hasRecursiveChild := false
	if len(xfile.ChildTables) > 0 {
		for _, childTable := range xfile.ChildTables {
			if childTable.Recursive && childTable.TableName == xfile.TableName {
				hasRecursiveChild = true
				preloadOpt.Recursive = true
				preloadOpt.RecursiveChildKey = childTable.RelatedKey
				logger.Debug("X-Files: Detected recursive child for %s, marking parent as recursive (recursive FK: %s)",
					relationPath, childTable.RelatedKey)
				break
			}
		}
	}

	// Skip adding this preload if it's a recursive child (it will be handled by parent's Recursive flag)
	if xfile.Recursive && basePath != "" {
		logger.Debug("X-Files: Skipping recursive child preload: %s (will be handled by parent)", relationPath)
		// Still process its parent/child tables for relations like DEF
		h.processXFilesRelations(xfile, options, relationPath)
		return
	}

	// Add the preload option
	options.Preload = append(options.Preload, preloadOpt)
	logger.Debug("X-Files: Added preload [%d]: Relation=%s, Recursive=%v, RelatedKey=%s, RecursiveChildKey=%s, Where=%s",
		len(options.Preload)-1, preloadOpt.Relation, preloadOpt.Recursive, preloadOpt.RelatedKey, preloadOpt.RecursiveChildKey, preloadOpt.Where)

	// Recursively process nested ParentTables and ChildTables
	// Skip processing child tables if we already detected and handled a recursive child
	if hasRecursiveChild {
		logger.Debug("X-Files: Skipping child table processing for %s (recursive child already handled)", relationPath)
		// But still process parent tables
		if len(xfile.ParentTables) > 0 {
			logger.Debug("X-Files: Processing %d parent tables for %s", len(xfile.ParentTables), relationPath)
			for _, parentTable := range xfile.ParentTables {
				h.addXFilesPreload(parentTable, options, relationPath)
			}
		}
	} else if len(xfile.ParentTables) > 0 || len(xfile.ChildTables) > 0 {
		h.processXFilesRelations(xfile, options, relationPath)
	}
}

// ColumnCastInfo holds information about whether a column needs casting
type ColumnCastInfo struct {
	NeedsCast     bool
	IsNumericType bool
}

// ValidateAndAdjustFilterForColumnType validates and adjusts a filter based on column type
// Returns ColumnCastInfo indicating whether the column should be cast to text in SQL
func (h *Handler) ValidateAndAdjustFilterForColumnType(filter *common.FilterOption, model interface{}) ColumnCastInfo {
	if filter == nil || model == nil {
		return ColumnCastInfo{NeedsCast: false, IsNumericType: false}
	}

	// Never cast geometry/geography or pgvector columns to TEXT — spatial and
	// vector operators need the native column type. Also bypass when the
	// operator itself is spatial/vector (e.g. st_dwithin, l2_within).
	if common.IsSpatialOperator(filter.Operator) || common.IsVectorOperator(filter.Operator) ||
		reflection.IsSpatialColumn(model, filter.Column) || reflection.IsVectorColumn(model, filter.Column) {
		return ColumnCastInfo{NeedsCast: false, IsNumericType: false}
	}

	// Never cast citext columns to TEXT: CAST(col AS TEXT) swaps in case-sensitive
	// comparison semantics and prevents PostgreSQL from using a citext index.
	if reflection.IsCitextColumn(model, filter.Column) {
		return ColumnCastInfo{NeedsCast: false, IsNumericType: false}
	}

	colType := reflection.GetColumnTypeFromModel(model, filter.Column)
	if colType == reflect.Invalid {
		// Column not found in model, no casting needed
		logger.Debug("Column %s not found in model, skipping type validation", filter.Column)
		return ColumnCastInfo{NeedsCast: false, IsNumericType: false}
	}

	// LIKE/ILIKE always compare against text, wildcards and all. Never coerce
	// the value to the column's native numeric/bool/time type here: doing so
	// strips the '%' wildcards and hands the driver a non-string argument,
	// which fails with "operator does not exist: text ~~* integer" once the
	// column is cast to TEXT below.
	if op := strings.ToLower(filter.Operator); op == "like" || op == "ilike" || op == "not_ilike" {
		if reflection.IsStringType(colType) {
			return ColumnCastInfo{NeedsCast: false, IsNumericType: false}
		}
		return ColumnCastInfo{NeedsCast: true, IsNumericType: reflection.IsNumericType(colType)}
	}

	// Check if the input value is numeric
	valueIsNumeric := false
	if strVal, ok := filter.Value.(string); ok {
		strVal = strings.Trim(strVal, "%")
		valueIsNumeric = reflection.IsNumericValue(strVal)
	}

	// Adjust based on column type
	switch {
	case reflection.IsNumericType(colType):
		// Column is numeric
		if valueIsNumeric {
			// Value is numeric - try to convert it
			if strVal, ok := filter.Value.(string); ok {
				strVal = strings.Trim(strVal, "%")
				numericVal, err := reflection.ConvertToNumericType(strVal, colType)
				if err != nil {
					logger.Debug("Failed to convert value '%s' to numeric type for column %s, will use text cast", strVal, filter.Column)
					return ColumnCastInfo{NeedsCast: true, IsNumericType: true}
				}
				filter.Value = numericVal
			}
			// No cast needed - numeric column with numeric value
			return ColumnCastInfo{NeedsCast: false, IsNumericType: true}
		} else {
			// Value is not numeric - cast column to text for comparison
			logger.Debug("Non-numeric value for numeric column %s, will cast to text", filter.Column)
			return ColumnCastInfo{NeedsCast: true, IsNumericType: true}
		}

	case reflection.IsStringType(colType):
		// String columns don't need casting
		return ColumnCastInfo{NeedsCast: false, IsNumericType: false}

	default:
		// For bool, time.Time, and other complex types - cast to text
		logger.Debug("Complex type column %s, will cast to text", filter.Column)
		return ColumnCastInfo{NeedsCast: true, IsNumericType: false}
	}
}