package analytics

import (
	"errors"
	"fmt"
	"strings"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/formula/spec"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
)

const (
	// AggMarkerPrefix is the delimiter used for aggregate placeholder markers in compiled SQL.
	// Format: ##AGG##<FUNC>##<ChildObject>##<ChildField>##<ForeignKey>##
	AggMarkerPrefix = "##AGG##"

	// DefaultDecimalPrecision is the DECIMAL precision used for VALUE() casts.
	DefaultDecimalPrecision = 18
	// DefaultDecimalScale is the DECIMAL scale used for VALUE() casts.
	DefaultDecimalScale = 6
)

// CompileContext provides the compiler with table alias mappings for
// cross-table references and aggregate subqueries.
type CompileContext struct {
	// TableAlias is the alias for the current (parent) table, e.g. "t"
	TableAlias string

	// JoinAliases maps relationship names to SQL aliases.
	// e.g. {"Parent__r": "p", "Owner": "own", "Parent": "par"}
	JoinAliases map[string]string

	// CrossRefFieldAliases maps logical cross-ref fields to the physical field
	// that should be selected from the joined table. This keeps formulas using
	// Salesforce's logical Name field working for objects whose authoritative
	// name field is represented by metadata as another column.
	//
	// Key: "Parent__r.Name" -> Value: "Display_Label__c"
	CrossRefFieldAliases map[string]string

	// CrossRefFieldExpressions maps a logical relationship field to a complete,
	// metadata-bound SQL read expression. It is used for virtual compound fields
	// whose value has no authoritative physical column. Expressions are built by
	// FormulaViewService from the same immutable schema as the relationship JOIN;
	// the parser/compiler never accepts caller-provided SQL here.
	//
	// Key: "Lead.Name" -> Value: "NULLIF(TRIM(CONCAT_WS(...)), '')"
	CrossRefFieldExpressions map[string]string

	// QualifyFields: if true, prefix bare FieldRef with TableAlias.
	QualifyFields bool

	// DeepJoinMap maps multi-hop cross-table refs to resolved SQL expressions.
	// Key: "Owner.Profile.Name" → Value: "dj_prof.`Name`"
	// Populated from resolved DeepJoinChain data by the ViewBuilder.
	DeepJoinMap map[string]string

	// FieldTypes maps identifier paths to their Salesforce field type so the
	// compiler can emit type-aware SQL (e.g. DATEDIFF for date - date instead
	// of the YYYYMMDD numeric subtraction MySQL would perform by default).
	//
	// Keys:
	//   - bare field: "End_Date__c"
	//   - cross-ref: "Parent__r.Start_Date__c"
	//   - deep ref:  "Owner.Profile.BirthDate"
	//
	// Values are lowercased Salesforce types: "date", "datetime", "currency",
	// "double", "percent", "int", "string", "picklist", "boolean", etc.
	// Missing keys are treated as unknown and fall back to type-naive emission.
	FieldTypes map[string]string

	// AllowedLookupValues seals LOOKUPVALUE to standard metadata projections
	// that explicitly opt in. Admin-authored formula fields compile with this
	// empty, so raw table/field lookup is not exposed as a generic primitive.
	AllowedLookupValues []LookupValueRule

	// BaseFieldPhysicalMap maps logical field API names to physical DB column
	// names for base (non-formula) fields on the primary object. Set only in
	// L0 and single-layer compile contexts where formulas read directly from the
	// physical table; outer layers (L1+) leave this nil so bare FieldRef nodes
	// resolve to the logical alias exposed by the inner projection.
	BaseFieldPhysicalMap map[string]string

	// FunctionExecutables is the active executable formula-function bundle
	// registry. App-owned functions use this single read path for arity,
	// return type, and formula_expr_v1 SQL lowering.
	FunctionExecutables *formula.ExecutableFunctionRegistry

	// FunctionRegistry is the arity/return-type catalog active for this compile.
	// Nil preserves the L0 default catalog. Prefer FunctionExecutables for
	// source-backed formula_function_bundle functions so SQL lowering can read
	// the same executable definition as runtime evaluation.
	FunctionRegistry *spec.Registry

	// BoundIdentifierSQL maps formula_expr_v1 evaluator parameter names to the
	// already-compiled SQL argument expressions for executable bundle lowering.
	BoundIdentifierSQL map[string]string

	// fenceNestedDecimalDivision is enabled only while lowering a division
	// subtree that contains another division. TiDB otherwise lets the inferred
	// DECIMAL precision grow across the nested quotients until a surrounding
	// additive operation fails with Error 1265.
	fenceNestedDecimalDivision bool

	// BoundIdentifierTypes carries optional formula value types for bound
	// evaluator parameters so nested helper lowering remains type-aware.
	BoundIdentifierTypes map[string]string

	// arithNullSites / blankNullSites count, during a single compile, the two
	// distinguishable origins of SQL NULL: undefined arithmetic (formula-domain
	// #Error) and everything else (a blank stored value, an unmatched join, an
	// unmodelled function). compileIf/compileCase take a delta window around
	// the condition and propagate NULL strictly only when the window contains
	// arithmetic and nothing else that could be NULL. See
	// formula_dsl_null_provenance.go (null provenance).
	//
	// POINTER TYPE IS LOAD-BEARING. CompileContext is copied by value in three
	// places — the nested-division fence (`cloned := *ctx` below),
	// formula_dsl_function_helpers.go (`next := *ctx`, executable bundle
	// bodies) and formula_dsl_viewbuilder.go (`next := *base`, per-formula
	// lookupValue rules). The copies must SHARE the counters, otherwise marks
	// made inside a bundle body or a fenced subtree are invisible to the
	// enclosing IF and the gate silently mis-fires.
	arithNullSites *int
	blankNullSites *int

	// compiledSQLByteLimit and formulaSQLScope are internal compiler controls.
	// Compile always seeds the production limit; tests may provide a smaller
	// positive limit to exercise rejection without constructing huge formulas.
	// Context copies retain both values, so nested fences and executable formula
	// bodies cannot bypass the bound.
	compiledSQLByteLimit int
	formulaSQLScope      string

	// divisionAnalysis is a per-Compile bottom-up summary used to decide whether
	// a quotient needs the nested DECIMAL fence before either operand is emitted.
	// Context copies share it so fenced subtrees and formula-function executable
	// bodies stay on the same linear compile path.
	divisionAnalysis *formulaDivisionAnalysis

	// compileWork is an optional package-private test probe. It is intentionally
	// not seeded in production; when present, context copies share the counters.
	compileWork *formulaCompileWork
}

var ErrRestrictedFormulaPrimitive = errors.New("restricted formula primitive")

// LookupValueRule is an allowlist entry for the internal LOOKUPVALUE
// projection helper. The rule is intentionally exact: target object, output
// field, match fields, and source fields must all match in order.
type LookupValueRule struct {
	TargetTable    string
	OutputField    string
	MatchPairs     []LookupValueMatchRule
	ExcludeDeleted bool
	// HoistedJoinAlias is non-empty when this lookupValue is emitted as a
	// pre-registered LEFT JOIN instead of a correlated scalar subquery.
	// compileLookupValue then emits <alias>.<outputField> directly.
	HoistedJoinAlias string
}

type LookupValueMatchRule struct {
	MatchField  string
	SourceField string
}

// Compile translates an AST node into a TiDB-compatible SQL expression.
func Compile(node Node, ctx *CompileContext) (string, error) {
	if ctx == nil {
		ctx = &CompileContext{TableAlias: "t", JoinAliases: map[string]string{}}
	}
	// Compile is the sole exported entry point, so allocating the NULL-provenance
	// counters here is what guarantees no caller can construct a gate-less
	// context by hand (null provenance). Callers that build a CompileContext literal leave
	// both nil; the mark helpers are nil-safe, but a nil counter would mean the
	// gate never fires, so seed them once here instead.
	if ctx.arithNullSites == nil {
		ctx.arithNullSites = new(int)
	}
	if ctx.blankNullSites == nil {
		ctx.blankNullSites = new(int)
	}
	if ctx.compiledSQLByteLimit <= 0 {
		ctx.compiledSQLByteLimit = maxCompiledFormulaSQLBytes
	}
	// A CompileContext may be reused by a caller. Keep AST pointers scoped to one
	// top-level compilation while allowing dynamically parsed executable bodies
	// to join the same analysis below.
	ctx.divisionAnalysis = newFormulaDivisionAnalysis()
	return compile(node, ctx)
}

func activeFunctionRegistry(ctx *CompileContext) *spec.Registry {
	if ctx != nil && ctx.FunctionExecutables != nil {
		return ctx.FunctionExecutables.SpecRegistry()
	}
	if ctx != nil && ctx.FunctionRegistry != nil {
		return ctx.FunctionRegistry
	}
	return spec.DefaultRegistry()
}

func activeFunctionExecutables(ctx *CompileContext) *formula.ExecutableFunctionRegistry {
	if ctx != nil && ctx.FunctionExecutables != nil {
		return ctx.FunctionExecutables
	}
	return nil
}

func checkFunctionArity(ctx *CompileContext, name string, argc int) error {
	return activeFunctionRegistry(ctx).CheckArity(name, argc)
}

func compile(node Node, ctx *CompileContext) (string, error) {
	if ctx.compileWork != nil {
		ctx.compileWork.compiledNodes++
	}
	ctx.divisionAnalysis.ensure(node, ctx.compileWork)
	sql, err := compileNode(node, ctx)
	if err != nil {
		return "", err
	}
	if err := checkFormulaSQLOutputSize(ctx.formulaSQLScope, len(sql), ctx.compiledSQLByteLimit); err != nil {
		return "", err
	}
	return sql, nil
}

