package analytics

import (
	"context"
	"strings"
	"testing"

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

// Tests for null provenance: undefined arithmetic (Salesforce #Error) must propagate out
// of an IF/CASE branch-selecting boundary instead of being absorbed as "false".
//
// Every assertion below is on EMITTED SQL. None inspects compiler source text —
// a source-regex pin would survive the mutation that matters (see the mutation
// table in the PR body) and would be defeated by any rename.
//
// The strictness marker in the emitted SQL. Its presence/absence is the single
// observable that says whether the provenance gate fired.
const nullStrictGuardMarker = "WHEN TRUE THEN"

// nullProvenanceFieldTypes is the shared field-type fixture. The two doubles are
// the null provenance shape (both sides of the ratio blank-as-zero coalesced by #4152, so
// the ONLY NULL a division over them can produce is 0/0). Every other entry is a
// leaf kind that can still carry raw SQL NULL and must therefore keep the gate
// shut.
func nullProvenanceFieldTypes() map[string]string {
	return map[string]string{
		"Num__c":                "double",
		"Den__c":                "double",
		"Num2__c":               "double",
		"Alt__c":                "double",
		"Prepaid__c":            "double",
		"MinCommit__c":          "double",
		"Years__c":              "double",
		"Amount":                "currency",
		"Roll__c":               "rollupsummary",
		"Str__c":                "string",
		"Pick__c":               "picklist",
		"Product_Family__c":     "picklist",
		"Sales_Channel_Type__c": "string",
		"Bool__c":               "boolean",
		"Due__c":                "date",
		"When__c":               "datetime",
		"Tags__c":               "multipicklist",
		"Stage":                 "picklist",
	}
}

func compileNullProvenanceSQL(t *testing.T, expr string) string {
	t.Helper()
	ast, err := Parse(expr)
	if err != nil {
		t.Fatalf("Parse(%q) error = %v", expr, err)
	}
	sql, err := Compile(ast, &CompileContext{
		TableAlias:    "t",
		QualifyFields: true,
		JoinAliases:   map[string]string{},
		FieldTypes:    nullProvenanceFieldTypes(),
	})
	if err != nil {
		t.Fatalf("Compile(%q) error = %v", expr, err)
	}
	return sql
}

// tieredBonusFormula is the verbatim Quote.Bonus_Rate__c expression from
// the merged pack — the field that displays an unearned 3% (Direct) / 1.5%
// (Indirect) tier on production quotes whose prepayment denominator
// MinCommit__c is NULL or 0.
const tieredBonusFormula = `IF(OR(TEXT(Product_Family__c)="TiDB Cloud",TEXT(Product_Family__c)="Tier_A"),` +
	`IF(Sales_Channel_Type__c="Direct",` +
	`IF(Years__c=2,0.03,IF(Years__c>=3,0.05,0))` +
	`+IF(Prepaid__c/MinCommit__c=1,0.05,` +
	`IF(Prepaid__c/MinCommit__c<0.5,0,0.03)),` +
	`IF(Years__c=2,0.015,IF(Years__c>=3,0.025,0))` +
	`+IF(Prepaid__c/MinCommit__c=1,0.025,` +
	`IF(Prepaid__c/MinCommit__c<0.5,0,0.015))),NULL)`

// T1 — the bug itself, on the real pack formula.
//
// Byte-exact so the shape cannot drift: the two prepay-ratio CASEs acquire the
// NULL guard, and — equally load-bearing — the outer product-family CASE, the
// middle channel CASE, the year-term CASEs and the literal `ELSE NULL END` are
// untouched. A test that only asserted "contains WHEN TRUE THEN" would pass
// under a globally-strict compiler, which is exactly the overreach this PR is
// engineered to avoid.
func TestCompileDivisionInConditionPropagatesUndefinedArithmetic(t *testing.T) {
	got := compileNullProvenanceSQL(t, tieredBonusFormula)

	ratio := "(COALESCE(t.`Prepaid__c`, 0) / COALESCE(t.`MinCommit__c`, 0))"
	yearTerm := func(two, three string) string {
		return "CASE WHEN (COALESCE(t.`Years__c`, 0) = 2) THEN " + two +
			" ELSE CASE WHEN (COALESCE(t.`Years__c`, 0) >= 3) THEN " + three +
			" ELSE 0 END END"
	}
	ratioTerm := func(full, mid string) string {
		return "CASE ((" + ratio + " = 1)) " + nullStrictGuardMarker + " " + full +
			" WHEN FALSE THEN CASE ((" + ratio + " < 0.5)) " + nullStrictGuardMarker +
			" 0 WHEN FALSE THEN " + mid + " END END"
	}
	want := "CASE WHEN ((COALESCE(t.`Product_Family__c`, '') = 'TiDB Cloud')" +
		" OR (COALESCE(t.`Product_Family__c`, '') = 'Tier_A'))" +
		" THEN CASE WHEN (t.`Sales_Channel_Type__c` = 'Direct')" +
		" THEN (" + yearTerm("0.03", "0.05") + " + " + ratioTerm("0.05", "0.03") + ")" +
		" ELSE (" + yearTerm("0.015", "0.025") + " + " + ratioTerm("0.025", "0.015") + ") END" +
		" ELSE NULL END"

	if got != want {
		t.Fatalf("null provenance formula SQL mismatch:\n got: %s\nwant: %s", got, want)
	}

	// Structural restatement of the three facts above, so a future failure
	// message names which one broke rather than dumping 1.5KB of SQL.
	if strings.Count(got, nullStrictGuardMarker) != 4 {
		t.Fatalf("expected exactly 4 strict guards (2 ratio CASEs x 2 channel arms), got %d in:\n%s",
			strings.Count(got, nullStrictGuardMarker), got)
	}
	if strings.Contains(got, "CASE WHEN ((t.`Sales_Channel_Type__c` = 'Direct')) "+nullStrictGuardMarker) {
		t.Fatalf("channel condition must NOT be strict (blank text is FALSE in Salesforce, not #Error):\n%s", got)
	}
	if !strings.HasSuffix(got, " ELSE NULL END") {
		t.Fatalf("literal-NULL else arm must survive verbatim:\n%s", got)
	}
}

// T2 — THE ANTI-OVERREACH GUARD. The most important test in this PR.
//
// Each row puts a field-denominator division in the SAME condition as a leaf
// that can still carry raw SQL NULL for a reason OTHER than undefined
// arithmetic. Salesforce treats those leaves as blank: the comparison is FALSE
// and the else arm is selected. If the gate fired here the formula would go
// blank instead — a silent wrong answer in the OPPOSITE direction from null provenance.
//
// This is the only test that fails when the blank-provenance half of the gate
// is deleted. Neither the 83-row byte-parity golden, nor the rest of the
// analytics suite, nor the merged-pack corpus detects that mutation, because no
// pack formula today mixes a blank-capable leaf with a division inside one
// condition. Do not weaken or delete these rows.
// Every row is BYTE-PINNED to the string HEAD emits, not merely checked for the
// absence of the guard marker. Absence-of-substring is too weak: a refactor that
// reshaped the strict emission and updated nullStrictGuardMarker alongside it
// would leave all these rows passing while asserting nothing. "Unchanged" has to
// mean literally unchanged.
func TestCompileGateRefusesWhenConditionHasBlankCapableLeaf(t *testing.T) {
	div := "((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) > 0.5)"
	rows := []struct {
		name string
		expr string
		want string
		why  string
	}{
		{"string_leaf_and", `IF(AND(Str__c = "Direct", Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN ((t.`Str__c` = 'Direct') AND " + div + ") THEN 1 ELSE 2 END",
			"blank text compares FALSE in Salesforce, it does not raise #Error"},
		{"boolean_leaf_and", `IF(AND(Bool__c, Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN (t.`Bool__c` AND " + div + ") THEN 1 ELSE 2 END",
			"a NULL boolean column is falsy, not #Error"},
		{"date_leaf_and", `IF(AND(Due__c > TODAY(), Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN ((t.`Due__c` > CURDATE()) AND " + div + ") THEN 1 ELSE 2 END",
			"a blank date comparison is FALSE, not #Error"},
		{"datetime_leaf_and", `IF(AND(When__c > NOW(), Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN ((t.`When__c` > NOW()) AND " + div + ") THEN 1 ELSE 2 END",
			"a blank datetime comparison is FALSE, not #Error"},
		// The two rows above route through TODAY()/NOW(), which are *FuncCall
		// nodes and therefore close the gate via the blanket function rule
		// regardless of what the date LEAF does. Measured: they stay green when
		// the leaf blank mark is deleted, so they do not test what their names
		// suggest. This function-free row is the one that actually pins date and
		// datetime leaf blankness.
		{"function_free_date_leaves", `IF(AND(Due__c > When__c, Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN ((t.`Due__c` > t.`When__c`) AND " + div + ") THEN 1 ELSE 2 END",
			"date/datetime columns are not coalesced by #4152, so either side can be NULL"},
		{"rollupsummary_leaf", `IF(Num__c/Den__c - Roll__c > 0.5, 1, 2)`,
			"CASE WHEN (((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) - t.`Roll__c`) > 0.5) THEN 1 ELSE 2 END",
			"rollupsummary bypasses #4152 blank-as-zero, so the column can still be NULL"},
		{"bare_string_truthiness", `IF(AND(Str__c, Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN (t.`Str__c` AND " + div + ") THEN 1 ELSE 2 END",
			"a bare string leaf used for truthiness is still NULL-capable"},
		{"trailing_string_inequality", `IF(Num__c/Den__c > 0.5 && Str__c != "x", 1, 2)`,
			"CASE WHEN (" + div + " AND (t.`Str__c` != 'x')) THEN 1 ELSE 2 END",
			"gate must be position-independent: the blank leaf comes AFTER the division"},
		{"function_call_wrapper", `IF(ROUND(Num__c/Den__c, 2) > 0.5, 1, 2)`,
			"CASE WHEN (ROUND((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)), 2) > 0.5) THEN 1 ELSE 2 END",
			"function lowering is not modelled, so it fails closed"},
		// A rollup subquery yields NULL for an empty child set — blankness, not
		// #Error. Without the *AggregateExpr blank mark this emission acquires
		// the guard and an empty child set blanks the whole formula. Note this
		// is a DIFFERENT mechanism from rollupsummary_leaf above, which is a
		// field TYPE routed through fieldReadSQL.
		{"aggregate_in_condition", `IF(SUM(Child__c.Amt__c) / Den__c > 1, 1, 2)`,
			"CASE WHEN ((##AGG##SUM##Child__c##Amt__c##_## / COALESCE(t.`Den__c`, 0)) > 1) THEN 1 ELSE 2 END",
			"an empty child set makes a rollup subquery NULL, which is blank, not #Error"},
		{"literal_null_in_condition", `IF(Num__c/Den__c = NULL, 1, 2)`,
			"CASE WHEN ((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) IS NULL) THEN 1 ELSE 2 END",
			"an authored NULL literal is a blank, and it is what the condition tests for"},
		// The row above routes through compileNullComparison, which closes the
		// gate on its own, so it does NOT exercise the *NullLit mark. This row
		// does: a bare NULL operand is the only blank site in the condition, and
		// the AND genuinely evaluates to NULL (`TRUE AND NULL` is NULL) for a
		// reason that is authored blankness, not undefined arithmetic.
		{"literal_null_operand", `IF(AND(Num__c/Den__c > 0.5, NULL), 1, 2)`,
			"CASE WHEN (" + div + " AND NULL) THEN 1 ELSE 2 END",
			"a bare authored NULL is blankness; the gate must not read it as #Error"},
		// The Salesforce empty-string lowering produces the same total boolean
		// as the `= NULL` row above and must therefore emit the same SQL. Before
		// the blank mark on that special case it acquired a guard TiDB can never
		// take (`((0/0) IS NULL) IS NULL` -> 0).
		{"empty_string_comparison", `IF(Num__c/Den__c = "", 1, 2)`,
			"CASE WHEN ((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) IS NULL) THEN 1 ELSE 2 END",
			"a total boolean can never be NULL, so a guard on it would be dead SQL"},
		{"untyped_field_leaf", `IF(Unknown__c/Den__c > 0.5, 1, 2)`,
			"CASE WHEN ((t.`Unknown__c` / COALESCE(t.`Den__c`, 0)) > 0.5) THEN 1 ELSE 2 END",
			"an unknown field type is not coalesced by #4152, so the numerator can be NULL"},
		// An even-arity CASE leaves Default nil and emits no ELSE, so the node
		// MANUFACTURES NULL on a no-match out of operands that are all
		// blank-safe. Every leaf here is a #4152-coalesced double, so before the
		// nil-Default mark the gate read the whole condition as pure #Error and
		// fired. Measured at runtime with Den__c = 100 (a TOTAL division) and
		// Num2__c = 3 (no branch matches): the row went 222 -> blank. The
		// division is well-defined; nothing here is #Error.
		{"case_without_else_arm", `IF(CASE(Num2__c, 1, 10, 2, 20) + Num__c/Den__c > 0, 111, 222)`,
			"CASE WHEN ((CASE COALESCE(t.`Num2__c`, 0) WHEN 1 THEN 10 WHEN 2 THEN 20 END" +
				" + (COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0))) > 0) THEN 111 ELSE 222 END",
			"a no-match CASE with no ELSE returns NULL, which is an absent value, not #Error"},
		// The two rows below pin the REACH of the blank-comparison retraction in
		// compileBinaryOp, which is wider than the node it fires on: the gate's
		// blank test spans the whole condition, so a `= ""` anywhere closes it
		// for every division in that condition. Deliberate and fail-closed (the
		// node-scoped alternative would desynchronise `= ""` from `= NULL`), but
		// it is behavior, so it is pinned rather than left to a comment.
		{"blank_comparison_retraction_reaches_sibling_division",
			`IF(AND(Num__c/Den__c > 0.5, Str__c = ""), 1, 2)`,
			"CASE WHEN (" + div + " AND (IFNULL(t.`Str__c` = '', TRUE))) THEN 1 ELSE 2 END",
			"an unrelated `= \"\"` closes the gate for the whole condition, by design"},
		{"blank_comparison_retraction_across_two_divisions",
			`IF(AND(Num2__c/Alt__c = "", Num__c/Den__c > 0.5), 1, 2)`,
			"CASE WHEN (((COALESCE(t.`Num2__c`, 0) / COALESCE(t.`Alt__c`, 0)) IS NULL) AND " +
				div + ") THEN 1 ELSE 2 END",
			"the second division keeps no guard either; retraction is condition-wide"},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) {
			got := compileNullProvenanceSQL(t, row.expr)
			if strings.Contains(got, nullStrictGuardMarker) {
				t.Fatalf("gate must stay shut (%s):\n%s", row.why, got)
			}
			if got != row.want {
				t.Fatalf("emission drifted (%s):\n got: %s\nwant: %s", row.why, got, row.want)
			}
		})
	}
}

// T3 — definedDivisor. A non-zero number literal cannot make a quotient
// undefined, so it must not enable strictness; a literal zero must.
func TestCompileLiteralDenominatorStaysAbsorbing(t *testing.T) {
	nonZero := compileNullProvenanceSQL(t, `IF(Num__c/10 > 0.5, 1, 2)`)
	if strings.Contains(nonZero, nullStrictGuardMarker) {
		t.Fatalf("non-zero literal denominator must not enable strictness:\n%s", nonZero)
	}
	want := "CASE WHEN ((COALESCE(t.`Num__c`, 0) / 10) > 0.5) THEN 1 ELSE 2 END"
	if nonZero != want {
		t.Fatalf("literal-denominator SQL = %q, want %q", nonZero, want)
	}

	zero := compileNullProvenanceSQL(t, `IF(Num__c/0 > 0.5, 1, 2)`)
	if !strings.Contains(zero, nullStrictGuardMarker) {
		t.Fatalf("literal ZERO denominator is always undefined and must be strict:\n%s", zero)
	}

	// Sign-prefixed literals. The lexer does not fold the sign, so `/-2` arrives
	// as *UnaryOp{"-", NumberLit{2}}; without unwrapping it, definedDivisor said
	// "not a literal" and the emission acquired a guard on
	// `COALESCE(x,0) / (-2)`, which can never be NULL — dead SQL and needless
	// golden churn. `-0` must still be treated as a zero divisor.
	for _, row := range []struct {
		name   string
		expr   string
		want   string
		strict bool
	}{
		{"negative_literal_denominator", `IF(Num__c/-2 > 0.5, 1, 2)`,
			"CASE WHEN ((COALESCE(t.`Num__c`, 0) / (-2)) > 0.5) THEN 1 ELSE 2 END", false},
		{"parenthesised_negative_literal", `IF(Num__c/(-2) > 0.5, 1, 2)`,
			"CASE WHEN ((COALESCE(t.`Num__c`, 0) / (-2)) > 0.5) THEN 1 ELSE 2 END", false},
		{"negative_zero_literal_is_still_zero", `IF(Num__c/-0 > 0.5, 1, 2)`, "", true},
	} {
		t.Run(row.name, func(t *testing.T) {
			got := compileNullProvenanceSQL(t, row.expr)
			if strings.Contains(got, nullStrictGuardMarker) != row.strict {
				t.Fatalf("strict = %v, want %v:\n%s", !row.strict, row.strict, got)
			}
			if row.want != "" && got != row.want {
				t.Fatalf("emission drifted:\n got: %s\nwant: %s", got, row.want)
			}
		})
	}
}

// The nil-Default CASE distinction, from the strictness side.
//
// TestCompileGateRefusesWhenConditionHasBlankCapableLeaf/case_without_else_arm
// owns the refusal. This is the POSITIVE control that keeps the fix from being
// implemented as "any CASE in a condition closes the gate": adding an else-arm
// makes the CASE total, at which point the division really is the only NULL
// source in the condition and the guard must come back.
//
// The two spellings differ only in what the else-arm IS, which is the point —
// `, 0)` is total, `, NULL)` authors a blank, and no else-arm at all manufactures
// one. Before the fix the first and third behaved identically, so the same
// semantic written two ways got two answers.
func TestCompileCaseElseArmDecidesStrictness(t *testing.T) {
	body := "CASE COALESCE(t.`Num2__c`, 0) WHEN 1 THEN 10 WHEN 2 THEN 20"
	div := "(COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0))"

	total := compileNullProvenanceSQL(t, `IF(CASE(Num2__c, 1, 10, 2, 20, 0) + Num__c/Den__c > 0, 111, 222)`)
	cond := "((" + body + " ELSE 0 END + " + div + ") > 0)"
	want := "CASE (" + cond + ") " + nullStrictGuardMarker + " 111 WHEN FALSE THEN 222 END"
	if total != want {
		t.Fatalf("a CASE with a total else-arm must not close the gate:\n got: %s\nwant: %s", total, want)
	}

	// An authored NULL else-arm is blankness and must close the gate, exactly
	// like the no-else spelling now does.
	authored := compileNullProvenanceSQL(t, `IF(CASE(Num2__c, 1, 10, 2, 20, NULL) + Num__c/Den__c > 0, 111, 222)`)
	if strings.Contains(authored, nullStrictGuardMarker) {
		t.Fatalf("an authored NULL else-arm is blank, not #Error:\n%s", authored)
	}
	if authored != "CASE WHEN (("+body+" ELSE NULL END + "+div+") > 0) THEN 111 ELSE 222 END" {
		t.Fatalf("authored-NULL emission drifted:\n%s", authored)
	}
}

// T3b — the `^` arith site. POWER is the second undefined-arithmetic operator
// and, like division, it must open the gate.
//
// Honest caveat, measured on TiDB v8.5.0 and stated in the compiler comment:
// the guard this produces is currently UNREACHABLE, because an undefined POWER
// aborts the statement (`SELECT POWER(0,-1)` -> ERROR 1690) rather than
// returning NULL. The mark is kept because it is the conservative direction and
// because the engine could gain NULL-on-overflow semantics later. This test
// exists so the mark cannot be deleted silently: without it the emission below
// loses its guard and the entire analytics suite stays green.
func TestCompileExponentMarksUndefinedArithmetic(t *testing.T) {
	power := "(POWER(COALESCE(t.`Num__c`, 0), COALESCE(t.`Den__c`, 0)) > 1)"
	got := compileNullProvenanceSQL(t, `IF(Num__c ^ Den__c > 1, 1, 2)`)
	want := "CASE (" + power + ") " + nullStrictGuardMarker + " 1 WHEN FALSE THEN 2 END"
	if got != want {
		t.Fatalf("exponent strictness:\n got: %s\nwant: %s", got, want)
	}
}

// T4 — accidental global strictness. Every one of these is byte-pinned to the
// string HEAD emits; the whole set is what would break if someone "simplified"
// the gate to "always strict".
func TestCompileNonArithmeticConditionsAreByteIdentical(t *testing.T) {
	rows := []struct {
		name string
		expr string
		want string
	}{
		{"numeric_comparison", `IF(Amount > 0, 1, 2)`,
			"CASE WHEN (COALESCE(t.`Amount`, 0) > 0) THEN 1 ELSE 2 END"},
		{"bare_boolean", `IF(Bool__c, 1, 2)`,
			"CASE WHEN t.`Bool__c` THEN 1 ELSE 2 END"},
		{"date_comparison", `IF(Due__c > TODAY(), 1, 2)`,
			"CASE WHEN (t.`Due__c` > CURDATE()) THEN 1 ELSE 2 END"},
		{"rollup_comparison", `IF(Roll__c > 0, 1, 2)`,
			"CASE WHEN (t.`Roll__c` > 0) THEN 1 ELSE 2 END"},
		{"picklist_equality", `IF(Pick__c = "Direct", 1, 2)`,
			"CASE WHEN (t.`Pick__c` = 'Direct') THEN 1 ELSE 2 END"},
		{"negated_equality", `IF(NOT(Str__c = "Direct"), 1, 2)`,
			"CASE WHEN (NOT (t.`Str__c` = 'Direct')) THEN 1 ELSE 2 END"},
		{"inequality", `IF(Str__c != "Direct", 1, 2)`,
			"CASE WHEN (t.`Str__c` != 'Direct') THEN 1 ELSE 2 END"},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) {
			if got := compileNullProvenanceSQL(t, row.expr); got != row.want {
				t.Fatalf("emission drifted:\n got: %s\nwant: %s", got, row.want)
			}
		})
	}
}

// T6 — the delta window must cover the CONDITION only. An #Error in a branch
// VALUE already propagates correctly today (the CASE returns the NULL it
// selected), so widening the window would churn this shape for no gain.
func TestCompileValuePositionDivisionUnchanged(t *testing.T) {
	got := compileNullProvenanceSQL(t, `IF(Amount > 0, Num__c/Den__c, 0)`)
	want := "CASE WHEN (COALESCE(t.`Amount`, 0) > 0) THEN (COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) ELSE 0 END"
	if got != want {
		t.Fatalf("value-position division must be untouched:\n got: %s\nwant: %s", got, want)
	}
}

// TestCompileCaseSubjectStaysAbsorbing pins a DELIBERATE non-fix.
//
// A simple-CASE subject cannot be substituted once and still separate "the
// subject is NULL" from "the subject matched no branch" — both yield NULL from
// `CASE expr WHEN v THEN r END`, and telling them apart needs a second
// `(expr) IS NULL` test. That second substitution is what makes a gated CASE
// nested inside another gated CASE subject double at every level: measured
// 505,628 characters of emitted SQL from a 290-character formula ten deep,
// against 1,295 with this fallback. compileIf has no such problem because its
// condition is boolean-shaped and one substitution covers all three values.
//
// So the subject position keeps the absorbing emission: no strictness, and
// byte-identical to pre-gate behavior. If this ever needs fixing, it needs a
// mechanism that binds the subject once, not a third substitution.
func TestCompileCaseSubjectStaysAbsorbing(t *testing.T) {
	ratio := "(COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0))"
	got := compileNullProvenanceSQL(t, `CASE(Num__c/Den__c, 1, 10, 2, 20, 99)`)
	want := "CASE " + ratio + " WHEN 1 THEN 10 WHEN 2 THEN 20 ELSE 99 END"
	if got != want {
		t.Fatalf("CASE subject must stay absorbing:\n got: %s\nwant: %s", got, want)
	}
	if strings.Contains(got, nullStrictGuardMarker) {
		t.Fatalf("CASE subject must not acquire a strict guard:\n%s", got)
	}

	for _, row := range []struct{ name, expr, want string }{
		{"literal_denominator_subject", `CASE(Num__c/10, 1, 10, 2, 20, 99)`,
			"CASE (COALESCE(t.`Num__c`, 0) / 10) WHEN 1 THEN 10 WHEN 2 THEN 20 ELSE 99 END"},
		{"picklist_subject", `CASE(Stage, "a", 1, "b", 2, 99)`,
			"CASE t.`Stage` WHEN 'a' THEN 1 WHEN 'b' THEN 2 ELSE 99 END"},
	} {
		t.Run(row.name, func(t *testing.T) {
			if got := compileNullProvenanceSQL(t, row.expr); got != row.want {
				t.Fatalf("CASE emission drifted:\n got: %s\nwant: %s", got, row.want)
			}
		})
	}
}

// T8 — the nested-division DECIMAL fence.
//
// A division whose subtree contains another division is emitted under a
// BY-VALUE COPY of the compile context (`cloned := *ctx`) selected before its
// operands are compiled. Two things have to hold for the gate to work there,
// and each row below isolates one of them.
func TestCompileNestedDivisionMarksFencedPath(t *testing.T) {
	// Row 1 — the realistic shape: both divisions have field divisors, so the
		// fence and the guard have to coexist in one emission. This is the broad
		// end-to-end shape check; row 2 isolates the fenced return's own mark.
	t.Run("fence_and_guard_coexist", func(t *testing.T) {
		got := compileNullProvenanceSQL(t, `IF((Num__c/Den__c)/Den__c > 1, 1, 2)`)
		if !strings.Contains(got, "CAST(COALESCE(t.`Num__c`, 0) AS DECIMAL(65,30))") {
			t.Fatalf("expected the nested-division DECIMAL fence in the emission:\n%s", got)
		}
		if !strings.Contains(got, nullStrictGuardMarker) {
			t.Fatalf("fenced division must still mark undefined arithmetic:\n%s", got)
		}
	})

	// Row 2 — THE DISCRIMINATOR. Here the mark on the FENCED return path is the
	// only one that can fire: the inner division has a non-zero literal divisor
		// so it never marks, and the outer division's plain return path is
		// unreachable (the bottom-up analysis selects the fence first).
	//
	// Two distinct mutations were measured to slip past row 1 and be caught
	// only here: deleting the mark on the fenced return, and allocating fresh
	// counters on the `cloned := *ctx` copy (i.e. what would happen if the
	// counters were plain ints rather than pointers).
	t.Run("only_the_fenced_return_can_mark", func(t *testing.T) {
		got := compileNullProvenanceSQL(t, `IF((Num__c/2)/Den__c > 1, 1, 2)`)
		if !strings.Contains(got, "AS DECIMAL(65,30)") {
			t.Fatalf("expected the nested-division DECIMAL fence in the emission:\n%s", got)
		}
		if !strings.Contains(got, nullStrictGuardMarker) {
			t.Fatalf("the outer field divisor is only reachable through the fenced "+
				"return, so that return must mark undefined arithmetic:\n%s", got)
		}
	})
}

// T9 — the gate must survive every way a CompileContext reaches the compiler.
func TestCompileGateSurvivesContextCopies(t *testing.T) {
	strictExpr := `IF(Num__c/Den__c > 0.5, 1, 2)`

	t.Run("nil_context", func(t *testing.T) {
		// With a nil context no field types are known, so the reads are NOT
		// coalesced and the gate correctly refuses. What is asserted here is
		// that Compile(ast, nil) still seeds the counters and does not panic.
		ast, err := Parse(strictExpr)
		if err != nil {
			t.Fatalf("Parse error = %v", err)
		}
		got, err := Compile(ast, nil)
		if err != nil {
			t.Fatalf("Compile(ast, nil) error = %v", err)
		}
		if strings.Contains(got, nullStrictGuardMarker) {
			t.Fatalf("untyped reads are NULL-capable, gate must stay shut:\n%s", got)
		}
	})

	t.Run("caller_built_context_with_nil_counters", func(t *testing.T) {
		// The realistic production shape: every caller constructs a
		// CompileContext literal, so the counters are nil on entry. Compile()
		// must seed them, otherwise the gate silently never fires anywhere.
		ast, err := Parse(strictExpr)
		if err != nil {
			t.Fatalf("Parse error = %v", err)
		}
		got, err := Compile(ast, &CompileContext{
			TableAlias:    "t",
			QualifyFields: true,
			FieldTypes:    nullProvenanceFieldTypes(),
		})
		if err != nil {
			t.Fatalf("Compile error = %v", err)
		}
		if !strings.Contains(got, nullStrictGuardMarker) {
			t.Fatalf("caller-built context must still gate:\n%s", got)
		}
	})

	t.Run("executable_bundle_body_fails_closed", func(t *testing.T) {
		// An L1 formula_expr_v1 bundle function whose BODY divides by a bound
		// parameter, called from inside an IF condition. The function-call node
		// marks a blank site unconditionally, so the gate refuses.
		//
		// This is the deliberately blunt FuncCall rule: NOTHING about a
		// function's lowering is modelled, so it fails closed to today's
		// emission. Registering a new function can therefore only ever CLOSE
		// the gate, never silently widen it. If someone adds a "provably
		// non-null functions" allowlist, this row is what catches it.
		bundle, err := formula.ParseFunctionBundlePayload([]byte(`{
			"functions": [
				{
					"name": "SAFE_RATIO",
					"minArgs": 2,
					"maxArgs": 2,
					"return": { "kind": "fixed", "type": "number" },
					"args": [
						{ "type": "number", "nullable": true },
						{ "type": "number", "nullable": true }
					],
					"evaluator": {
						"kind": "formula_expr_v1",
						"params": ["num", "den"],
						"body": "num / den"
					}
				}
			]
		}`))
		if err != nil {
			t.Fatalf("ParseFunctionBundlePayload: %v", err)
		}
		registry, err := formula.ExecutableFunctionRegistryWithBundles([]formula.FunctionBundle{bundle})
		if err != nil {
			t.Fatalf("ExecutableFunctionRegistryWithBundles: %v", err)
		}

		ast, err := Parse(`IF(SAFE_RATIO(Num__c, Den__c) > 0.5, 1, 2)`)
		if err != nil {
			t.Fatalf("Parse error = %v", err)
		}
		got, err := Compile(ast, &CompileContext{
			TableAlias:          "t",
			QualifyFields:       true,
			FieldTypes:          nullProvenanceFieldTypes(),
			FunctionExecutables: registry,
		})
		if err != nil {
			t.Fatalf("Compile error = %v", err)
		}
		if !strings.Contains(got, "/") {
			t.Fatalf("expected the bundle body's division to be lowered inline:\n%s", got)
		}
		if strings.Contains(got, nullStrictGuardMarker) {
			t.Fatalf("function calls fail closed; gate must stay shut:\n%s", got)
		}
	})

	t.Run("bundle_body_own_if_fails_closed", func(t *testing.T) {
		// The row above puts the division in the CALLER's condition, so the
		// *FuncCall blank mark fires first and masks everything underneath it.
		// Here the IF lives INSIDE the bundle body, so its delta window opens
		// after that mark and the only thing keeping the gate shut is the
		// bound-identifier rule: `num` and `den` resolve to SQL the caller
		// already compiled in another context, whose NULL provenance is
		// unknowable from here.
		//
		// Measured: deleting that mark flips this emission to the strict shape
		// while the whole analytics suite otherwise stays green. This is the
		// only test that sees it.
		bundle, err := formula.ParseFunctionBundlePayload([]byte(`{
			"functions": [
				{
					"name": "RATIO_FLAG",
					"minArgs": 2,
					"maxArgs": 2,
					"return": { "kind": "fixed", "type": "number" },
					"args": [
						{ "type": "number", "nullable": true },
						{ "type": "number", "nullable": true }
					],
					"evaluator": {
						"kind": "formula_expr_v1",
						"params": ["num", "den"],
						"body": "IF(num / den > 0.5, 1, 2)"
					}
				}
			]
		}`))
		if err != nil {
			t.Fatalf("ParseFunctionBundlePayload: %v", err)
		}
		registry, err := formula.ExecutableFunctionRegistryWithBundles([]formula.FunctionBundle{bundle})
		if err != nil {
			t.Fatalf("ExecutableFunctionRegistryWithBundles: %v", err)
		}

		ast, err := Parse(`RATIO_FLAG(Num__c, Den__c)`)
		if err != nil {
			t.Fatalf("Parse error = %v", err)
		}
		got, err := Compile(ast, &CompileContext{
			TableAlias:          "t",
			QualifyFields:       true,
			FieldTypes:          nullProvenanceFieldTypes(),
			FunctionExecutables: registry,
		})
		if err != nil {
			t.Fatalf("Compile error = %v", err)
		}
		want := "CASE WHEN ((COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) > 0.5) THEN 1 ELSE 2 END"
		if got != want {
			t.Fatalf("bundle body IF must fail closed on bound identifiers:\n got: %s\nwant: %s", got, want)
		}
	})
}

// T12 — the save path, which is the ONLY consumer whose semantics change the
// moment the binary lands.
//
// Read paths go through the materialized `v_<object>` DDL, and EnsureViews
// (formula_view_service.go) only builds views that are MISSING, so a restart
// does not rebuild v_quote; the read side keeps the old stored SQL until
// RebuildAllFormulaViews runs. resolveActiveFormulaValue is different: it
// recompiles from the binary on every save and wraps the result in its own
// SELECT, with no view involved. So it flips immediately, on its own, before any
// rebuild — which is exactly the split-brain window the rollout note describes.
//
// This asserts the strict guard reaches that SELECT. Nothing else in the PR
// exercises this path: every other compile-level test builds a CompileContext
// literal by hand and the integration test hand-writes its own SELECT, so a
// regression that dropped strictness only here would go unnoticed.
func TestValidationActiveFormulaResolverPropagatesUndefinedArithmetic(t *testing.T) {
	mode := "active"
	percentReturnType := validationFormulaFieldTypePtr(constants.FieldTypePercent)

	schema := &models.ObjectMetadata{
		APIName: "Quote",
		Fields: []models.FieldMetadata{
			{APIName: "Prepaid__c", Type: constants.FieldTypeDouble},
			{APIName: "MinCommit__c", Type: constants.FieldTypeDouble},
			{
				APIName:     "Bonus_Rate__c",
				Type:        constants.FieldTypeFormula,
				Formula:     validationFormulaStringPtr(`IF(Prepaid__c/MinCommit__c=1,0.05,IF(Prepaid__c/MinCommit__c<0.5,0,0.03))`),
				FormulaMode: &mode,
				ReturnType:  percentReturnType,
			},
		},
	}

	viewSvc := newValidationFormulaViewService(t, schema)
	queryExec := &validationFormulaQueryExecStub{rows: []models.SObject{{"Value": nil}}}
	resolver := NewValidationActiveFormulaResolver(viewSvc, queryExec)

	if _, err := resolver(context.Background(), schema, models.SObject{
		"Prepaid__c":   0,
		"MinCommit__c": 0,
	}, &schema.Fields[2]); err != nil {
		t.Fatalf("resolver error = %v", err)
	}

	ratio := "(COALESCE(t.`Prepaid__c`, 0) / COALESCE(t.`MinCommit__c`, 0))"
	inner := "CASE ((" + ratio + " < 0.5)) " + nullStrictGuardMarker + " 0 WHEN FALSE THEN 0.03 END"
	wantExpr := "SELECT (CASE ((" + ratio + " = 1)) " + nullStrictGuardMarker +
		" 0.05 WHEN FALSE THEN " + inner + " END) * 100 AS `Value`"
	if !strings.Contains(queryExec.lastSQL, wantExpr) {
		t.Fatalf("save-path SQL lost the null provenance guard:\n got: %s\nwant substring: %s",
			queryExec.lastSQL, wantExpr)
	}
	if strings.Count(queryExec.lastSQL, nullStrictGuardMarker) != 2 {
		t.Fatalf("expected exactly 2 guards on the save path, got %d in:\n%s",
			strings.Count(queryExec.lastSQL, nullStrictGuardMarker), queryExec.lastSQL)
	}
}

// TestCompileStrictEmissionStaysLinear is the regression guard for the review
// finding that produced the single-substitution emission.
//
// The first revision emitted `CASE WHEN (c) IS NULL THEN NULL WHEN c THEN ..`,
// naming the condition twice, so a gated IF nested inside another gated IF's
// CONDITION doubled the emitted text at every level. Measured then: 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 an admin-authored formula could make a view
// rebuild allocate hundreds of megabytes.
//
// Asserting a ratio rather than an absolute size keeps this honest under future
// formatting changes: doubling per level would put depth 14 at 128x depth 7,
// so any small constant factor falsifies exponential growth.
func TestCompileStrictEmissionStaysLinear(t *testing.T) {
	// Nest through BOTH slots a doubling could hide in. The first revision
	// doubled the CONDITION; a later reshape could just as easily double a
	// branch VALUE, and a guard that only nests conditions would not see it.
	shapes := map[string]func(int) string{
		"condition_slot": func(depth int) string {
			inner := "Num__c/Den__c"
			for i := 0; i < depth; i++ {
				inner = "IF(" + inner + " > 1, 1, 2)/Den__c"
			}
			return "IF(" + inner + " > 1, 10, 20)"
		},
		"then_slot": func(depth int) string {
			inner := "1"
			for i := 0; i < depth; i++ {
				inner = "IF(Num__c/Den__c > 1, " + inner + ", 2)"
			}
			return inner
		},
	}

	for name, build := range shapes {
		t.Run(name, func(t *testing.T) {
			// Ratio of emitted SQL to source text. Linear growth keeps it flat;
			// any per-level multiplier k makes it grow like k^depth. Comparing
			// ratios rather than raw lengths is what makes this bite: a bound
			// like len(deep) < 4*len(shallow) is satisfied by quadratic growth
			// (14²/7² = 4.0 exactly) and by any k below 1.22.
			ratio := func(depth int) float64 {
				src := build(depth)
				sql := compileNullProvenanceSQL(t, src)
				if !strings.Contains(sql, nullStrictGuardMarker) {
					t.Fatalf("%s depth %d is not gated, so this proves nothing:\n%.500s",
						name, depth, sql)
				}
				return float64(len(sql)) / float64(len(src))
			}
			shallow, deep := ratio(4), ratio(14)
			if deep > 1.5*shallow {
				t.Fatalf("%s: emitted SQL per source char grew %.2f -> %.2f over depths 4..14 "+
					"(%.2fx). Linear emission keeps this flat; growth means a compiled "+
					"subtree is substituted more than once.",
					name, shallow, deep, deep/shallow)
			}
		})
	}
}

// TestCompileNonBooleanConditionStaysAbsorbing pins why the strict emission is
// gated on conditionIsBooleanShaped.
//
// `CASE (c) WHEN TRUE .. WHEN FALSE .. END` covers all three values with one
// substitution only when c is three-valued. For a bare condition subject to
// MySQL truthiness the two forms disagree — verified on TiDB v8.5.3:
//
//	c        CASE WHEN (c) IS NULL THEN NULL WHEN c ..   CASE (c) WHEN TRUE ..
//	5        T                                           NULL
//	0        F                                           F
//	1        T                                           T
//
// So a bare arithmetic condition keeps the absorbing emission: no strictness,
// byte-identical to pre-gate behavior, rather than a silently wrong branch.
func TestCompileNonBooleanConditionStaysAbsorbing(t *testing.T) {
	got := compileNullProvenanceSQL(t, `IF(Num__c/Den__c, 1, 2)`)
	want := "CASE WHEN (COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0)) THEN 1 ELSE 2 END"
	if got != want {
		t.Fatalf("a non-boolean condition must stay absorbing:\n got: %s\nwant: %s", got, want)
	}
	if strings.Contains(got, nullStrictGuardMarker) {
		t.Fatalf("non-boolean condition must not take the single-substitution form:\n%s", got)
	}
}

// TestCompileSingleOperandLogicalStaysAbsorbing pins the hole that the first
// version of conditionIsBooleanShaped left open.
//
// compileLogicalOp joins operands with strings.Join, so a ONE-operand AND()/OR()
// emits the bare operand with no boolean coercion at all — and both the parser
// and the shared function catalog accept single-argument AND/OR (MinArgs: 1),
// with the catalog declaring a boolean return, so nothing upstream rejects
// AND(<number>). Admitting *LogicalOp unconditionally therefore applied the
// single-substitution form to arbitrary arithmetic. Verified on TiDB v8.5.3
// with Num__c=10, Den__c=2 (ratio 5, no division by zero anywhere):
//
//	CASE WHEN ((10/2)) THEN 10 ELSE 20 END                   -> 10   (pre-gate)
//	CASE (((10/2))) WHEN TRUE THEN 10 WHEN FALSE THEN 20 END -> NULL (wrong)
//
// A single operand is now judged by what it actually emits: the operand itself.
func TestCompileSingleOperandLogicalStaysAbsorbing(t *testing.T) {
	ratio := "(COALESCE(t.`Num__c`, 0) / COALESCE(t.`Den__c`, 0))"
	for _, row := range []struct{ name, expr, want string }{
		{"and_bare_arithmetic", `IF(AND(Num__c/Den__c), 10, 20)`,
			"CASE WHEN (" + ratio + ") THEN 10 ELSE 20 END"},
		{"or_bare_arithmetic", `IF(OR(Num__c/Den__c), 10, 20)`,
			"CASE WHEN (" + ratio + ") THEN 10 ELSE 20 END"},
	} {
		t.Run(row.name, func(t *testing.T) {
			got := compileNullProvenanceSQL(t, row.expr)
			if got != row.want {
				t.Fatalf("a single-operand logical wrapping bare arithmetic must stay "+
					"absorbing:\n got: %s\nwant: %s", got, row.want)
			}
			if strings.Contains(got, nullStrictGuardMarker) {
				t.Fatalf("must not take the single-substitution form:\n%s", got)
			}
		})
	}

	// A single operand that IS boolean-shaped keeps the fix: the emission is
	// the operand, and the operand is three-valued.
	got := compileNullProvenanceSQL(t, `IF(AND(Num__c/Den__c > 1), 10, 20)`)
	if !strings.Contains(got, nullStrictGuardMarker) {
		t.Fatalf("AND(<comparison>) emits the comparison itself and must stay strict:\n%s", got)
	}
}