func compileNode(node Node, ctx *CompileContext) (string, error) {
	switch n := node.(type) {
	case *NumberLit:
		return n.Value, nil

	case *StringLit:
		return "'" + utils.EscapeSQLString(n.Value) + "'", nil

	case *BoolLit:
		if n.Value {
			return "1", nil
		}
		return "0", nil

	case *NullLit:
		// A literal NULL in formula text is an authored BLANK, not #Error.
		ctx.markBlankNullSite()
		return "NULL", nil

	case *FieldRef:
		if ctx != nil {
			if bound, ok := lookupCaseInsensitive(ctx.BoundIdentifierSQL, n.Name); ok {
				// The SQL is the caller's, already compiled in another context;
				// its NULL provenance is unknown here, so treat it as blank.
				ctx.markBlankNullSite()
				return bound, nil
			}
		}
		colName := n.Name
		if phys, ok := ctx.BaseFieldPhysicalMap[colName]; ok && phys != "" {
			colName = phys
		}
		fieldSQL := fmt.Sprintf("`%s`", colName)
		if ctx.QualifyFields && ctx.TableAlias != "" {
			fieldSQL = fmt.Sprintf("%s.`%s`", ctx.TableAlias, colName)
		}
		return fieldReadSQL(fieldSQL, lookupFieldType(ctx, n.Name), ctx), nil

	case *CrossRef:
		alias, ok := ctx.JoinAliases[n.Object]
		if !ok {
			// Try case-insensitive lookup
			for k, v := range ctx.JoinAliases {
				if strings.EqualFold(k, n.Object) {
					alias = v
					ok = true
					break
				}
			}
		}
		if !ok {
			return "", fmt.Errorf("unresolved cross-reference %s.%s: missing metadata join", n.Object, n.Field)
		}
		if expression, found := ctx.resolveCrossRefFieldExpression(n.Object, n.Field); found {
			return toFormulaDomainSQL(expression, lookupFieldType(ctx, n.Object+"."+n.Field)), nil
		}
		fieldSQL := fmt.Sprintf("%s.`%s`", alias, ctx.resolveCrossRefField(n.Object, n.Field))
		return fieldReadSQL(fieldSQL, lookupFieldType(ctx, n.Object+"."+n.Field), ctx), nil

	case *DeepCrossRef:
		// Check if this deep ref was resolved to a chain JOIN
		if ctx.DeepJoinMap != nil {
			key := strings.Join(n.Parts, ".")
			if resolved, ok := lookupCaseInsensitive(ctx.DeepJoinMap, key); ok {
				return fieldReadSQL(resolved, lookupFieldType(ctx, key), ctx), nil
			}
		}
		return "", fmt.Errorf("unresolved deep cross-reference %s: missing metadata join chain", strings.Join(n.Parts, "."))

	case *BinaryOp:
		return compileBinaryOp(n, ctx)

	case *UnaryOp:
		return compileUnaryOp(n, ctx)

	case *LogicalOp:
		return compileLogicalOp(n, ctx)

	case *FuncCall:
		// Deliberately blunt (null provenance): EVERY function call marks a blank-NULL
		// site, including ones that provably cannot yield NULL (TEXT, ISBLANK,
		// ROUND, ABS). This fails closed to today's emission and means a newly
		// registered function or L1 bundle passthrough DISABLES the strictness
		// gate by default rather than silently widening it. Do not add an
		// allowlist — the cost is zero for null provenance (its gated conditions contain
		// no function calls) and the failure mode of getting the allowlist
		// wrong is a silently blanked production formula.
		ctx.markBlankNullSite()
		return compileFuncCall(n, ctx)

	case *IfExpr:
		return compileIf(n, ctx)

	case *CaseExpr:
		// A CASE with no else-arm MANUFACTURES NULL (null provenance). The parser accepts
		// an even-arity CASE(expr, v1, r1, v2, r2) and leaves Default nil, and
		// compileCase then emits `CASE x WHEN .. THEN .. END` with no ELSE,
		// which SQL evaluates to NULL when no branch matches. That NULL is not
		// traceable to any operand — every leaf inside can be blank-safe — so
		// without this mark the node is transparent and an enclosing condition
		// containing it plus a division is misread as pure #Error. Measured: a
		// TOTAL division (Den__c = 100) plus a no-match CASE flipped a row from
		// 222 to blank.
		//
		// It is marked blank rather than arith because SQL's no-match NULL is
		// the absence of a value, not undefined arithmetic. Persisted pack formulas
		// require else_result, so no pack-ported formula can
		// carry this shape, but admin-authored NexusCRM formulas go through the
		// same parser. The mark is made here rather than inside compileCase so
		// that it lands in the delta window of whatever CONDITION encloses the
		// CASE, which is the window that still decides strictness.
		if n.Default == nil {
			ctx.markBlankNullSite()
		}
		return compileCase(n, ctx)

	case *AggregateExpr:
		// A rollup subquery returns NULL when the child set is empty, which is
		// blankness, not #Error.
		ctx.markBlankNullSite()
		return compileAggregate(n, ctx)

	default:
		return "", fmt.Errorf("unsupported AST node type: %T", node)
	}
}

func (ctx *CompileContext) resolveCrossRefFieldExpression(objectName, fieldName string) (string, bool) {
	if ctx == nil || len(ctx.CrossRefFieldExpressions) == 0 {
		return "", false
	}
	return lookupCaseInsensitive(ctx.CrossRefFieldExpressions, objectName+"."+fieldName)
}

func (ctx *CompileContext) resolveCrossRefField(objectName, fieldName string) string {
	if ctx == nil || len(ctx.CrossRefFieldAliases) == 0 {
		return fieldName
	}
	key := objectName + "." + fieldName
	if mapped, ok := ctx.CrossRefFieldAliases[key]; ok && mapped != "" {
		return mapped
	}
	for k, mapped := range ctx.CrossRefFieldAliases {
		if strings.EqualFold(k, key) && mapped != "" {
			return mapped
		}
	}
	return fieldName
}

func compileBinaryOp(n *BinaryOp, ctx *CompileContext) (string, error) {
	compileCtx := ctx
	if n.Op == "/" && !ctx.fenceNestedDecimalDivision && ctx.divisionAnalysis.containsDivision(n.Left, n.Right) {
		// Select the fenced context before compiling either operand. The previous
		// path compiled both operands, scanned their subtrees, discarded the SQL,
		// and recursively compiled the quotient again. A shared bottom-up summary
		// makes the same decision once and keeps AST emission single-pass.
		cloned := *ctx
		cloned.fenceNestedDecimalDivision = true
		compileCtx = &cloned
	}

	left, err := compile(n.Left, compileCtx)
	if err != nil {
		return "", err
	}
	right, err := compile(n.Right, compileCtx)
	if err != nil {
		return "", err
	}

	// Both blank-comparison lowerings below replace the whole node with a TOTAL
	// boolean — `X IS NULL`, `IFNULL(X = '', TRUE)`, or their negations — which
	// can never itself evaluate to SQL NULL. So whatever arithmetic sits inside
	// X cannot reach the enclosing branch boundary through this node, and any
	// arith mark the operands already made (they are compiled above, before
	// these special cases get a chance to rewrite) no longer describes the
	// emitted expression. Mark blank to retract it (null provenance).
	//
	// Without this, `IF(Num__c/Den__c = "", 1, 2)` acquired a strict guard that
	// TiDB can never take (`((0/0) IS NULL) IS NULL` -> 0), while the
	// semantically identical `IF(Num__c/Den__c = NULL, 1, 2)` did not — dead SQL
	// plus a duplicated condition, and two identical lowerings behaving
	// differently. Using the blank mark as a gate-closer here is the same
	// fail-closed idiom as the blanket *FuncCall rule: it can only ever restore
	// today's emission, never widen strictness.
	//
	// SCOPE OF THE RETRACTION, stated precisely because it is wider than this
	// node: the gate's blank test is `blankAfter == blankBefore` over the WHOLE
	// enclosing condition, so a mark made here closes the gate for every
	// division in that condition, not only for the one inside the rewritten
	// operand. Measured collateral:
	//
	//	IF(AND(Num__c/Den__c > 0.5, Other__c = ""), 1, 2)   absorbs
	//	IF(AND(A__c/B__c = "", C__c/D__c > 0.5), 1, 2)      absorbs
	//
	// Both contain an #Error-capable division that is unrelated to the `= ""`
	// node and still lose the guard. That is deliberate, not an oversight. A
	// node-scoped alternative exists (snapshot the arith counter at the top of
	// compileBinaryOp and rewind it here instead of marking blank), and it was
	// rejected: it would make `= ""` permit strictness while `= NULL` still
	// closes the gate through its own *NullLit blank mark, reintroducing exactly
	// the two-spellings-two-behaviors asymmetry this retraction removes. The
	// current form is symmetric, fail-closed, and has zero merged-pack impact —
	// no pack formula has any division in an IF/CASE condition other than the
	// one null provenance fixes. Pinned by the two rows above in
	// TestCompileGateRefusesWhenConditionHasBlankCapableLeaf.
	if sql, ok := compileSalesforceEmptyStringComparison(n, left, right, compileCtx); ok {
		compileCtx.markBlankNullSite()
		return sql, nil
	}

	if sql, ok := compileNullComparison(n, left, right); ok {
		compileCtx.markBlankNullSite()
		return sql, nil
	}

	switch n.Op {
	case "&":
		// String concatenation → CONCAT
		return fmt.Sprintf("CONCAT(%s, %s)", left, right), nil
	case "^":
		// Exponent → POWER. Marked as an undefined-arithmetic site to match the
		// division rule (null provenance), but MEASURED CAVEAT: on TiDB v8.5.0 an
		// undefined POWER does NOT return NULL — it aborts the statement:
		//
		//	SELECT POWER(-2, 0.5);  -- ERROR 1690 (22003): DOUBLE value is out of range
		//	SELECT POWER(0, -1);    -- ERROR 1690 (22003)
		//	SELECT POWER(NULL, 2);  -- NULL (ordinary NULL propagation, not #Error)
		//
		// So the strict guard this mark enables is currently DEAD CODE for `^`:
		// the condition never evaluates to NULL via POWER, it kills the whole
		// view SELECT first. The mark is kept anyway because it is the
		// conservative direction (it can only add an unreachable guard, never
		// remove one) and because the engine could gain NULL-on-overflow
		// semantics later. Making an undefined POWER yield the formula-domain
		// #Error rather than a statement error is a separate, larger problem —
		// zero merged-pack formulas use `^`, so it is not solved here.
		compileCtx.markArithNullSite()
		return fmt.Sprintf("POWER(%s, %s)", left, right), nil
	case "<>":
		return fmt.Sprintf("(%s != %s)", left, right), nil
	case "-":
		// Salesforce semantics for date arithmetic diverge from MySQL:
		//   date1 - date2       → number of days (integer)
		//   date - number       → date shifted back by N days
		// MySQL, given raw `-`, coerces date values to YYYYMMDD integers
		// and returns the arithmetic difference (e.g. 20240330 - 20230531
		// = 9799 instead of 304), which silently corrupts every formula
		// that does date subtraction (Term__c, Closed_Days__c, ValidDays__c).
		// Detect the types and emit DATEDIFF / DATE_SUB instead.
		if isDateLike(n.Left, compileCtx) && isDateLike(n.Right, compileCtx) {
			// Salesforce: datetime - datetime returns elapsed time in fractional
			// days (6h = 0.25), date - date returns whole days. DATEDIFF drops
			// the time component (integer days only), so datetime operands need
			// TIMESTAMPDIFF. TIMESTAMPDIFF(SECOND, e1, e2) = e2 - e1, so pass
			// (right, left) to match the formula's left - right semantics.
			if isDatetimeLike(n.Left, compileCtx) && isDatetimeLike(n.Right, compileCtx) {
				return fmt.Sprintf("(TIMESTAMPDIFF(SECOND, %s, %s) / 86400.0)", right, left), nil
			}
			return fmt.Sprintf("DATEDIFF(%s, %s)", left, right), nil
		}
		if isDateLike(n.Left, compileCtx) && isNumeric(n.Right, compileCtx) {
			return fmt.Sprintf("DATE_SUB(%s, INTERVAL (%s) DAY)", left, right), nil
		}
		return fmt.Sprintf("(%s %s %s)", left, n.Op, right), nil
	case "+":
		// Salesforce: date + number → date shifted forward by N days.
		if isDateLike(n.Left, compileCtx) && isNumeric(n.Right, compileCtx) {
			return fmt.Sprintf("DATE_ADD(%s, INTERVAL (%s) DAY)", left, right), nil
		}
		if isNumeric(n.Left, compileCtx) && isDateLike(n.Right, compileCtx) {
			return fmt.Sprintf("DATE_ADD(%s, INTERVAL (%s) DAY)", right, left), nil
		}
		return fmt.Sprintf("(%s %s %s)", left, n.Op, right), nil
	case "/":
		// x/0 is undefined. TiDB/MySQL render it as NULL, which a plain
		// branch-selecting CASE then absorbs as "false" — the defect this file addresses.
		// Mark the site unless the divisor is a non-zero number literal and the
		// quotient therefore cannot be undefined (null provenance).
		if compileCtx.fenceNestedDecimalDivision {
			decimalType, err := persistence.ResolveSQLType(string(constants.FieldTypeDouble))
			if err != nil {
				return "", fmt.Errorf("resolve formula decimal SQL type: %w", err)
			}
			if !definedDivisor(n.Right) {
				compileCtx.markArithNullSite()
			}
			return fmt.Sprintf(
				"(CAST(%s AS %s) / CAST(%s AS %s))",
				left,
				decimalType,
				right,
				decimalType,
			), nil
		}
		if !definedDivisor(n.Right) {
			compileCtx.markArithNullSite()
		}
		return fmt.Sprintf("(%s %s %s)", left, n.Op, right), nil
	default:
		return fmt.Sprintf("(%s %s %s)", left, n.Op, right), nil
	}
}

// nullCoalescingFuncs are the functions whose result type is the type-merge of
// their operands (a NULL operand is absorbed): the result is date/datetime-like
// iff every non-null operand is. BLANKVALUE/NULLVALUE are cataloged
// (ReturnMergeArgs); IFNULL/COALESCE/NULLIF are DSL-passthrough and absent from
// the catalog. classifyBlankOperand, isDateLike, and isDatetimeLike all key off
// this single set so a new coalescer is handled everywhere by one edit.
var nullCoalescingFuncs = map[string]bool{
	"BLANKVALUE": true,
	"NULLVALUE":  true,
	"IFNULL":     true,
	"COALESCE":   true,
	"NULLIF":     true,
}

// isDateLike reports whether a node evaluates to a date/datetime value. For
// function calls the date/datetime return type is read from the shared catalog
// (spec) for cataloged functions — TODAY/DATE/DATEVALUE/__DATE_ADD_MONTHS__/
// __DATE_ADD_MONTHS_TARGET_CLAMP__/__DATE_DIFF_DAYS__/DATE_ADD
// return date and NOW/DATETIMEVALUE return datetime per VS2 — plus the active
// DSL-only DATE_SUB helper, which has no catalog entry. Field references fall
// back to the FieldTypes map populated by the ViewBuilder from the live schema;
// missing entries are treated as non-date (safe fallback to raw `-`).
func isDateLike(node Node, ctx *CompileContext) bool {
	switch n := node.(type) {
	case *FieldRef:
		return isDateType(lookupFieldType(ctx, n.Name))
	case *CrossRef:
		return isDateType(lookupFieldType(ctx, n.Object+"."+n.Field))
	case *DeepCrossRef:
		return isDateType(lookupFieldType(ctx, strings.Join(n.Parts, ".")))
	case *FuncCall:
		upper := strings.ToUpper(n.Name)
		// Null-coalescing functions yield a date/datetime when every non-null
		// operand is date/datetime. Mirror the validator's mergeSchemaTypes: a
		// NULL literal is absorbed (it preserves the other operands' type). One
		// SSoT (nullCoalescingFuncs) covers both cataloged (BLANKVALUE/NULLVALUE)
		// and DSL-passthrough (IFNULL/COALESCE/NULLIF) coalescers.
		if nullCoalescingFuncs[upper] {
			sawDateLike := false
			for _, arg := range n.Args {
				if _, isNull := arg.(*NullLit); isNull {
					continue
				}
				if !isDateLike(arg, ctx) {
					return false
				}
				sawDateLike = true
			}
			return sawDateLike
		}
		return funcReturnsDateLike(upper, activeFunctionRegistry(ctx))
	case *BinaryOp:
		// date ± number → date-like; date − date → number (not date-like).
		// Only the number-operand shift forms produce a date result.
		switch n.Op {
		case "+":
			return (isDateLike(n.Left, ctx) && isNumeric(n.Right, ctx)) ||
				(isNumeric(n.Left, ctx) && isDateLike(n.Right, ctx))
		case "-":
			return isDateLike(n.Left, ctx) && isNumeric(n.Right, ctx)
		}
		return false
	case *IfExpr:
		return isDateLike(n.Then, ctx) || isDateLike(n.Else, ctx)
	case *CaseExpr:
		for _, br := range n.Branches {
			if isDateLike(br.Result, ctx) {
				return true
			}
		}
		return isDateLike(n.Default, ctx)
	}
	return false
}

// isNumeric reports whether a node evaluates to a numeric value — either a
// number literal or a field whose type is currency/double/int/percent.
// A number literal adjacent to a date operand is the trigger for DATE_ADD
// / DATE_SUB; without this guard we would over-eagerly rewrite every `-`.
func isNumeric(node Node, ctx *CompileContext) bool {
	switch n := node.(type) {
	case *NumberLit:
		return true
	case *FieldRef:
		return isNumericType(lookupFieldType(ctx, n.Name))
	case *CrossRef:
		return isNumericType(lookupFieldType(ctx, n.Object+"."+n.Field))
	case *DeepCrossRef:
		return isNumericType(lookupFieldType(ctx, strings.Join(n.Parts, ".")))
	case *FuncCall:
		return funcReturnsNumeric(strings.ToUpper(n.Name), activeFunctionRegistry(ctx))
	case *BinaryOp:
		// Nested arithmetic that is itself numeric (e.g. a date-date DATEDIFF
		// returns number — so (date - date) + 1 stays as (..) + 1).
		if n.Op == "-" && isDateLike(n.Left, ctx) && isDateLike(n.Right, ctx) {
			return true
		}
		switch n.Op {
		case "+", "-", "*", "/", "^":
			return isNumeric(n.Left, ctx) || isNumeric(n.Right, ctx)
		}
	case *UnaryOp:
		if n.Op == "-" {
			return isNumeric(n.Operand, ctx)
		}
	}
	return false
}

// Active-formula SQL lowering must match the runtime Evaluate path's
// Salesforce blank semantics for identifier-like expressions:
// `<ident> = ""` rewrites to ISBLANK(<ident>) and therefore stays
// type-aware for string, reference, date, numeric, and unknown operands.
// Issue #1301.
func compileSalesforceEmptyStringComparison(n *BinaryOp, leftSQL, rightSQL string, ctx *CompileContext) (string, bool) {
	switch n.Op {
	case "=", "!=", "<>":
	default:
		return "", false
	}

	if isEmptyStringLiteral(n.Left) {
		return emptyStringComparisonSQL(n.Op, n.Right, rightSQL, ctx)
	}
	if isEmptyStringLiteral(n.Right) {
		return emptyStringComparisonSQL(n.Op, n.Left, leftSQL, ctx)
	}
	return "", false
}

func isEmptyStringLiteral(node Node) bool {
	lit, ok := node.(*StringLit)
	return ok && lit.Value == ""
}

// compileNullComparison lowers `<expr> = NULL` / `<expr> != NULL` / `<expr> <> NULL`
// (and the NULL-on-left mirror) to SQL `IS NULL` / `IS NOT NULL`.
//
// Salesforce evaluates `blank = NULL` as true, and this platform's own kernel
// expression walker (pkg/expression/walker.go) already rewrites `== null` /
// `!= null` to `IS NULL` / `IS NOT NULL`. The DSL view compiler, however,
// previously fell through to the default binary lowering and emitted a raw
// `(expr = NULL)` — which under SQL three-valued logic is never true — so a
// ported `Field = NULL` idiom silently took the wrong branch on the
// query/report/list-view projection path while the same formula evaluated
// correctly through the Go engine. NX#3908.
//
// Precedence note: this runs AFTER compileSalesforceEmptyStringComparison so the
// degenerate `"" = NULL` (empty-string literal vs NULL) keeps the existing
// blank-aware ISBLANK shape rather than being read as `(” IS NULL)`.
func compileNullComparison(n *BinaryOp, leftSQL, rightSQL string) (string, bool) {
	switch n.Op {
	case "=", "!=", "<>":
	default:
		return "", false
	}

	leftIsNull := isNullLiteral(n.Left)
	rightIsNull := isNullLiteral(n.Right)
	if !leftIsNull && !rightIsNull {
		return "", false
	}

	isEquality := n.Op == "="

	// NULL = NULL is true, NULL != NULL is false — constant-fold to a SQL
	// predicate that is unconditionally true/false, matching SF and the kernel.
	if leftIsNull && rightIsNull {
		if isEquality {
			return "(NULL IS NULL)", true
		}
		return "(NULL IS NOT NULL)", true
	}

	operandSQL := leftSQL
	if leftIsNull {
		operandSQL = rightSQL
	}
	if isEquality {
		return fmt.Sprintf("(%s IS NULL)", operandSQL), true
	}
	return fmt.Sprintf("(%s IS NOT NULL)", operandSQL), true
}

func isNullLiteral(node Node) bool {
	_, ok := node.(*NullLit)
	return ok
}

func emptyStringComparisonSQL(op string, node Node, exprSQL string, ctx *CompileContext) (string, bool) {
	blankCheck, ok := blankCheckSQL(node, exprSQL, ctx)
	if !ok {
		return "", false
	}
	switch op {
	case "=":
		return blankCheck, true
	case "!=", "<>":
		return fmt.Sprintf("(NOT %s)", blankCheck), true
	default:
		return "", false
	}
}

func blankCheckSQL(node Node, exprSQL string, ctx *CompileContext) (string, bool) {
	switch classifyBlankOperand(node, ctx) {
	case blankOperandNullOnly:
		return fmt.Sprintf("(%s IS NULL)", exprSQL), true
	case blankOperandStringLike:
		// IFNULL applies to the boolean comparison, not to the value. This is
		// equivalent to `expr IS NULL OR expr = ''` under SQL three-valued
		// logic, names the compiled subtree once, and avoids coercing a raw
		// multipicklist JSON value to the invalid empty-string JSON document.
		return fmt.Sprintf("(IFNULL(%s = '', TRUE))", exprSQL), true
	case blankOperandUnknown:
		return fmt.Sprintf("(IFNULL(CAST(%s AS CHAR) = '', TRUE))", exprSQL), true
	default:
		return "", false
	}
}

type blankOperandKind int

const (
	blankOperandUnknown blankOperandKind = iota
	blankOperandStringLike
	blankOperandNullOnly
)

func classifyBlankOperand(node Node, ctx *CompileContext) blankOperandKind {
	switch n := node.(type) {
	case *StringLit:
		return blankOperandStringLike
	case *NumberLit, *BoolLit:
		return blankOperandNullOnly
	case *AggregateExpr:
		// SUM/COUNT/AVG are numeric by construction, and their lowered read is
		// coalesced at marker resolution (#4152) — the empty-string arm of a
		// blank function would coerce that 0 back to NULL. MIN/MAX fall through
		// to unknown: the child field's type is not visible here and a date
		// MIN/MAX must keep the blank-aware shape.
		switch strings.ToUpper(strings.TrimSpace(n.Func)) {
		case "SUM", "COUNT", "AVG":
			return blankOperandNullOnly
		}
		return blankOperandUnknown
	case *NullLit:
		return blankOperandUnknown
	case *FieldRef:
		return blankKindFromFieldType(lookupFieldType(ctx, n.Name))
	case *CrossRef:
		return blankKindFromFieldType(lookupFieldType(ctx, n.Object+"."+n.Field))
	case *DeepCrossRef:
		return blankKindFromFieldType(lookupFieldType(ctx, strings.Join(n.Parts, ".")))
	case *FuncCall:
		upper := strings.ToUpper(n.Name)
		if nullCoalescingFuncs[upper] {
			// Null-coalescing functions merge their operands' blank shapes
			// rather than a single declared return type, so they stay
			// argument-driven. Same nullCoalescingFuncs SSoT as the date-like
			// predicates; this branch only refines the blank-comparison SQL shape.
			return mergeBlankOperandKinds(n.Args, ctx)
		}
		if kind, ok := funcReturnBlankKind(upper, activeFunctionRegistry(ctx)); ok {
			return kind
		}
	case *BinaryOp:
		if n.Op == "&" {
			return blankOperandStringLike
		}
		switch n.Op {
		case "=", "!=", "<>", "<", "<=", ">", ">=":
			return blankOperandNullOnly
		}
		if isDateLike(n, ctx) || isNumeric(n, ctx) {
			return blankOperandNullOnly
		}
	case *UnaryOp:
		if n.Op == "NOT" || n.Op == "-" {
			return blankOperandNullOnly
		}
	case *LogicalOp:
		return blankOperandNullOnly
	case *IfExpr:
		return mergeBlankOperandKind(classifyBlankOperand(n.Then, ctx), classifyBlankOperand(n.Else, ctx))
	case *CaseExpr:
		kind := classifyBlankOperand(n.Default, ctx)
		for _, br := range n.Branches {
			kind = mergeBlankOperandKind(kind, classifyBlankOperand(br.Result, ctx))
		}
		return kind
	}
	return blankOperandUnknown
}

func blankKindFromFieldType(t string) blankOperandKind {
	switch {
	case isNullOnlyType(t):
		return blankOperandNullOnly
	case isStringLikeType(t):
		return blankOperandStringLike
	default:
		return blankOperandUnknown
	}
}

func mergeBlankOperandKinds(nodes []Node, ctx *CompileContext) blankOperandKind {
	kind := blankOperandUnknown
	for _, node := range nodes {
		kind = mergeBlankOperandKind(kind, classifyBlankOperand(node, ctx))
	}
	return kind
}

func mergeBlankOperandKind(left, right blankOperandKind) blankOperandKind {
	if left == blankOperandUnknown {
		return right
	}
	if right == blankOperandUnknown || left != right {
		return blankOperandUnknown
	}
	return left
}

// funcReturnBlankKind maps a function's declared return type to the
// blank-comparison operand shape used by classifyBlankOperand. For functions
// modelled in the shared catalog (pkg/formula/spec) the return type is read
// from the single source of truth via spec.DefaultRegistry — the active DSL
// compiler no longer re-declares the arity or return type of any cataloged
// function (VS3, issue #1467). Only DSL-only functions that have no catalog
// entry (SQL passthrough names like CONCAT/TRIM/SUBSTRING and the active-only
// VALUE/FLOOR/CEIL/ISNUMBER/DATE_SUB/BR helpers) keep a local classification —
// they are not shared Salesforce-modelled facts.
//
// The boolean result reports whether the function had a usable fixed return
// type; an unrecognised name (or a merge-kind return that this blank-shape
// classification does not model) returns false so the caller falls through to
// its argument-merge / unknown handling.
func funcReturnBlankKind(upper string, registry *spec.Registry) (blankOperandKind, bool) {
	if registry == nil {
		registry = spec.DefaultRegistry()
	}
	if fnSpec, ok := registry.Lookup(upper); ok {
		if fnSpec.Return.Kind == spec.ReturnFixed {
			return blankKindForValueType(fnSpec.Return.Fixed), true
		}
		return blankOperandUnknown, false
	}
	if kind, ok := dslOnlyReturnBlankKind[upper]; ok {
		return kind, true
	}
	return blankOperandUnknown, false
}

// blankKindForValueType maps a spec.ValueType return type to the blank-operand
// shape: string-like values may be the empty string, so they take the
// string-like (NULL-or-empty) shape; date/datetime/number/boolean values are
// never the empty string, so a NULL check alone is sufficient.
func blankKindForValueType(t spec.ValueType) blankOperandKind {
	switch t {
	case spec.TypeString:
		return blankOperandStringLike
	case spec.TypeNumber, spec.TypeBoolean, spec.TypeDate, spec.TypeDatetime:
		return blankOperandNullOnly
	default:
		return blankOperandUnknown
	}
}

// dslOnlyReturnBlankKind classifies the active-formula-only functions that have
// no shared catalog entry. Shared (cataloged) functions are intentionally
// absent here: their return type is read from spec via funcReturnBlankKind, so
// this table never re-declares a catalog fact (the drift guard enforces this).
var dslOnlyReturnBlankKind = map[string]blankOperandKind{
	// Date/datetime-returning (NULL-only): SQL date helper with no catalog entry.
	"DATE_SUB": blankOperandNullOnly,
	// Numeric-returning (NULL-only): active-only numeric helpers.
	"VALUE":    blankOperandNullOnly,
	"FLOOR":    blankOperandNullOnly,
	"CEIL":     blankOperandNullOnly,
	"ISNUMBER": blankOperandNullOnly, // boolean-returning, NULL-only shape
	// String-returning (string-like): SQL string passthrough + active-only helpers.
	"TRIM":        blankOperandStringLike,
	"BR":          blankOperandStringLike,
	"CONCAT":      blankOperandStringLike,
	"CONCAT_WS":   blankOperandStringLike,
	"REPLACE":     blankOperandStringLike,
	"REVERSE":     blankOperandStringLike,
	"LTRIM":       blankOperandStringLike,
	"RTRIM":       blankOperandStringLike,
	"SUBSTRING":   blankOperandStringLike,
	"SUBSTR":      blankOperandStringLike,
	"DATE_FORMAT": blankOperandStringLike,
}

// funcReturnsDateLike reports whether the named function returns a date or
// datetime value, for the date-arithmetic SQL lowering in compileBinaryOp.
// Cataloged functions are resolved from the shared spec return type (date or
// datetime); the active DSL-only DATE_SUB helper — which has no catalog entry —
// is the only local addition. No catalog return type is re-declared here.
func funcReturnsDateLike(upper string, registry *spec.Registry) bool {
	if registry == nil {
		registry = spec.DefaultRegistry()
	}
	if fnSpec, ok := registry.Lookup(upper); ok {
		return fnSpec.Return.Kind == spec.ReturnFixed &&
			(fnSpec.Return.Fixed == spec.TypeDate || fnSpec.Return.Fixed == spec.TypeDatetime)
	}
	return dslOnlyDateLikeFuncs[upper]
}

// dslOnlyDateLikeFuncs lists the active-formula-only date helpers with no shared
// catalog entry. Cataloged date/datetime functions are intentionally absent —
// they are resolved from spec in funcReturnsDateLike.
var dslOnlyDateLikeFuncs = map[string]bool{
	"DATE_SUB": true,
}

// funcReturnsNumeric reports whether the named function returns a number, for
// the date-arithmetic SQL lowering in compileBinaryOp. Cataloged functions are
// resolved from the shared spec return type; the active DSL-only numeric
// helpers (VALUE/FLOOR/CEIL) — none of which have a catalog entry — are the only
// local additions. No catalog return type is re-declared here.
func funcReturnsNumeric(upper string, registry *spec.Registry) bool {
	if registry == nil {
		registry = spec.DefaultRegistry()
	}
	if fnSpec, ok := registry.Lookup(upper); ok {
		return fnSpec.Return.Kind == spec.ReturnFixed && fnSpec.Return.Fixed == spec.TypeNumber
	}
	return dslOnlyNumericFuncs[upper]
}

// dslOnlyNumericFuncs lists the active-formula-only numeric helpers with no
// shared catalog entry. Cataloged number-returning functions are intentionally
// absent — they are resolved from spec in funcReturnsNumeric.
var dslOnlyNumericFuncs = map[string]bool{
	"VALUE": true,
	"FLOOR": true,
	"CEIL":  true,
}

// dslPassthroughFuncs is the allowlist of known-safe SQL functions the generic
// compileFuncCall fallback emits verbatim, preventing injection of dangerous
// functions. Every entry is a DSL-only function with NO shared catalog entry:
// cataloged functions (MOD, DATE_ADD, ROUND, CEILING, LPAD, RPAD, ISNULL,
// NOW, ...) all have explicit cases that route arity through the active
// function registry, so the generic passthrough never owns a shared catalog
// function's arity. The drift guard asserts this list and the catalog stay
// disjoint.
var dslPassthroughFuncs = map[string]bool{
	"COALESCE": true, "CONCAT": true, "CONCAT_WS": true, "NULLIF": true,
	"IFNULL": true, "CAST": true, "CONVERT": true,
	"FLOOR": true, "CEIL": true,
	"POWER": true, "POW": true, "SQRT": true, "SIGN": true,
	"SUBSTRING": true, "SUBSTR": true, "REPLACE": true, "REVERSE": true,
	"LTRIM": true, "RTRIM": true,
	"DATE_FORMAT": true, "STR_TO_DATE": true, "DATE_SUB": true,
	"DATEDIFF": true, "TIMESTAMPDIFF": true, "CURDATE": true,
	"GREATEST": true, "LEAST": true, "LENGTH": true, "CHAR_LENGTH": true,
}

func lookupFieldType(ctx *CompileContext, key string) string {
	if ctx == nil {
		return ""
	}
	if v, ok := lookupCaseInsensitive(ctx.BoundIdentifierTypes, key); ok {
		return v
	}
	if v, ok := lookupCaseInsensitive(ctx.FieldTypes, key); ok {
		return v
	}
	return ""
}

// toFormulaDomainSQL converts a raw field read into the value the formula
// domain sees. Two conversions compose here, and both are field-type driven:
//
//   - blank-as-zero (#4152): a blank numeric operand reads as 0, matching the
//     Salesforce formulaTreatBlanksAs=BlankAsZero default. Without this, any
//     arithmetic or comparison over an optional numeric field NULL-propagates
//     and the whole formula projects blank. The COALESCE applies to every
//     numeric read this compiler lowers — including ISBLANK/ISNULL arguments,
//     which therefore see 0 and report "not blank", exactly as the vendor
//     documents for BlankAsZero. Types outside IsBlankAsZeroNumericType
//     (strings, dates, booleans, references) keep NULL untouched.
//   - percent scaling: storage/API carry percents in the 0–100 display domain
//     while formula text computes in the 0–1 fraction domain, so percent reads
//     divide by 100 (the write-back side multiplies, fromFormulaDomainSQL).
//
// Fields with no known type (empty fieldType) are left untouched: fail open to
// today's NULL propagation rather than guessing a numeric coercion.
func toFormulaDomainSQL(sqlExpr string, fieldType string) string {
	if constants.NormalizeFieldType(fieldType) == constants.FieldTypePercent {
		return fmt.Sprintf("(COALESCE(%s, 0) / 100.0)", sqlExpr)
	}
	if isNumericType(fieldType) {
		return fmt.Sprintf("COALESCE(%s, 0)", sqlExpr)
	}
	return sqlExpr
}

func isDateType(t string) bool {
	switch strings.ToLower(t) {
	case "date", "datetime":
		return true
	}
	return false
}

// isDatetimeType reports whether the field type string is specifically
// "datetime" (not "date"). Mirrors isDateType's normalisation.
func isDatetimeType(t string) bool {
	return strings.ToLower(strings.TrimSpace(t)) == "datetime"
}

// funcReturnsDatetime reports whether the named function (already uppercased)
// returns a datetime value per the shared spec catalog. Date-only functions
// (TODAY, DATE, DATEVALUE, __DATE_ADD_MONTHS__, __DATE_ADD_MONTHS_TARGET_CLAMP__,
// __DATE_DIFF_DAYS__, DATE_ADD) and DSL-only helpers
// (DATE_SUB) return false.
func funcReturnsDatetime(upper string, registry *spec.Registry) bool {
	if registry == nil {
		registry = spec.DefaultRegistry()
	}
	if fnSpec, ok := registry.Lookup(upper); ok {
		return fnSpec.Return.Kind == spec.ReturnFixed && fnSpec.Return.Fixed == spec.TypeDatetime
	}
	return false
}

// isDatetimeLike reports whether a node evaluates specifically to a datetime
// value (not a date-only value). Structurally parallel to isDateLike but
// datetime-only: used by compileBinaryOp to choose TIMESTAMPDIFF over DATEDIFF
// when both operands carry time-of-day information.
func isDatetimeLike(node Node, ctx *CompileContext) bool {
	switch n := node.(type) {
	case *FieldRef:
		return isDatetimeType(lookupFieldType(ctx, n.Name))
	case *CrossRef:
		return isDatetimeType(lookupFieldType(ctx, n.Object+"."+n.Field))
	case *DeepCrossRef:
		return isDatetimeType(lookupFieldType(ctx, strings.Join(n.Parts, ".")))
	case *FuncCall:
		upper := strings.ToUpper(n.Name)
		// Null-coalescing functions yield a datetime when every non-null operand
		// is datetime. Mirror the validator's mergeSchemaTypes: a NULL literal is
		// absorbed (it preserves the other operands' type). Same nullCoalescingFuncs
		// SSoT as isDateLike / classifyBlankOperand.
		if nullCoalescingFuncs[upper] {
			sawDatetime := false
			for _, arg := range n.Args {
				if _, isNull := arg.(*NullLit); isNull {
					continue
				}
				if !isDatetimeLike(arg, ctx) {
					return false
				}
				sawDatetime = true
			}
			return sawDatetime
		}
		return funcReturnsDatetime(upper, activeFunctionRegistry(ctx))
	case *BinaryOp:
		// datetime ± number → datetime-like; datetime − datetime → number.
		// Only the number-operand shift forms preserve the datetime type.
		switch n.Op {
		case "+":
			return (isDatetimeLike(n.Left, ctx) && isNumeric(n.Right, ctx)) ||
				(isNumeric(n.Left, ctx) && isDatetimeLike(n.Right, ctx))
		case "-":
			return isDatetimeLike(n.Left, ctx) && isNumeric(n.Right, ctx)
		}
		return false
	case *IfExpr:
		return isDatetimeLike(n.Then, ctx) || isDatetimeLike(n.Else, ctx)
	case *CaseExpr:
		for _, br := range n.Branches {
			if isDatetimeLike(br.Result, ctx) {
				return true
			}
		}
		return isDatetimeLike(n.Default, ctx)
	}
	return false
}

// isNumericType delegates to the shared blank-as-zero numeric bucket so this
// compiler and the Go evaluator classify the same types as numeric. It also
// remains the numeric side of date-arithmetic detection (isNullOnlyType).
func isNumericType(t string) bool {
	return schemautil.IsBlankAsZeroNumericType(t)
}

func isStringLikeType(t string) bool {
	// JSON intentionally stays in the unknown bucket: schema validation treats
	// it as schemaTypeUnknown and fieldTypes.json stores it as SQL JSON, so the
	// TiDB-safe CAST fallback is the least surprising emission if it appears.
	switch constants.NormalizeFieldType(t) {
	case constants.FieldTypeString,
		constants.FieldTypeTextarea,
		constants.FieldTypeRichtextarea,
		constants.FieldTypeEmail,
		constants.FieldTypePhone,
		constants.FieldTypeUrl,
		constants.FieldTypePicklist,
		constants.FieldTypeMultiPicklist,
		constants.FieldTypeAutonumber,
		constants.FieldTypeEncryptedString,
		constants.FieldTypePassword,
		// Reference-like IDs can be persisted as empty string on existing rows,
		// so ISBLANK must stay blank-aware rather than null-only.
		constants.FieldTypeReference,
		constants.FieldTypeMasterDetail,
		constants.SchemaFieldType("lookup"):
		return true
	default:
		return false
	}
}

func isBooleanType(t string) bool {
	switch constants.NormalizeFieldType(t) {
	case constants.FieldTypeBoolean, constants.SchemaFieldType("checkbox"):
		return true
	default:
		return false
	}
}

func isNullOnlyType(t string) bool {
	return isDateType(t) || isNumericType(t) || isBooleanType(t)
}

func compileUnaryOp(n *UnaryOp, ctx *CompileContext) (string, error) {
	operand, err := compile(n.Operand, ctx)
	if err != nil {
		return "", err
	}
	switch n.Op {
	case "-":
		return fmt.Sprintf("(-%s)", operand), nil
	case "NOT":
		return fmt.Sprintf("(NOT %s)", operand), nil
	default:
		return "", fmt.Errorf("unsupported unary op: %s", n.Op)
	}
}

// compileLogicalOp emits bare SQL AND/OR.
//
// KNOWN, DELIBERATE FORMULA-DOMAIN DIVERGENCE (null provenance). SQL three-valued logic
// short-circuits over NULL: a decisive operand wins and swallows an #Error
// sibling. Verified on TiDB v8.5.0:
//
//	SELECT ((0=1) AND NULL) IS NULL;  -- 0  (FALSE, the NULL is swallowed)
//	SELECT ((1=1) AND NULL) IS NULL;  -- 1  (NULL survives)
//
// The formula error contract instead poisons the whole formula whenever any
// operand raises
// #Error, regardless of the other operand. So AND(FALSE, x/0) is FALSE here and
// #Error under formula evaluation.
//
// Left unfixed ON PURPOSE, not overlooked: a scan of all merged-pack formula
// fields found ZERO AND()/OR() groups containing a field-denominator division
// and zero uses of MOD/SQRT/LOG/LN, so strictness here would change no
// production value while perturbing the emitted SQL of ~50 formulas.
//
// PRECISELY WHAT IS AND IS NOT UNFIXED — do not read the above as "AND/OR
// conditions were left alone". An enclosing IF still guards the JOINED
// expression whenever the whole condition passes the provenance gate, so
// AND/OR-bearing conditions do get partial, data-dependent strictness:
//
//	IF(Num__c/Den__c > 0.5 || Amount > 100, 1, 2)   with Den__c = 0
//	  NULL OR TRUE  -> TRUE  -> THEN arm   (formula domain: blank)   still divergent
//	  NULL OR FALSE -> NULL  -> blank      (formula domain: blank)   now correct
//
// Never worse than today, sometimes better. What remains unfixed is strictly
// OPERAND-LEVEL short-circuit: a decisive sibling operand swallowing an #Error
// before the joined expression ever reaches the IF boundary.
//
// The fix shape, if a future formula needs it: make the operands themselves
// extra NULL guards, i.e. wrap the join as
// a form that binds each operand once — NOT
// `CASE WHEN (op1) IS NULL OR (op2) IS NULL THEN NULL ELSE (op1 AND op2) END`,
// which names every operand twice and would reintroduce the doubling this
// file removed,
// gated by the same NULL-provenance delta window compileIf uses. Do not
// rediscover this as a fresh bug.
func compileLogicalOp(n *LogicalOp, ctx *CompileContext) (string, error) {
	parts := make([]string, len(n.Operands))
	for i, op := range n.Operands {
		s, err := compile(op, ctx)
		if err != nil {
			return "", err
		}
		parts[i] = s
	}
	return "(" + strings.Join(parts, " "+n.Op+" ") + ")", nil
}

func compileFuncCall(n *FuncCall, ctx *CompileContext) (string, error) {
	// Map SF function names to TiDB equivalents
	upper := strings.ToUpper(n.Name)

	// Functions whose SQL lowering is the multipicklist "contains all" strategy
	// keep native emission even when their call shape is supplied by an L1 bundle;
	// their formula_expr_v1 body delegates back to INCLUDES and would otherwise
	// recurse through the generic executable lowering path. Dispatch is on the
	// registered SQLLowering capability flag, with a by-name INCLUDES fallback
	// retained transitionally (removed in PR-B).
	if upper != "INCLUDES" && !compileFuncUsesMultipicklistContainsLowering(ctx, upper) {
		if sql, handled, err := compileExecutableFunctionCall(n, upper, ctx); handled || err != nil {
			return sql, err
		}
	}

	if compileFuncUsesMultipicklistContainsLowering(ctx, upper) {
		return compileIncludesNativeSQL(n, ctx)
	}

	switch upper {
	case "LOOKUPVALUE":
		sql, err := compileLookupValue(n, ctx)
		if err != nil {
			return "", fmt.Errorf("%w: %v", ErrRestrictedFormulaPrimitive, err)
		}
		return sql, nil

	case "ISBLANK":
		if err := checkFunctionArity(ctx, "ISBLANK", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		blankCheck, ok := blankCheckSQL(n.Args[0], arg, ctx)
		if !ok {
			return "", fmt.Errorf("unsupported ISBLANK operand: %T", n.Args[0])
		}
		return blankCheck, nil

	case "TEXT":
		if err := checkFunctionArity(ctx, "TEXT", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		switch classifyBlankOperand(n.Args[0], ctx) {
		case blankOperandStringLike:
			return fmt.Sprintf("COALESCE(%s, '')", arg), nil
		default:
			return fmt.Sprintf("COALESCE(CAST(%s AS CHAR), '')", arg), nil
		}

	case "TODAY":
		if err := checkFunctionArity(ctx, "TODAY", len(n.Args)); err != nil {
			return "", err
		}
		return "CURDATE()", nil

	case "NOW":
		if err := checkFunctionArity(ctx, "NOW", len(n.Args)); err != nil {
			return "", err
		}
		return "NOW()", nil

	case "LEN":
		if err := checkFunctionArity(ctx, "LEN", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("CHAR_LENGTH(%s)", arg), nil

	case "CONTAINS":
		if err := checkFunctionArity(ctx, "CONTAINS", len(n.Args)); err != nil {
			return "", err
		}
		field, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		substr, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("(%s LIKE CONCAT('%%', %s, '%%'))", field, substr), nil

	case "BEGINS":
		if err := checkFunctionArity(ctx, "BEGINS", len(n.Args)); err != nil {
			return "", err
		}
		field, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		prefix, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("(%s LIKE CONCAT(%s, '%%'))", field, prefix), nil

	case "FIND":
		if err := checkFunctionArity(ctx, "FIND", len(n.Args)); err != nil {
			return "", err
		}
		search, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		str, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		if len(n.Args) == 3 {
			pos, err := compile(n.Args[2], ctx)
			if err != nil {
				return "", err
			}
			return fmt.Sprintf("LOCATE(%s, %s, %s)", search, str, pos), nil
		}
		return fmt.Sprintf("LOCATE(%s, %s)", search, str), nil

	case "NULLVALUE":
		if err := checkFunctionArity(ctx, "NULLVALUE", len(n.Args)); err != nil {
			return "", err
		}
		field, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		def, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("IFNULL(%s, %s)", field, def), nil

	case "BLANKVALUE":
		if err := checkFunctionArity(ctx, "BLANKVALUE", len(n.Args)); err != nil {
			return "", err
		}
		field, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		def, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		// The NULLIF('') empty-string arm exists for string-like operands where
		// '' and NULL both mean blank. For null-only operands (numeric, date,
		// boolean — classifyBlankOperand's blankOperandNullOnly) that arm is a
		// type-coercion hazard: MySQL/TiDB coerce '' to 0/false in a numeric or
		// boolean comparison, so NULLIF(field, '') turned a legally stored 0
		// (or false) into the default. Under blank-as-zero (#4152) a blank
		// numeric operand is already coalesced to 0 before it reaches this
		// function, so the vendor-correct result is that 0 — never the default.
		if classifyBlankOperand(n.Args[0], ctx) == blankOperandNullOnly {
			return fmt.Sprintf("IFNULL(%s, %s)", field, def), nil
		}
		return fmt.Sprintf("IFNULL(NULLIF(%s, ''), %s)", field, def), nil

	case "SUBSTITUTE":
		if err := checkFunctionArity(ctx, "SUBSTITUTE", len(n.Args)); err != nil {
			return "", err
		}
		// SUBSTITUTE's 4th arg (instance number) is a valid Salesforce arity
		// per the shared catalog (3-4), but the active-formula SQL compiler
		// lowers SUBSTITUTE to TiDB REPLACE(), which replaces all occurrences
		// and cannot express "replace the Nth instance". Fail loudly rather
		// than silently dropping the 4th arg.
		if len(n.Args) == 4 {
			return "", fmt.Errorf("SUBSTITUTE with an instance-number 4th argument is not supported in active formulas")
		}
		text, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		old, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		new, err := compile(n.Args[2], ctx)
		if err != nil {
			return "", err
		}
		// In SF, NULL as replacement means empty string. TiDB REPLACE returns NULL if any arg is NULL.
		return fmt.Sprintf("REPLACE(%s, %s, IFNULL(%s, ''))", text, old, new), nil

	case "MID":
		if err := checkFunctionArity(ctx, "MID", len(n.Args)); err != nil {
			return "", err
		}
		text, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		start, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		length, err := compile(n.Args[2], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("SUBSTRING(%s, %s, %s)", text, start, length), nil

	case "LEFT":
		if err := checkFunctionArity(ctx, "LEFT", len(n.Args)); err != nil {
			return "", err
		}
		text, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		length, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("LEFT(%s, %s)", text, length), nil

	case "RIGHT":
		if err := checkFunctionArity(ctx, "RIGHT", len(n.Args)); err != nil {
			return "", err
		}
		text, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		length, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("RIGHT(%s, %s)", text, length), nil

	case "__DATE_ADD_MONTHS__":
		if err := checkFunctionArity(ctx, "__DATE_ADD_MONTHS__", len(n.Args)); err != nil {
			return "", err
		}
		date, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		months, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		// Salesforce last-day anchoring without repeating a compiled operand
		// four times. Candidate 1 is MySQL/TiDB's ordinary month-add clamp.
		// Candidate 2 shifts the source one day forward, adds the months, then
		// shifts back: it equals candidate 1 for ordinary days, is one day
		// earlier when only the shifted date overflows the target month, and is
		// the target month's last day when the source was month-end. GREATEST
		// therefore selects exactly the Salesforce result while naming each
		// compiled operand only twice.
		//
		// DATE_ADD('9999-12-31', INTERVAL 1 DAY) overflows to NULL. The inner
		// IFNULL keeps candidate 2 below candidate 1 for zero/negative month
		// shifts. The shifted candidate can also overflow while candidate 1 is
		// still representable (for example a month-end source targeting December
		// 9999); in exactly that case Salesforce's anchored answer is the maximum
		// SQL DATE, so the outer IFNULL supplies 9999-12-31. For an out-of-range
		// candidate 1 or a NULL input, GREATEST's NULL propagation preserves the
		// old NULL result regardless of the fallback candidate.
		result := fmt.Sprintf(
			"GREATEST(DATE_ADD(%s, INTERVAL %s MONTH), IFNULL(DATE_SUB(DATE_ADD(IFNULL(DATE_ADD(%s, INTERVAL 1 DAY), DATE('9999-12-31')), INTERVAL %s MONTH), INTERVAL 1 DAY), DATE('9999-12-31')))",
			date, months, date, months,
		)
		// GREATEST may select a common scalar representation even when every
		// operand is Date, so keep the existing outer DATE normalization here.
		return fmt.Sprintf("DATE(%s)", result), nil

	case "__DATE_ADD_MONTHS_TARGET_CLAMP__":
		if err := checkFunctionArity(ctx, "__DATE_ADD_MONTHS_TARGET_CLAMP__", len(n.Args)); err != nil {
			return "", err
		}
		date, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		months, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		result := fmt.Sprintf("DATE_ADD(%s, INTERVAL %s MONTH)", date, months)
		return normalizeCatalogDateSQL(result, n.Args[0], ctx), nil

	case "__DATE_DIFF_DAYS__":
		if err := checkFunctionArity(ctx, "__DATE_DIFF_DAYS__", len(n.Args)); err != nil {
			return "", err
		}
		end, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		start, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("DATEDIFF(%s, %s)", end, start), nil

	case "YEAR":
		if err := checkFunctionArity(ctx, "YEAR", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("YEAR(%s)", arg), nil

	case "MONTH":
		if err := checkFunctionArity(ctx, "MONTH", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("MONTH(%s)", arg), nil

	case "DAY":
		if err := checkFunctionArity(ctx, "DAY", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("DAY(%s)", arg), nil

	case "ROUND":
		// Salesforce ROUND is exactly 2 args (number, places); the prior 1-arg
		// form is rejected by the shared catalog. Arity is now a single
		// catalog-owned fact.
		if err := checkFunctionArity(ctx, "ROUND", len(n.Args)); err != nil {
			return "", err
		}
		val, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		places, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("ROUND(%s, %s)", val, places), nil

	case "CEILING":
		if err := checkFunctionArity(ctx, "CEILING", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("CEIL(%s)", arg), nil

	case "FLOOR":
		if len(n.Args) != 1 {
			return "", fmt.Errorf("FLOOR requires 1 argument")
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("FLOOR(%s)", arg), nil

	case "ABS":
		if err := checkFunctionArity(ctx, "ABS", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("ABS(%s)", arg), nil

	case "MAX", "MIN":
		// These can be regular SQL functions with multiple args
		args := make([]string, len(n.Args))
		for i, a := range n.Args {
			s, err := compile(a, ctx)
			if err != nil {
				return "", err
			}
			args[i] = s
		}
		if len(args) == 2 {
			if upper == "MIN" {
				return fmt.Sprintf("LEAST(%s, %s)", args[0], args[1]), nil
			}
			return fmt.Sprintf("GREATEST(%s, %s)", args[0], args[1]), nil
		}
		return fmt.Sprintf("%s(%s)", upper, strings.Join(args, ", ")), nil

	case "TRIM":
		if len(n.Args) != 1 {
			return "", fmt.Errorf("TRIM requires 1 argument")
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("TRIM(%s)", arg), nil

	case "UPPER":
		if err := checkFunctionArity(ctx, "UPPER", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("UPPER(%s)", arg), nil

	case "LOWER":
		if err := checkFunctionArity(ctx, "LOWER", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("LOWER(%s)", arg), nil

	case "LPAD":
		if err := checkFunctionArity(ctx, "LPAD", len(n.Args)); err != nil {
			return "", err
		}
		args := make([]string, 3)
		for i := range n.Args {
			s, err := compile(n.Args[i], ctx)
			if err != nil {
				return "", err
			}
			args[i] = s
		}
		return fmt.Sprintf("LPAD(%s, %s, %s)", args[0], args[1], args[2]), nil

	case "RPAD":
		if err := checkFunctionArity(ctx, "RPAD", len(n.Args)); err != nil {
			return "", err
		}
		args := make([]string, 3)
		for i := range n.Args {
			s, err := compile(n.Args[i], ctx)
			if err != nil {
				return "", err
			}
			args[i] = s
		}
		return fmt.Sprintf("RPAD(%s, %s, %s)", args[0], args[1], args[2]), nil

	case "VALUE":
		// VALUE(text) → CAST to number
		if len(n.Args) != 1 {
			return "", fmt.Errorf("VALUE requires 1 argument")
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("CAST(%s AS DECIMAL(%d,%d))", arg, DefaultDecimalPrecision, DefaultDecimalScale), nil

	case "DATEVALUE":
		if err := checkFunctionArity(ctx, "DATEVALUE", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("DATE(%s)", arg), nil

	case "DATETIMEVALUE":
		if err := checkFunctionArity(ctx, "DATETIMEVALUE", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("TIMESTAMP(%s)", arg), nil

	case "ISNULL":
		if err := checkFunctionArity(ctx, "ISNULL", len(n.Args)); err != nil {
			return "", err
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("(%s IS NULL)", arg), nil

	case "ISNUMBER":
		if len(n.Args) != 1 {
			return "", fmt.Errorf("ISNUMBER requires 1 argument")
		}
		arg, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("(%s REGEXP '^-?[0-9]+(\\\\.[0-9]+)?$')", arg), nil

	case "INCLUDES":
		return compileIncludesNativeSQL(n, ctx)

	case "BR":
		// BR() → newline in formula output
		return "'\\n'", nil

	case "DATE":
		// DATE(year, month, day) → STR_TO_DATE. Arity (exactly 3) is enforced
		// by the shared catalog, matching the Salesforce DATE() contract; the
		// prior 1-arg form is now rejected.
		if err := checkFunctionArity(ctx, "DATE", len(n.Args)); err != nil {
			return "", err
		}
		y, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", fmt.Errorf("DATE arg 1: %w", err)
		}
		m, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", fmt.Errorf("DATE arg 2: %w", err)
		}
		d, err := compile(n.Args[2], ctx)
		if err != nil {
			return "", fmt.Errorf("DATE arg 3: %w", err)
		}
		return fmt.Sprintf("STR_TO_DATE(CONCAT(%s,'-',%s,'-',%s), '%%Y-%%m-%%d')", y, m, d), nil

	case "MOD":
		// MOD(number, divisor) → native SQL MOD. A cataloged function (arity 2);
		// it is routed explicitly through the active function registry rather than
		// the generic passthrough so its arity is the single catalog-owned fact,
		// not a silent pass-through that would accept any argument count.
		if err := checkFunctionArity(ctx, "MOD", len(n.Args)); err != nil {
			return "", err
		}
		a, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		b, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("MOD(%s, %s)", a, b), nil

	case "DATE_ADD":
		// Salesforce DATE_ADD(date, days) shifts a date forward by N days. A
		// cataloged function (arity 2); routed explicitly through the active
		// function registry and lowered to the TiDB INTERVAL form —
		// DATE_ADD(date, INTERVAL days DAY) — matching the date + number emission
		// elsewhere in this compiler, instead of the prior raw passthrough that emitted a literal
		// DATE_ADD(date, days) MySQL rejects.
		if err := checkFunctionArity(ctx, "DATE_ADD", len(n.Args)); err != nil {
			return "", err
		}
		date, err := compile(n.Args[0], ctx)
		if err != nil {
			return "", err
		}
		days, err := compile(n.Args[1], ctx)
		if err != nil {
			return "", err
		}
		result := fmt.Sprintf("DATE_ADD(%s, INTERVAL (%s) DAY)", date, days)
		return normalizeCatalogDateSQL(result, n.Args[0], ctx), nil

	default:
		if !dslPassthroughFuncs[upper] {
			return "", fmt.Errorf("unsupported function: %s", upper)
		}
		args := make([]string, len(n.Args))
		for i, a := range n.Args {
			s, err := compile(a, ctx)
			if err != nil {
				return "", err
			}
			args[i] = s
		}
		return fmt.Sprintf("%s(%s)", upper, strings.Join(args, ", ")), nil
	}
}

// normalizeCatalogDateSQL enforces a catalog-declared Date result without
// charging known Date-only hot paths for a redundant DATE projection. Unknown
// and DateTime inputs normalize explicitly because TiDB DATE_ADD otherwise
// preserves their time component, diverging from the runtime evaluator.
func normalizeCatalogDateSQL(sql string, input Node, ctx *CompileContext) string {
	if isDateLike(input, ctx) && !isDatetimeLike(input, ctx) {
		return sql
	}
	return fmt.Sprintf("DATE(%s)", sql)
}

func includesStringLiteralRHSTokens(arg Node) ([]string, bool) {
	rhs, ok := arg.(*StringLit)
	if !ok || !strings.Contains(rhs.Value, ";") {
		return nil, false
	}
	return dedupeTrimmedFormulaStrings(strings.Split(rhs.Value, ";")), true
}

func compileIncludesMultiTokenSQL(field string, tokens []string) string {
	tokenSQLs := make([]string, 0, len(tokens))
	for _, token := range tokens {
		tokenSQLs = append(tokenSQLs, "'"+utils.EscapeSQLString(token)+"'")
	}
	return utils.MultiPicklistJSONContainsAllSQL(field, tokenSQLs)
}

func dedupeTrimmedFormulaStrings(in []string) []string {
	if len(in) == 0 {
		return nil
	}
	out := make([]string, 0, len(in))
	seen := make(map[string]struct{}, len(in))
	for _, item := range in {
		item = strings.TrimSpace(item)
		if item == "" {
			continue
		}
		if _, ok := seen[item]; ok {
			continue
		}
		seen[item] = struct{}{}
		out = append(out, item)
	}
	return out
}

// conditionIsBooleanShaped reports whether a condition node evaluates to the
// three-valued {TRUE, FALSE, NULL} domain rather than to an arbitrary value
// subjected to MySQL truthiness. Only these shapes may use the
// single-substitution strict emission in compileIf/compileCase.
//
// Deliberately structural rather than type-inferred: a node either is a
// comparison, a logical join or a negation, or it is not. Anything else —
// including a function call that happens to return a boolean — returns false
// and keeps the absorbing emission, which is never worse than pre-gate
// behavior.
func conditionIsBooleanShaped(node Node) bool {
	switch n := node.(type) {
	case *BinaryOp:
		switch n.Op {
		case "=", "==", "!=", "<>", "<", "<=", ">", ">=":
			return true
		}
		return false
	case *LogicalOp:
		// Two or more operands always join through SQL AND/OR, which is
		// three-valued whatever the operand types are. ONE operand is not:
		// compileLogicalOp joins with strings.Join, so AND(x) emits the bare
		// `(x)` with no boolean coercion, and the parser and the shared
		// function catalog both accept single-argument AND/OR. Admitting it
		// would apply the single-substitution form to an arbitrary numeric
		// expression: IF(AND(Num__c/Den__c), 10, 20) over 10/2 returns 10
		// today and NULL under the strict form, a wrong blank on a row whose
		// arithmetic is perfectly defined.
		//
		// A single operand is therefore judged by what it actually emits: the
		// operand itself. AND(x/y > 1) stays strict, AND(x/y) does not.
		if len(n.Operands) == 1 {
			return conditionIsBooleanShaped(n.Operands[0])
		}
		return true
	case *UnaryOp:
		return n.Op == "NOT" || n.Op == "!"
	}
	return false
}

func compileIf(n *IfExpr, ctx *CompileContext) (string, error) {
	// Delta window covers the CONDITION ONLY. An #Error in a branch VALUE
	// already propagates correctly (the CASE returns the NULL it selected), and
	// widening the window would churn shapes like IF(Amount > 0, a/b, 0) for no
	// semantic gain. See formula_dsl_null_provenance.go (null provenance).
	arithBefore, blankBefore := ctx.nullSiteCounts()
	cond, err := compile(n.Condition, ctx)
	if err != nil {
		return "", err
	}
	arithAfter, blankAfter := ctx.nullSiteCounts()
	strict := strictNullPropagation(arithBefore, blankBefore, arithAfter, blankAfter)

	then, err := compile(n.Then, ctx)
	if err != nil {
		return "", err
	}
	elseVal, err := compile(n.Else, ctx)
	if err != nil {
		return "", err
	}
	if strict && conditionIsBooleanShaped(n.Condition) {
		// The formula error contract poisons the whole formula on #Error; mirror
		// that by
		// returning NULL rather than letting the CASE fall through to the else
		// arm.
		//
		// The condition is substituted ONCE. An earlier revision emitted
		// `CASE WHEN (c) IS NULL THEN NULL WHEN c THEN t ELSE f END`, which
		// names the condition twice; a guarded IF nested inside another
		// guarded IF's CONDITION then doubles the emitted text at every level.
		// Measured on this compiler: a 291-character formula nested 14 deep
		// produced 7,503,664 characters of SQL against 1,611 before the gate
		// existed. The parser admits 256 levels and nothing bounds output
		// size, so that is a real hazard for admin-authored metadata even
		// though no pack formula has the shape.
		//
		// The single-substitution form relies on the condition being
		// three-valued {TRUE, FALSE, NULL}, which is why it is gated on
		// conditionIsBooleanShaped: a comparison, a logical join or a
		// negation. A bare non-boolean condition (MySQL truthiness, e.g.
		// IF(5, a, b)) would compare 5 = TRUE and wrongly fall to the implicit
		// ELSE, so those keep the absorbing emission below — no strictness,
		// but identical to pre-gate behavior. Fail-closed, same as every other
		// unclassifiable node in this gate.
		return fmt.Sprintf(
			"CASE (%s) WHEN TRUE THEN %s WHEN FALSE THEN %s END",
			cond, then, elseVal,
		), nil
	}
	return fmt.Sprintf("CASE WHEN %s THEN %s ELSE %s END", cond, then, elseVal), nil
}

func compileCase(n *CaseExpr, ctx *CompileContext) (string, error) {
	// A CASE SUBJECT is never made strict, unlike an IF condition. Separating
	// "the subject is NULL" from "the subject matched no branch" needs a second
	// `(expr) IS NULL` test alongside the subject itself, and that second
	// substitution is what makes a gated CASE nested inside another gated CASE
	// subject double at every level. compileIf escapes this because its
	// condition is boolean-shaped, so one substitution covers all three values;
	// no such form exists for a simple-CASE subject. The subject position
	// therefore keeps the absorbing emission: no strictness, byte-identical to
	// pre-gate behavior, linear. Pinned by TestCompileCaseSubjectStaysAbsorbing.
	expr, err := compile(n.Expr, ctx)
	if err != nil {
		return "", err
	}

	var sb strings.Builder
	sb.WriteString("CASE ")
	sb.WriteString(expr)

	for _, b := range n.Branches {
		val, err := compile(b.Value, ctx)
		if err != nil {
			return "", err
		}
		result, err := compile(b.Result, ctx)
		if err != nil {
			return "", err
		}
		sb.WriteString(" WHEN ")
		sb.WriteString(val)
		sb.WriteString(" THEN ")
		sb.WriteString(result)
	}

	if n.Default != nil {
		def, err := compile(n.Default, ctx)
		if err != nil {
			return "", err
		}
		sb.WriteString(" ELSE ")
		sb.WriteString(def)
	}

	sb.WriteString(" END")
	return sb.String(), nil
}

// compileAggregate generates a correlated subquery for rollup aggregates.
// In the View DDL builder, these get hoisted into LEFT JOIN subqueries.
// For the expression compiler, we emit a placeholder that the DDL builder
// can recognize and transform.
func compileAggregate(n *AggregateExpr, ctx *CompileContext) (string, error) {
	// Emit a marker that the DDL builder will replace with a proper subquery JOIN.
	// Format: ##AGG##<FUNC>##<ChildObject>##<ChildField>##<ForeignKey>##
	fk := n.ForeignKey
	if fk == "" {
		fk = "_"
	}
	field := n.ChildField
	if field == "" {
		field = "*"
	}
	return fmt.Sprintf("%s%s##%s##%s##%s##", AggMarkerPrefix, n.Func, n.ChildObject, field, fk), nil
}