package analytics

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
)

const projectionKindLookupValue = "lookupValue"

// FormulaViewService manages TiDB Views for formula field computation.
// Physical tables store base fields only; Views handle dynamic cross-table
// formula computation via standard SQL. TiFlash MPP engine provides
// millisecond-level performance for the multi-table JOINs.
//
// Architecture:
//
//	Write path: App → PersistenceService → INSERT/UPDATE physical table (never touches View)
//	Read path:  App → QueryRepository → query.From("v_quote") → TiDB View → TiFlash MPP
//	Management: Schema change → FormulaViewService.RebuildView() → CREATE OR REPLACE VIEW
type FormulaViewService struct {
	db       *sql.DB
	metadata MetadataSource
	logger   *logging.Logger

	// Cache of objects that have active views
	mu      sync.RWMutex
	viewSet map[string]bool // lowercase objectName → true if view exists

	// Precomputed validation projection closures per object
	validationProjection *validationProjectionCache

	functionExecutables *formula.ExecutableFunctionRegistry
	// functionExecutableResolver selects the formula vocabulary from the same
	// request-bound immutable metadata view as the schema. Independent Runtime
	// installs this resolver so formula SQL compilation cannot observe a
	// process-global L1 bundle from another Catalog revision.
	functionExecutableResolver func(context.Context) (*formula.ExecutableFunctionRegistry, error)
}

// NewFormulaViewService creates a new FormulaViewService
func NewFormulaViewService(db *sql.DB, metadata MetadataSource, logger *logging.Logger) *FormulaViewService {
	svc := &FormulaViewService{
		db:                   db,
		metadata:             metadata,
		logger:               logger,
		viewSet:              make(map[string]bool),
		validationProjection: newValidationProjectionCache(),
	}
	_ = svc.RefreshRuntimeFunctionRegistry()
	return svc
}

// SetFunctionExecutables installs the active formula-function executable bundle
// registry used by the SQL compiler for active formula views. Nil resets to the
// L0 default catalog.
func (s *FormulaViewService) SetFunctionExecutables(registry *formula.ExecutableFunctionRegistry) {
	if s == nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if registry == nil {
		s.functionExecutables = nil
		return
	}
	s.functionExecutables = registry.Clone()
}

// SetFunctionExecutableResolver installs request-bound formula authority. A
// resolver error is returned to the caller; compilation never falls back to
// the process snapshot after the resolver has been installed.
func (s *FormulaViewService) SetFunctionExecutableResolver(
	resolver func(context.Context) (*formula.ExecutableFunctionRegistry, error),
) {
	if s == nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.functionExecutableResolver = resolver
}

func (s *FormulaViewService) activeFunctionExecutables(
	ctx context.Context,
) (*formula.ExecutableFunctionRegistry, error) {
	if s == nil {
		return nil, nil
	}
	s.mu.RLock()
	resolver := s.functionExecutableResolver
	static := s.functionExecutables
	s.mu.RUnlock()
	if resolver != nil {
		registry, err := resolver(ctx)
		if err != nil {
			return nil, err
		}
		if registry == nil {
			return nil, fmt.Errorf("formula function executable resolver returned no authority")
		}
		return registry.Clone(), nil
	}
	if static == nil {
		return nil, nil
	}
	return static.Clone(), nil
}

// RefreshRuntimeFunctionRegistry reloads the active SQL-compiler executable
// function registry from the process runtime formula-function bundle snapshot.
func (s *FormulaViewService) RefreshRuntimeFunctionRegistry() error {
	if s == nil {
		return nil
	}
	registry, err := formula.ActiveRuntimeExecutableFunctionRegistry()
	if err != nil {
		return err
	}
	s.SetFunctionExecutables(registry)
	return nil
}

// ViewName returns the view name for a given object
func ViewName(objectName string) string {
	return "v_" + strings.ToLower(objectName)
}

// CreateOrReplaceViewStmt is the header shared by every formula view DDL emitted
// by this package. SQL SECURITY INVOKER is required: MySQL/TiDB otherwise
// defaults to SQL SECURITY DEFINER and stamps the creating user into the view,
// so views created on one cluster fail with 1356 after a TiDB Serverless branch
// restore or user rotation where the stamped DEFINER no longer exists.
const CreateOrReplaceViewStmt = "CREATE OR REPLACE SQL SECURITY INVOKER VIEW"

// RebuildView generates and executes CREATE OR REPLACE VIEW for an object.
// It reads formula field metadata, analyzes dependencies, translates SF formulas
// to SQL expressions, and builds a nested-subquery View DDL.
func (s *FormulaViewService) RebuildView(ctx context.Context, objectName string) error {
	// Capture the projection generation and read the schema as one epoch via
	// projectionEpochForObject, which captures the generation BEFORE the read so
	// a concurrent invalidation bumps the generation and the stale-rebuild guard
	// at store time drops a closure built from a now-superseded schema. Capturing
	// before the read is the fail-safe direction: a false-drop merely falls back
	// to all-fields (correct, slower); a false-store would be the correctness
	// regression.
	projectionGen, schema := s.projectionEpochForObject(ctx, objectName)
	if schema == nil {
		return fmt.Errorf("object %s not found", objectName)
	}

	// Collect active computed fields. Custom formula fields and standard
	// system-derived fields share the same generic projection compiler.
	formulaFields := formulastorage.ActiveComputedFields(schema)

	if len(formulaFields) == 0 {
		// No active computed fields — drop view if it exists
		return s.DropView(ctx, objectName)
	}

	s.logger.Info(fmt.Sprintf("Using DSL engine for %s (%d formula fields)", objectName, len(formulaFields)))
	ddl, err := s.buildViewFromDSL(ctx, objectName, schema, formulaFields, projectionGen, true)
	if err != nil {
		return fmt.Errorf("failed to build view DDL for %s: %w", objectName, err)
	}

	// Execute DDL
	s.logger.Info(fmt.Sprintf("Rebuilding view %s (%d formula fields)",
		ViewName(objectName), len(formulaFields)))

	_, err = s.db.ExecContext(ctx, ddl)
	if err != nil {
		return fmt.Errorf("failed to create view %s: %w\nDDL: %s", ViewName(objectName), err, ddl)
	}

	// Update cache
	s.mu.Lock()
	s.viewSet[strings.ToLower(objectName)] = true
	s.mu.Unlock()

	return nil
}

// buildViewFromDSL generates View DDL using the new DSL parser/compiler pipeline.
// This is the primary path for formula fields created via the CRM UI.
func (s *FormulaViewService) buildViewFromDSL(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	formulaFields []models.FieldMetadata,
	projectionGen uint64,
	publishProjection bool,
) (string, error) {
	baseReadTable, err := resolveSchemaReadSourceTable(schema)
	if err != nil {
		return "", err
	}

	defs, err := s.buildFormulaDefs(ctx, objectName, schema, formulaFields)
	if err != nil {
		return "", err
	}

	// Precompute validation projection closure now that ExtraDeps are resolved.
	// Only committed rebuild paths publish into the live cache. Read-only flows
	// (PreviewDDL, ValidateFormulaForField) pass candidate/unsaved field defs and
	// must NOT mutate the runtime cache — doing so could narrow live validation
	// against a closure derived from hypothetical deps.
	if publishProjection {
		s.computeValidationProjectionClosure(ctx, objectName, schema, defs, projectionGen)
	}

	// Collect physical pass-through columns from metadata.
	baseSet := make(map[string]bool)
	var baseColumns []string
	baseColumnAliases := make(map[string]string) // APIName → PhysicalName when they differ
	var joins []JoinSpec
	usedAliases := make(map[string]bool)

	// Track API names of formula fields that this view will compute so the
	// INFORMATION_SCHEMA supplement below does not add a stale physical
	// snapshot with the same output alias. Frozen or unparseable formulas are
	// not in defs, so their physical snapshot columns remain readable.
	computedFormulaNames := make(map[string]bool, len(defs))
	for _, def := range defs {
		computedFormulaNames[formulaViewProjectionKey(def.Name)] = true
	}

	// Query INFORMATION_SCHEMA.COLUMNS first to get the set of physical columns.
	// This is used to filter out metadata-declared columns that don't have a
	// physical counterpart (issue #1392: LastActivityDate in _System_Field but
	// missing from physical table causes Error 1054 in VIEW DDL).
	// Store lowercase key → original name for both filtering and DDL emission.
	physicalColumnNames := make(map[string]string) // lowercase → original
	if s.db != nil && schema.ReadSource == nil {
		physRows, err := s.db.QueryContext(ctx,
			`SELECT COLUMN_NAME FROM information_schema.COLUMNS
			 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
			 ORDER BY ORDINAL_POSITION`, baseReadTable)
		if err != nil {
			return "", fmt.Errorf("formula view physical column probe for %s: %w", baseReadTable, err)
		}
		for physRows.Next() {
			var colName string
			if err := physRows.Scan(&colName); err != nil {
				physRows.Close()
				return "", fmt.Errorf("formula view physical column probe scan for %s: %w", baseReadTable, err)
			}
			physicalColumnNames[strings.ToLower(colName)] = colName
		}
		if err := physRows.Err(); err != nil {
			physRows.Close()
			return "", fmt.Errorf("formula view physical column probe rows for %s: %w", baseReadTable, err)
		}
		if err := physRows.Close(); err != nil {
			return "", fmt.Errorf("formula view physical column probe close for %s: %w", baseReadTable, err)
		}
	}

	for _, f := range schema.Fields {
		if formulastorage.IsActiveComputedField(&f) {
			continue // skip fields computed by the view projection
		}
		if f.IsVirtual || f.IsCompound {
			continue // immutable metadata owns the derived projection, not a stale physical column
		}
		// Issue #1392: skip metadata-declared columns that don't physically exist.
		// The repair path (repairManagedSystemFieldColumnsFromLiveMetadata) should
		// have added missing columns, but if it didn't (e.g., field not in spec),
		// we must not reference a non-existent column in the VIEW DDL.
		if len(physicalColumnNames) > 0 {
			if _, exists := physicalColumnNames[strings.ToLower(f.APIName)]; !exists {
				continue
			}
		}
		if key := formulaViewProjectionKey(f.APIName); !baseSet[key] {
			baseColumns = append(baseColumns, f.APIName)
			baseSet[key] = true
		}
		// Record PhysicalName alias so L0 emits t.`PhysicalName` AS `APIName`.
		// Use the same normalization as the query-path SSoT (persistence.actualColumnName):
		// backtick-wrapped PhysicalName values mean "the DB column IS the API name" and must
		// not be forwarded as a physical alias (they would produce invalid SQL like t.``Name``).
		if f.PhysicalName != "" && f.PhysicalName != f.APIName {
			if physCol := analyticsPhysicalColumnName(f.APIName, f.PhysicalName); physCol != f.APIName {
				baseColumnAliases[f.APIName] = physCol
			}
		}

		if join, ok := relationshipJoinSpecForField(f, usedAliases); ok {
			joins = append(joins, join)
		}
	}

	// When the object has no physical Name column but carries a physical IsNameField
	// (e.g. OrderNumber on autonumber objects), map logical "Name" to that column so
	// the base projection emits t.`OrderNumber` AS `Name`. Outer formula layers (L1+)
	// can then reference L0.`Name` without an unknown-column error.
	// Active-computed IsNameField cases return "" — dep-layering handles those.
	if alias := logicalNamePhysicalAlias(schema); alias != "" {
		baseColumnAliases[constants.FieldName] = alias
		if key := formulaViewProjectionKey(constants.FieldName); !baseSet[key] {
			baseColumns = append(baseColumns, constants.FieldName)
			baseSet[key] = true
		}
	}

	usedCrossRefs, err := formulaCrossRefObjects(defs)
	if err != nil {
		return "", err
	}
	ambiguousTargets := crossRefAmbiguousTargetKeys(joins)

	// Supplement base columns from the physical column set (already queried above)
	// to catch physical columns that exist on the table but are not yet registered
	// in _System_Field. Skip any column whose name is owned by a formula alias
	// computed by this view — otherwise the stale physical snapshot collides with
	// the computed alias. Keep non-computed formula snapshots (for example frozen
	// formulas).
	for _, colName := range physicalColumnNames {
		key := formulaViewProjectionKey(colName)
		if key != "" && !baseSet[key] && !computedFormulaNames[key] {
			baseColumns = append(baseColumns, colName)
			baseSet[key] = true
		}
	}

	// Validate JOINs: active formula views must not hide missing metadata
	// materialization by silently dropping relationship joins.
	var validJoins []JoinSpec
	for _, j := range joins {
		if !joinReferencedByFormula(j, usedCrossRefs, ambiguousTargets) {
			continue
		}
		// Determine whether any field read on this join target via formulas is
		// active-computed. If so, route to v_<target> so that the runtime SQL
		// selects from the formula view (which exposes computed columns) rather
		// than the physical table (which does not have them).
		targetSchema := s.metadata.GetSchema(ctx, j.TargetTable)
		readsActiveComputed := false
		if targetSchema != nil {
			for _, resolvedField := range crossRefFieldsReadOnTarget(j, defs, targetSchema, ambiguousTargets) {
				if formulastorage.SchemaFieldIsActiveComputed(targetSchema, resolvedField) {
					readsActiveComputed = true
					break
				}
			}
		}
		isSelfRef := strings.EqualFold(j.TargetTable, objectName)
		if isSelfRef && readsActiveComputed {
			actualTable, err := s.materializedTableName(ctx, j.TargetTable)
			if err != nil {
				return "", fmt.Errorf("reference target table %q for relationship %q is not materialized", j.TargetTable, j.RelationshipName)
			}
			j.UsesCurrentObjectProjection = true
			j.TargetObject = j.TargetTable
			j.TargetTable = actualTable
			validJoins = append(validJoins, j)
			continue
		}
		actualTable, err := s.joinTargetForFields(ctx, j.TargetTable, readsActiveComputed)
		if err != nil {
			return "", fmt.Errorf("reference target table %q for relationship %q is not materialized", j.TargetTable, j.RelationshipName)
		}
		// Preserve the logical object name for schema/metadata lookups; TargetTable
		// becomes the SQL emission name (may differ when routed to a formula view).
		j.TargetObject = j.TargetTable
		j.TargetTable = actualTable
		validJoins = append(validJoins, j)
	}

	// Hoist single-PK-match lookupValue projections from correlated subqueries to
	// LEFT JOINs. When a def has exactly one LookupValueRule whose sole match pair
	// targets the primary key (Id), TiDB can plan an IndexJoin instead of a
	// per-row Apply (CARTESIAN nested loop). The join targets the base table (same
	// as the subquery) and is appended directly to validJoins, bypassing the
	// joinReferencedByFormula filter which only accepts CrossRef-referenced joins.
	for i, def := range defs {
		if len(def.LookupValueRules) != 1 {
			continue
		}
		rule := def.LookupValueRules[0]
		if len(rule.MatchPairs) != 1 {
			continue
		}
		if !strings.EqualFold(rule.MatchPairs[0].MatchField, constants.FieldID) {
			continue
		}
		// L0-only guard: if the match SourceField is itself an active-computed
		// (formula) field on the source object, the lookupValue definition lives in
		// a layer above L0 in the multi-layer DDL path. Emitting the lv_ JOIN only
		// in the L0 inner FROM but referencing lv_x.field in an outer SELECT would
		// produce "Unknown table 'lv_x'" at CREATE VIEW time. Keep the portable
		// correlated-subquery form for formula-source-field lookupValues.
		if formulastorage.SchemaFieldIsActiveComputed(schema, rule.MatchPairs[0].SourceField) {
			continue
		}
		// The JOIN ON clause references the source column on the base table t.
		// Resolve it through the same physical-name map the base projection uses
		// (baseColumnAliases == the compiler's BaseFieldPhysicalMap), so a source
		// field whose physical column differs from its API name emits t.<physical>,
		// exactly as the old correlated-subquery path did. Without this, RebuildView
		// would reference a non-existent column for metadata-aliased source fields.
		sourceCol := rule.MatchPairs[0].SourceField
		if phys := baseColumnAliases[sourceCol]; phys != "" {
			sourceCol = phys
		}
		alias := makeJoinAlias(rule.TargetTable, "lv_", usedAliases)
		validJoins = append(validJoins, JoinSpec{
			Alias:               alias,
			TargetObject:        rule.TargetTable,
			TargetTable:         rule.TargetTable,
			SourceField:         sourceCol,
			TargetField:         constants.FieldID,
			FilterTargetDeleted: rule.ExcludeDeleted,
			HoistedLookupValue:  true,
		})
		defs[i].LookupValueRules[0].HoistedJoinAlias = alias
	}

	// Build aggregate joins from metadata-declared child relationships.
	var aggJoins []AggregateJoinSpec
	aggIdx := 0
	for _, def := range defs {
		ast, err := Parse(def.Expression)
		if err != nil {
			continue
		}
		for _, agg := range extractAggregates(ast) {
			spec, err := s.resolveAggregateJoinSpec(ctx, schema, agg, aggIdx)
			if err != nil {
				return "", fmt.Errorf("aggregate in %s.%s: %w", objectName, def.Name, err)
			}
			aggJoins = append(aggJoins, spec)
			aggIdx++
		}
	}

	// Resolve deep cross-table references (multi-hop JOINs like Owner.Profile.Name)
	deepJoins, err := s.resolveDeepJoinChains(ctx, objectName, schema, defs)
	if err != nil {
		return "", err
	}
	if len(deepJoins) > 0 && s.logger != nil {
		s.logger.Info(fmt.Sprintf("DSL: resolved %d deep JOIN chains for %s", len(deepJoins), objectName))
	}

	// Build config and generate DDL. FieldTypes threads Salesforce field
	// types (date, currency, etc.) into the compiler so binary ops can emit
	// DATEDIFF / DATE_ADD / DATE_SUB when operands are dates instead of
	// MySQL's default YYYYMMDD numeric subtraction.
	crossRefFieldAliases := s.buildCrossRefFieldAliasMap(ctx, validJoins)
	crossRefFieldExpressions, err := s.buildCrossRefFieldExpressionMap(ctx, validJoins)
	if err != nil {
		return "", err
	}
	functionExecutables, err := s.activeFunctionExecutables(ctx)
	if err != nil {
		return "", fmt.Errorf("resolve formula function authority: %w", err)
	}
	fieldTypes := s.buildFieldTypeMap(ctx, schema, validJoins, deepJoins, crossRefFieldAliases)
	cfg := ViewBuildConfig{
		ObjectName:               objectName,
		PhysicalTable:            baseReadTable,
		TableAlias:               "t",
		BaseColumns:              baseColumns,
		BaseColumnAliases:        baseColumnAliases,
		Formulas:                 defs,
		Joins:                    validJoins,
		AggJoins:                 aggJoins,
		DeepJoins:                deepJoins,
		CrossRefFieldAliases:     crossRefFieldAliases,
		CrossRefFieldExpressions: crossRefFieldExpressions,
		FieldTypes:               fieldTypes,
		FunctionExecutables:      functionExecutables,
	}

	builder := NewDSLViewBuilder()
	return builder.BuildViewDDL(cfg)
}

func (s *FormulaViewService) resolveAggregateJoinSpec(
	ctx context.Context,
	parentSchema *models.ObjectMetadata,
	agg AggregateExpr,
	ordinal int,
) (AggregateJoinSpec, error) {
	if parentSchema == nil || strings.TrimSpace(parentSchema.APIName) == "" {
		return AggregateJoinSpec{}, fmt.Errorf("aggregate parent schema is missing")
	}
	if s == nil || s.metadata == nil {
		return AggregateJoinSpec{}, fmt.Errorf("metadata service is required to resolve aggregate relationship")
	}
	childSchema := s.metadata.GetSchema(ctx, agg.ChildObject)
	if childSchema == nil {
		return AggregateJoinSpec{}, fmt.Errorf("child object %q not found in metadata", agg.ChildObject)
	}
	if agg.ChildField == "*" && !strings.EqualFold(agg.Func, "COUNT") {
		return AggregateJoinSpec{}, fmt.Errorf("%s aggregate does not support wildcard child fields", agg.Func)
	}
	if agg.ChildField != "" && agg.ChildField != "*" && !metadataFieldExists(childSchema, agg.ChildField) {
		return AggregateJoinSpec{}, fmt.Errorf("aggregate field %q not found on child object %q", agg.ChildField, childSchema.APIName)
	}

	fk, err := resolveAggregateForeignKey(parentSchema, childSchema, agg.ForeignKey)
	if err != nil {
		return AggregateJoinSpec{}, err
	}
	childTable, err := s.materializedTableName(ctx, childSchema.APIName)
	if err != nil {
		return AggregateJoinSpec{}, fmt.Errorf("child object %q table is not materialized: %w", childSchema.APIName, err)
	}

	aggField := agg.ChildField
	if aggField == "" {
		aggField = "*"
	}
	return AggregateJoinSpec{
		Alias:       fmt.Sprintf("rsq%d", ordinal),
		ChildObject: childSchema.APIName,
		ChildTable:  childTable,
		AggFunc:     agg.Func,
		AggField:    aggField,
		ForeignKey:  fk,
		ParentKey:   constants.FieldID,
		OutputAlias: fmt.Sprintf("agg_%d", ordinal),
	}, nil
}

func resolveAggregateForeignKey(parentSchema, childSchema *models.ObjectMetadata, explicitFK string) (string, error) {
	if parentSchema == nil || childSchema == nil {
		return "", fmt.Errorf("parent and child schemas are required")
	}
	parentName := strings.TrimSpace(parentSchema.APIName)
	if parentName == "" {
		return "", fmt.Errorf("parent object API name is required")
	}
	if explicitFK = strings.TrimSpace(explicitFK); explicitFK != "" && explicitFK != "_" {
		field := findReferenceField(childSchema, explicitFK)
		if field == nil {
			return "", fmt.Errorf("declared aggregate foreign key %q is not a reference field on %s", explicitFK, childSchema.APIName)
		}
		if !referenceToContains(field.ReferenceTo, parentName) {
			return "", fmt.Errorf("declared aggregate foreign key %s.%s does not reference %s", childSchema.APIName, explicitFK, parentName)
		}
		return field.APIName, nil
	}

	var matches []string
	for _, field := range childSchema.Fields {
		if !constants.IsReferenceType(field.Type) {
			continue
		}
		if referenceToContains(field.ReferenceTo, parentName) {
			matches = append(matches, field.APIName)
		}
	}
	switch len(matches) {
	case 0:
		return "", fmt.Errorf("no reference field on %s points to aggregate parent %s", childSchema.APIName, parentName)
	case 1:
		return matches[0], nil
	default:
		sort.Strings(matches)
		return "", fmt.Errorf("multiple reference fields on %s point to aggregate parent %s: %s", childSchema.APIName, parentName, strings.Join(matches, ", "))
	}
}

func findReferenceField(schema *models.ObjectMetadata, apiName string) *models.FieldMetadata {
	if schema == nil || apiName == "" {
		return nil
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if strings.EqualFold(field.APIName, apiName) &&
			constants.IsReferenceType(field.Type) &&
			len(field.ReferenceTo) > 0 {
			return field
		}
	}
	return nil
}

func referenceToContains(referenceTo []string, objectAPIName string) bool {
	for _, candidate := range referenceTo {
		if strings.EqualFold(candidate, objectAPIName) {
			return true
		}
	}
	return false
}

func (s *FormulaViewService) materializedTableName(ctx context.Context, objectAPIName string) (string, error) {
	objectAPIName = strings.TrimSpace(objectAPIName)
	if objectAPIName == "" {
		return "", fmt.Errorf("object API name is required")
	}
	baseReadTable := objectAPIName
	if schema := s.resolveSchema(ctx, objectAPIName); schema != nil {
		resolved, err := resolveSchemaReadSourceTable(schema)
		if err != nil {
			return "", err
		}
		baseReadTable = resolved
	}
	if s == nil || s.db == nil {
		return baseReadTable, nil
	}
	var actualTable string
	if err := s.db.QueryRowContext(ctx, `SELECT TABLE_NAME FROM information_schema.TABLES
		WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = ? LIMIT 1`, strings.ToLower(baseReadTable)).Scan(&actualTable); err != nil {
		return "", err
	}
	if actualTable == "" {
		return "", fmt.Errorf("table not found")
	}
	return actualTable, nil
}

func (s *FormulaViewService) resolveSchema(ctx context.Context, objectAPIName string) *models.ObjectMetadata {
	if s == nil || s.metadata == nil {
		return nil
	}
	return s.metadata.GetSchema(ctx, objectAPIName)
}

func resolveSchemaReadSourceTable(schema *models.ObjectMetadata) (string, error) {
	if schema == nil {
		return "", fmt.Errorf("object metadata is required")
	}
	if schema.ReadSource == nil {
		return strings.ToLower(strings.TrimSpace(schema.APIName)), nil
	}
	baseReadTable, err := schema.ResolveReadSourceTable()
	if err != nil {
		return "", fmt.Errorf("resolve readSource table for %s: %w", schema.APIName, err)
	}
	return baseReadTable, nil
}

// joinTargetForFields returns the physical SQL name to JOIN against for the
// given target object. When the JOIN reads at least one active-computed field
// on the target (readsActiveComputed == true), the target's formula view
// v_<obj> is returned instead of the physical table — otherwise the formula
// column does not exist on the physical table and the runtime SQL would fail.
//
// For physical-only reads (readsActiveComputed == false), the existing
// materializedTableName lookup is used (queries information_schema.TABLES for
// the actual physical table name).
//
// Aggregate JOINs are intentionally excluded from this routing: aggregate
// subqueries emit FROM <child_physical_table> directly and read physical
// columns only.
func (s *FormulaViewService) joinTargetForFields(ctx context.Context, targetAPIName string, readsActiveComputed bool) (string, error) {
	if readsActiveComputed {
		return ViewName(targetAPIName), nil
	}
	return s.materializedTableName(ctx, targetAPIName)
}

func (s *FormulaViewService) attachSelfProjectionDeps(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	defs []FormulaDefinition,
) ([]FormulaDefinition, error) {
	if schema == nil || len(defs) == 0 {
		return defs, nil
	}

	nameSet := make(map[string]string, len(defs))
	for _, def := range defs {
		nameSet[strings.ToLower(def.Name)] = def.Name
	}

	baseTargets := buildRelationshipTargetsFromSchema(schema)

	// When the object has no physical Name column and the IsNameField is
	// view-computed (e.g. Subject on a Case-like object), bare "Name" FieldRef
	// expressions inside formulas must be rewritten to reference the computed
	// display field directly. This ensures (a) the formula is ordered after the
	// display field in the DAG, and (b) the compiler emits the correct
	// outer-layer column reference (L0.`Subject`) rather than the non-existent
	// physical column (t.`Name`).
	var activeComputedNameAPIName string
	if !metadataFieldExists(schema, constants.FieldName) {
		for _, f := range schema.Fields {
			if f.IsNameField && !strings.EqualFold(f.APIName, constants.FieldName) &&
				formulastorage.IsActiveComputedField(&f) {
				activeComputedNameAPIName = f.APIName
				break
			}
		}
	}

	out := make([]FormulaDefinition, len(defs))
	copy(out, defs)
	for i, def := range out {
		ast, err := Parse(def.Expression)
		if err != nil {
			return nil, fmt.Errorf("parse active formula %s.%s: %w", objectName, def.Name, err)
		}

		var extraDeps []string
		for _, ref := range extractCrossRefs(ast) {
			target := baseTargets[strings.ToLower(ref.Object)]
			if !strings.EqualFold(target, objectName) {
				continue
			}
			resolvedField, err := resolveReferencedField(schema, ref.Field)
			if err != nil {
				continue
			}
			if dep, ok := nameSet[strings.ToLower(resolvedField)]; ok {
				extraDeps = append(extraDeps, dep)
			}
		}

		for _, dcr := range extractDeepCrossRefs(ast) {
			if len(dcr.Parts) < 2 {
				continue
			}
			currentSchema := schema
			lastTarget := ""
			for hopIdx := 0; hopIdx < len(dcr.Parts)-1; hopIdx++ {
				hopTargets := buildRelationshipTargetsFromSchema(currentSchema)
				target := hopTargets[strings.ToLower(dcr.Parts[hopIdx])]
				if target == "" {
					lastTarget = ""
					break
				}
				lastTarget = target
				if hopIdx == len(dcr.Parts)-2 {
					break
				}
				nextSchema := s.metadata.GetSchema(ctx, target)
				if nextSchema == nil {
					lastTarget = ""
					break
				}
				currentSchema = nextSchema
			}
			if !strings.EqualFold(lastTarget, objectName) {
				continue
			}
			finalField, err := resolveReferencedField(schema, dcr.Parts[len(dcr.Parts)-1])
			if err != nil {
				continue
			}
			if dep, ok := nameSet[strings.ToLower(finalField)]; ok {
				extraDeps = append(extraDeps, dep)
			}
		}

		// Rewrite bare "Name" FieldRef references to the active-computed display
		// field when the object has no physical Name column. AST-level rewrite
		// ensures only standalone FieldRef{Name:"Name"} nodes are replaced —
		// CrossRef / DeepCrossRef nodes like Parent.Name are left untouched.
		if activeComputedNameAPIName != "" {
			for _, ref := range extractFieldRefs(ast) {
				if strings.EqualFold(ref, constants.FieldName) {
					out[i].Expression = formulaExprString(rewriteBareNameFieldRefAST(ast, activeComputedNameAPIName))
					break
				}
			}
		}

		out[i].ExtraDeps = dedupePreserveOrder(extraDeps)
	}
	return out, nil
}

// rewriteBareNameFieldRefAST walks the formula AST and replaces *FieldRef nodes
// whose Name equals "Name" (case-insensitive) with *FieldRef{Name: replacement}.
// CrossRef and DeepCrossRef nodes (e.g. Parent.Name) are left untouched — only
// bare field references are rewritten, never cross-object dotted paths.
func rewriteBareNameFieldRefAST(node Node, replacement string) Node {
	if node == nil {
		return nil
	}
	switch n := node.(type) {
	case *FieldRef:
		if strings.EqualFold(n.Name, constants.FieldName) {
			return &FieldRef{Name: replacement}
		}
		return n
	case *BinaryOp:
		return &BinaryOp{
			Op:    n.Op,
			Left:  rewriteBareNameFieldRefAST(n.Left, replacement),
			Right: rewriteBareNameFieldRefAST(n.Right, replacement),
		}
	case *UnaryOp:
		return &UnaryOp{
			Op:      n.Op,
			Operand: rewriteBareNameFieldRefAST(n.Operand, replacement),
		}
	case *LogicalOp:
		newOps := make([]Node, len(n.Operands))
		for i, o := range n.Operands {
			newOps[i] = rewriteBareNameFieldRefAST(o, replacement)
		}
		return &LogicalOp{Op: n.Op, Operands: newOps}
	case *FuncCall:
		newArgs := make([]Node, len(n.Args))
		for i, a := range n.Args {
			newArgs[i] = rewriteBareNameFieldRefAST(a, replacement)
		}
		return &FuncCall{Name: n.Name, Args: newArgs}
	case *IfExpr:
		return &IfExpr{
			Condition: rewriteBareNameFieldRefAST(n.Condition, replacement),
			Then:      rewriteBareNameFieldRefAST(n.Then, replacement),
			Else:      rewriteBareNameFieldRefAST(n.Else, replacement),
		}
	case *CaseExpr:
		newBranches := make([]CaseBranch, len(n.Branches))
		for i, b := range n.Branches {
			newBranches[i] = CaseBranch{
				Value:  rewriteBareNameFieldRefAST(b.Value, replacement),
				Result: rewriteBareNameFieldRefAST(b.Result, replacement),
			}
		}
		var newDefault Node
		if n.Default != nil {
			newDefault = rewriteBareNameFieldRefAST(n.Default, replacement)
		}
		return &CaseExpr{
			Expr:     rewriteBareNameFieldRefAST(n.Expr, replacement),
			Branches: newBranches,
			Default:  newDefault,
		}
	default:
		// Leaf nodes (literals, CrossRef, DeepCrossRef, AggregateExpr) — unchanged.
		return n
	}
}

// formulaExprString serializes an AST node back to parseable formula DSL text.
// Unlike nodeString (which is a debug printer), this is a round-trip serializer:
// it preserves AggregateExpr.ForeignKey and escapes embedded quotes in StringLit.
func formulaExprString(n Node) string {
	switch v := n.(type) {
	case *NumberLit:
		return v.Value
	case *StringLit:
		esc := strings.ReplaceAll(v.Value, `\`, `\\`)
		esc = strings.ReplaceAll(esc, `'`, `\'`)
		return "'" + esc + "'"
	case *BoolLit:
		if v.Value {
			return "TRUE"
		}
		return "FALSE"
	case *NullLit:
		return "NULL"
	case *FieldRef:
		return v.Name
	case *CrossRef:
		return fmt.Sprintf("%s.%s", v.Object, v.Field)
	case *DeepCrossRef:
		return strings.Join(v.Parts, ".")
	case *BinaryOp:
		return fmt.Sprintf("(%s %s %s)", formulaExprString(v.Left), v.Op, formulaExprString(v.Right))
	case *UnaryOp:
		return fmt.Sprintf("(%s %s)", v.Op, formulaExprString(v.Operand))
	case *LogicalOp:
		parts := make([]string, len(v.Operands))
		for i, o := range v.Operands {
			parts[i] = formulaExprString(o)
		}
		return fmt.Sprintf("(%s)", strings.Join(parts, " "+v.Op+" "))
	case *FuncCall:
		args := make([]string, len(v.Args))
		for i, a := range v.Args {
			args[i] = formulaExprString(a)
		}
		return fmt.Sprintf("%s(%s)", v.Name, strings.Join(args, ", "))
	case *IfExpr:
		return fmt.Sprintf("IF(%s, %s, %s)", formulaExprString(v.Condition), formulaExprString(v.Then), formulaExprString(v.Else))
	case *CaseExpr:
		s := fmt.Sprintf("CASE(%s", formulaExprString(v.Expr))
		for _, b := range v.Branches {
			s += fmt.Sprintf(", %s, %s", formulaExprString(b.Value), formulaExprString(b.Result))
		}
		if v.Default != nil {
			s += fmt.Sprintf(", %s", formulaExprString(v.Default))
		}
		return s + ")"
	case *AggregateExpr:
		var base string
		if v.ChildField == "" {
			base = fmt.Sprintf("%s(%s", v.Func, v.ChildObject)
		} else {
			base = fmt.Sprintf("%s(%s.%s", v.Func, v.ChildObject, v.ChildField)
		}
		if v.ForeignKey != "" {
			return base + ", " + v.ForeignKey + ")"
		}
		return base + ")"
	default:
		return fmt.Sprintf("?<%T>", n)
	}
}

// effectiveFieldType adapts the shared schema result type to the compiler's
// string-keyed field type map.
func effectiveFieldType(f models.FieldMetadata) string {
	return string(schemautil.EffectiveFieldType(f))
}

// DropView drops the view for an object
func (s *FormulaViewService) DropView(ctx context.Context, objectName string) error {
	viewName := ViewName(objectName)
	if err := validateIdentifier(viewName, "view"); err != nil {
		return err
	}
	_, err := s.db.ExecContext(ctx, fmt.Sprintf("DROP VIEW IF EXISTS `%s`", viewName))
	if err != nil {
		return fmt.Errorf("failed to drop view %s: %w", viewName, err)
	}

	s.mu.Lock()
	delete(s.viewSet, strings.ToLower(objectName))
	s.mu.Unlock()

	if s.validationProjection != nil {
		s.validationProjection.remove(objectName)
	}

	return nil
}

// ViewExists checks if a view exists for the given object (cached check + DB fallback)
func (s *FormulaViewService) ViewExists(ctx context.Context, objectName string) bool {
	key := strings.ToLower(objectName)

	s.mu.RLock()
	exists, cached := s.viewSet[key]
	s.mu.RUnlock()
	if cached {
		return exists
	}

	if s.db == nil {
		return false
	}

	// DB fallback: check information_schema
	viewName := ViewName(objectName)
	var count int
	err := s.db.QueryRowContext(ctx,
		"SELECT COUNT(*) FROM information_schema.VIEWS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?",
		viewName,
	).Scan(&count)
	if err != nil {
		return false
	}

	result := count > 0
	s.mu.Lock()
	s.viewSet[key] = result
	s.mu.Unlock()

	return result
}

func (s *FormulaViewService) cachedViewExists(objectName string) (bool, bool) {
	key := strings.ToLower(objectName)
	s.mu.RLock()
	defer s.mu.RUnlock()
	exists, cached := s.viewSet[key]
	return exists, cached
}

// ResolveTableName returns v_{object} when the object has active computed
// fields. Missing views fail closed at query time instead of silently reading
// stale physical snapshot columns.
//
// Runtime query routing must not fall back to a fresh information_schema probe:
// callers frequently hold an open pessimistic transaction while resolving the
// read target, and a DB fallback here would need a second pooled connection.
// Under high same-parent concurrency that can exhaust the pool and deadlock the
// in-flight writers against each other. The authoritative miss behavior is
// already fail-closed, so a cold cache routes directly to the view name.
// This is called by QueryRepository to route reads to views.
func (s *FormulaViewService) ResolveTableName(ctx context.Context, schema *models.ObjectMetadata) string {
	if schema == nil {
		return ""
	}

	// Quick check: does the schema have any active computed fields?
	hasActive := false
	for _, f := range schema.Fields {
		if formulastorage.IsActiveComputedField(&f) {
			hasActive = true
			break
		}
	}
	if !hasActive {
		return schema.APIName
	}

	viewName := ViewName(schema.APIName)
	if exists, cached := s.cachedViewExists(schema.APIName); cached {
		if exists {
			return viewName
		}
		if s.logger != nil {
			s.logger.Error(fmt.Sprintf("Formula view %s is missing for %s with active computed fields; routing to the view to fail closed", viewName, schema.APIName), nil)
		}
		return viewName
	}
	return viewName
}

// RefreshViewCache scans the database for existing formula views and updates the cache.
// Call this at startup to warm the cache.
func (s *FormulaViewService) RefreshViewCache(ctx context.Context) error {
	rows, err := s.db.QueryContext(ctx,
		"SELECT TABLE_NAME FROM information_schema.VIEWS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME LIKE 'v\\_%'",
	)
	if err != nil {
		return fmt.Errorf("failed to query views: %w", err)
	}
	defer rows.Close()

	newSet := make(map[string]bool)
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			continue
		}
		// v_quote → quote
		if strings.HasPrefix(name, "v_") {
			newSet[name[2:]] = true
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("error iterating views: %w", err)
	}

	s.mu.Lock()
	s.viewSet = newSet
	s.mu.Unlock()

	s.logger.Info(fmt.Sprintf("Formula view cache refreshed: %d views found", len(newSet)))
	return nil
}

// PreviewDDL returns the View DDL that would be generated without executing it.
// Useful for debugging and review before applying changes.
func (s *FormulaViewService) PreviewDDL(ctx context.Context, objectName string) (string, error) {
	schema := s.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		// Typed NotFound so the boundary renders a 404, not a generic 500 (a bare
		// fmt.Errorf falls through GlobalErrorHandler to InternalServerError).
		return "", apperrors.NewNotFoundError(fmt.Sprintf("object %q", objectName), "")
	}

	formulaFields := formulastorage.ActiveComputedFields(schema)

	// No active computed fields → no formula view exists for this object, so there
	// is no DDL to preview. Return a TYPED NotFound (renders a clean 404 instead of
	// the previous bare-error-to-500). It stays an *error* deliberately: PreviewDDL
	// doubles as the pack-verify formula compile-proof (controlplane.source_pack_ops
	// / verify.go), which only calls it for objects the DB flags as having active
	// formulas and treats any error as a verify failure. Returning ("",nil) here
	// would let a cache/DB desync (DB says active formula, cache says none) pass the
	// proof silently — a fail-open the formula-view integrity contract forbids.
	if len(formulaFields) == 0 {
		return "", apperrors.NewNotFoundError(fmt.Sprintf("formula view for object %q", objectName), "")
	}

	// Read-only preview: never publish into the live projection cache (gen 0,
	// publish=false).
	return s.buildViewFromDSL(ctx, objectName, schema, formulaFields, 0, false)
}

// RebuildAllViews rebuilds views for all objects that have active computed fields,
// in dependency order computed from live metadata via OrderForBuild.
func (s *FormulaViewService) RebuildAllViews(ctx context.Context) error {
	schemas := s.metadata.GetSchemas(ctx)

	// Collect objects with at least one active computed field.
	var candidates []string
	for _, schema := range schemas {
		if formulastorage.HasActiveComputedFields(schema) {
			candidates = append(candidates, schema.APIName)
		}
	}
	if len(candidates) == 0 {
		return nil
	}

	// Order by live-metadata dependency graph. Fails closed on cycles or missing
	// metadata — no lexical / pack-payload fallback per design Forbidden Move #5/#11.
	ordered, err := s.OrderForBuild(ctx, candidates)
	if err != nil {
		return fmt.Errorf("compute formula view build order: %w", err)
	}

	var errs []string
	for _, name := range ordered {
		if err := s.RebuildView(ctx, name); err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", name, err))
			s.logger.Warn(fmt.Sprintf("Failed to rebuild view for %s: %v", name, err))
		}
	}

	if len(errs) > 0 {
		return fmt.Errorf("view rebuild errors: %s", strings.Join(errs, "; "))
	}
	return nil
}

// buildFormulaDefs converts active-computed FieldMetadata into resolved DSL
// FormulaDefinitions (expressions parsed, lookup rules attached, self-projection
// ExtraDeps resolved). This is the metadata-only prefix shared by view DDL
// generation and DDL-free validation-projection recompute.
func (s *FormulaViewService) buildFormulaDefs(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	formulaFields []models.FieldMetadata,
) ([]FormulaDefinition, error) {
	var defs []FormulaDefinition
	for _, f := range formulaFields {
		expr, lookupRules, err := s.formulaExpressionAndRulesForField(ctx, schema, f)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(expr) == "" {
			return nil, fmt.Errorf("active computed field %s.%s has no formula or projection expression", objectName, f.APIName)
		}
		if _, err := Parse(expr); err != nil {
			return nil, fmt.Errorf("parse active formula %s.%s: %w", objectName, f.APIName, err)
		}
		rt := "Text"
		if f.ReturnType != nil && string(*f.ReturnType) != "" {
			rt = string(*f.ReturnType)
		}
		defs = append(defs, FormulaDefinition{
			Name:             f.APIName,
			Expression:       expr,
			ReturnType:       rt,
			LookupValueRules: lookupRules,
		})
	}

	if len(defs) == 0 {
		return nil, fmt.Errorf("no active formulas to build for %s", objectName)
	}
	defs, err := s.attachSelfProjectionDeps(ctx, objectName, schema, defs)
	if err != nil {
		return nil, err
	}
	return defs, nil
}

// RecomputeAllValidationProjections rebuilds the validation-projection closure
// for every object with active computed fields, WITHOUT executing any view DDL.
//
// This is the rehydration path for the process-local projection cache. The cache
// is otherwise only populated as a side effect of RebuildView (DDL emission), so
// without this entry point the optimization is absent on the canonical flows:
//   - control-plane apply (RebuildDerived) ends with a final RefreshCache that
//     invalidates the cache RebuildAllViews just populated; and
//   - a plain process restart starts with an empty cache and never rebuilds views.
//
// Call this after the final metadata refresh (apply) and at startup so live
// validation uses the narrowed active-field set instead of falling back to all
// active computed fields on every save.
func (s *FormulaViewService) RecomputeAllValidationProjections(ctx context.Context) error {
	if s == nil || s.metadata == nil || s.validationProjection == nil {
		return nil
	}
	// Capture the projection generation and read all schemas as one epoch. The
	// helper captures the generation BEFORE the read, so any invalidation after
	// the snapshot (a metadata change, or a concurrent RebuildView publishing a
	// fresh closure) bumps the generation and every generation-guarded store below
	// drops rather than overwriting fresh state or mixing old defs with newly-read
	// rules. The whole recompute therefore runs against a single epoch.
	gen, schemas := s.projectionEpochForAll(ctx)
	var errs []string
	for _, schema := range schemas {
		if schema == nil || !formulastorage.HasActiveComputedFields(schema) {
			continue
		}
		formulaFields := formulastorage.ActiveComputedFields(schema)
		defs, err := s.buildFormulaDefs(ctx, schema.APIName, schema, formulaFields)
		if err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", schema.APIName, err))
			continue
		}
		s.computeValidationProjectionClosure(ctx, schema.APIName, schema, defs, gen)
	}
	if len(errs) > 0 {
		return fmt.Errorf("validation projection recompute errors: %s", strings.Join(errs, "; "))
	}
	return nil
}

// EnsureViews checks all objects for active formula/rollup fields and creates
// missing Views. Unlike RebuildAllViews which always recreates, this only
// builds Views that don't already exist. Safe to call on every server startup.
// Views are created in dependency order computed from live metadata.
func (s *FormulaViewService) EnsureViews(ctx context.Context) (created []string, errs []string) {
	schemas := s.metadata.GetSchemas(ctx)

	// Collect objects that need a view but don't have one yet.
	var candidates []string
	for _, schema := range schemas {
		if formulastorage.HasActiveComputedFields(schema) && !s.ViewExists(ctx, schema.APIName) {
			candidates = append(candidates, schema.APIName)
		}
	}
	if len(candidates) == 0 {
		return nil, nil
	}

	// Order by live-metadata dependency graph. Fails closed on cycles or missing
	// metadata — no lexical / pack-payload fallback per design Forbidden Move #5/#11.
	ordered, err := s.OrderForBuild(ctx, candidates)
	if err != nil {
		errs = append(errs, fmt.Sprintf("compute formula view build order: %v", err))
		return created, errs
	}

	for _, name := range ordered {
		if err := s.RebuildView(ctx, name); err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", name, err))
			s.logger.Warn(fmt.Sprintf("Failed to create view for %s: %v", name, err))
		} else {
			created = append(created, name)
			s.logger.Info(fmt.Sprintf("Created missing view v_%s", strings.ToLower(name)))
		}
	}

	return created, errs
}

// ---- Internal helpers ----

func (s *FormulaViewService) formulaExpressionAndRulesForField(
	ctx context.Context,
	schema *models.ObjectMetadata,
	f models.FieldMetadata,
) (string, []LookupValueRule, error) {
	formula := ""
	if f.Formula != nil {
		formula = strings.TrimSpace(*f.Formula)
	}
	if f.ProjectionConfig == nil {
		return formula, nil, nil
	}
	if formula != "" {
		return "", nil, fmt.Errorf("%s.%s cannot declare both formula and projection", schema.APIName, f.APIName)
	}

	return s.lookupValueProjectionExpressionAndRules(ctx, schema, f.APIName, f.ProjectionConfig)
}

func (s *FormulaViewService) lookupValueProjectionExpressionAndRules(
	ctx context.Context,
	schema *models.ObjectMetadata,
	fieldName string,
	projection *models.FieldProjectionConfig,
) (string, []LookupValueRule, error) {
	if projection == nil {
		return "", nil, nil
	}
	if !strings.EqualFold(projection.Kind, projectionKindLookupValue) {
		return "", nil, fmt.Errorf("%s.%s projection kind %q is unsupported", schema.APIName, fieldName, projection.Kind)
	}
	if projection.TargetObject == "" {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection missing targetObject", schema.APIName, fieldName)
	}
	if projection.SelectField == "" {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection missing selectField", schema.APIName, fieldName)
	}
	if len(projection.Matches) == 0 {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection requires at least one match", schema.APIName, fieldName)
	}
	if err := validateIdentifier(projection.TargetObject, "lookupValue projection targetObject"); err != nil {
		return "", nil, err
	}
	if err := validateIdentifier(projection.SelectField, "lookupValue projection selectField"); err != nil {
		return "", nil, err
	}
	if !fieldExists(schema, fieldName) {
		return "", nil, fmt.Errorf("%s.%s projection output field does not exist on source schema", schema.APIName, fieldName)
	}

	if s.metadata == nil {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection requires metadata service", schema.APIName, fieldName)
	}
	targetSchema := s.metadata.GetSchema(ctx, projection.TargetObject)
	if targetSchema == nil {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection target object %q not found", schema.APIName, fieldName, projection.TargetObject)
	}
	if !fieldExists(targetSchema, projection.SelectField) {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection select field %q not found on %s", schema.APIName, fieldName, projection.SelectField, projection.TargetObject)
	}
	if projection.ExcludeDeleted && !fieldExists(targetSchema, constants.FieldIsDeleted) {
		return "", nil, fmt.Errorf("%s.%s lookupValue projection excludeDeleted requires %s.%s", schema.APIName, fieldName, projection.TargetObject, constants.FieldIsDeleted)
	}

	args := []string{strconv.Quote(projection.TargetObject), strconv.Quote(projection.SelectField)}
	rule := LookupValueRule{
		TargetTable:    projection.TargetObject,
		OutputField:    projection.SelectField,
		ExcludeDeleted: projection.ExcludeDeleted,
		MatchPairs:     make([]LookupValueMatchRule, 0, len(projection.Matches)),
	}

	for i, match := range projection.Matches {
		if match.TargetField == "" || match.SourceField == "" {
			return "", nil, fmt.Errorf("%s.%s lookupValue projection match %d requires targetField and sourceField", schema.APIName, fieldName, i+1)
		}
		if err := validateIdentifier(match.TargetField, "lookupValue projection targetField"); err != nil {
			return "", nil, err
		}
		if err := validateIdentifier(match.SourceField, "lookupValue projection sourceField"); err != nil {
			return "", nil, err
		}
		if !fieldExists(targetSchema, match.TargetField) {
			return "", nil, fmt.Errorf("%s.%s lookupValue projection target field %q not found on %s", schema.APIName, fieldName, match.TargetField, projection.TargetObject)
		}
		if !fieldExists(schema, match.SourceField) {
			return "", nil, fmt.Errorf("%s.%s lookupValue projection source field %q not found", schema.APIName, fieldName, match.SourceField)
		}
		args = append(args, strconv.Quote(match.TargetField), match.SourceField)
		rule.MatchPairs = append(rule.MatchPairs, LookupValueMatchRule{
			MatchField:  match.TargetField,
			SourceField: match.SourceField,
		})
	}

	return fmt.Sprintf("LOOKUPVALUE(%s)", strings.Join(args, ", ")), []LookupValueRule{rule}, nil
}

func fieldExists(schema *models.ObjectMetadata, fieldName string) bool {
	if schema == nil {
		return false
	}
	for _, f := range schema.Fields {
		if strings.EqualFold(f.APIName, fieldName) {
			return true
		}
	}
	return false
}

// deriveRelationshipName converts a lookup field name to its SF relationship name.
// ParentId → Parent, Custom_Lookup__c → Custom_Lookup__r
func deriveRelationshipName(fieldAPIName string) string {
	if strings.HasSuffix(fieldAPIName, "__c") {
		return strings.TrimSuffix(fieldAPIName, "__c") + "__r"
	}
	if strings.HasSuffix(fieldAPIName, "Id") {
		return strings.TrimSuffix(fieldAPIName, "Id")
	}
	return fieldAPIName
}

// resolveDeepJoinChains resolves multi-hop cross-table references (DeepCrossRef)
// into chained JOIN specifications by walking FK relationships through metadata.
// Example: Parent.Manager.Name on WorkItem →
//
//	Hop 0: WorkItem.ParentId → ParentObject (alias: dj_parent)
//	Hop 1: ParentObject.ManagerId → PersonObject (alias: dj_manager)
//	Final: dj_manager.Name
func (s *FormulaViewService) resolveDeepJoinChains(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	defs []FormulaDefinition,
) ([]DeepJoinChain, error) {
	// Collect all unique DeepCrossRef nodes from all formula ASTs
	seen := make(map[string]bool)
	var allDCRs []DeepCrossRef
	for _, def := range defs {
		ast, err := Parse(def.Expression)
		if err != nil {
			return nil, fmt.Errorf("parse active formula %s.%s: %w", objectName, def.Name, err)
		}
		for _, dcr := range extractDeepCrossRefs(ast) {
			key := strings.ToLower(strings.Join(dcr.Parts, "."))
			if !seen[key] {
				seen[key] = true
				allDCRs = append(allDCRs, dcr)
			}
		}
	}
	if len(allDCRs) == 0 {
		return nil, nil
	}

	usedAliases := make(map[string]bool)
	var chains []DeepJoinChain
	for _, dcr := range allDCRs {
		chain, err := s.resolveOneDeepChain(ctx, schema, dcr, usedAliases)
		if err != nil {
			return nil, fmt.Errorf("resolve deep reference %s.%s: %w", objectName, strings.Join(dcr.Parts, "."), err)
		}
		chains = append(chains, chain)
	}
	return chains, nil
}

func (s *FormulaViewService) resolveOneDeepChain(
	ctx context.Context,
	baseSchema *models.ObjectMetadata,
	dcr DeepCrossRef,
	usedAliases map[string]bool,
) (DeepJoinChain, error) {
	parts := dcr.Parts
	if len(parts) < 3 {
		return DeepJoinChain{}, fmt.Errorf("deep reference requires at least one relationship hop")
	}

	chain := DeepJoinChain{
		Parts: parts,
	}

	currentSchema := baseSchema
	sourceAlias := "t"
	var finalSchema *models.ObjectMetadata

	// Walk all intermediate hops (all parts except the last, which is the field)
	for i := 0; i < len(parts)-1; i++ {
		relName := parts[i]

		// Find the FK field matching this relationship on current schema
		fkField, targetObj := findFKForRelationship(currentSchema, relName)
		if fkField == "" || targetObj == "" {
			schemaName := ""
			if currentSchema != nil {
				schemaName = currentSchema.APIName
			}
			return DeepJoinChain{}, fmt.Errorf("cannot resolve hop %d %q on %s", i, relName, schemaName)
		}

		// For intermediate hops we only read the Id field (for joining) — always
		// a physical column — so route to the physical table. For the last hop
		// (i.e. the hop whose target exposes the final field), check whether the
		// final field is active-computed on the target and route to v_<obj> if so.
		isLastHop := i == len(parts)-2
		readsActiveComputedOnHop := false
		if isLastHop {
			hopSchema := s.metadata.GetSchema(ctx, targetObj)
			if hopSchema != nil {
				finalFieldName, err := resolveReferencedField(hopSchema, parts[len(parts)-1])
				if err != nil {
					return DeepJoinChain{}, err
				}
				readsActiveComputedOnHop = formulastorage.SchemaFieldIsActiveComputed(hopSchema, finalFieldName)
			}
		}
		isSelfRefFinalHop := isLastHop && strings.EqualFold(targetObj, baseSchema.APIName)
		if isSelfRefFinalHop && readsActiveComputedOnHop {
			actualTable, err := s.materializedTableName(ctx, targetObj)
			if err != nil {
				return DeepJoinChain{}, fmt.Errorf("target object %q is not materialized", targetObj)
			}
			alias := makeJoinAlias(relName, "dj_", usedAliases)
			hop := DeepJoinHop{
				Alias:                       alias,
				LogicalObject:               targetObj,
				TargetTable:                 actualTable,
				SourceAlias:                 sourceAlias,
				SourceField:                 fkField,
				TargetField:                 "Id",
				UsesCurrentObjectProjection: true,
			}
			chain.Hops = append(chain.Hops, hop)
			sourceAlias = alias
			nextSchema := s.metadata.GetSchema(ctx, targetObj)
			if nextSchema == nil {
				return DeepJoinChain{}, fmt.Errorf("no metadata for hop %d target %q", i, targetObj)
			}
			currentSchema = nextSchema
			finalSchema = nextSchema
			continue
		}
		actualTable, err := s.joinTargetForFields(ctx, targetObj, readsActiveComputedOnHop)
		if err != nil {
			return DeepJoinChain{}, fmt.Errorf("target object %q is not materialized", targetObj)
		}

		alias := makeJoinAlias(relName, "dj_", usedAliases)

		hop := DeepJoinHop{
			Alias:         alias,
			LogicalObject: targetObj,
			TargetTable:   actualTable,
			SourceAlias:   sourceAlias,
			SourceField:   fkField,
			TargetField:   "Id",
		}
		chain.Hops = append(chain.Hops, hop)

		// Advance to next hop: get schema for the target object
		if s == nil || s.metadata == nil {
			return DeepJoinChain{}, fmt.Errorf("metadata service is required for hop %d target %q", i, targetObj)
		}
		nextSchema := s.metadata.GetSchema(ctx, targetObj)
		if nextSchema == nil {
			return DeepJoinChain{}, fmt.Errorf("no metadata for hop %d target %q", i, targetObj)
		}
		if i < len(parts)-2 {
			currentSchema = nextSchema
		} else {
			finalSchema = nextSchema
		}
		sourceAlias = alias
	}

	finalField, err := resolveReferencedField(finalSchema, parts[len(parts)-1])
	if err != nil {
		return DeepJoinChain{}, err
	}
	chain.FinalAlias = chain.Hops[len(chain.Hops)-1].Alias
	chain.FinalField = finalField
	return chain, nil
}

// analyticsPhysicalColumnName normalises a field's PhysicalName for use in SQL
// generation, matching the semantics of persistence.actualColumnName.
// Backtick-wrapped values (e.g. "`RecipientId`") signal that the DB column IS the
// API name; they must not be emitted verbatim as physical aliases.
func analyticsPhysicalColumnName(apiName, physicalName string) string {
	trimmed := strings.TrimSpace(physicalName)
	if strings.HasPrefix(trimmed, "`") && strings.HasSuffix(trimmed, "`") {
		return apiName
	}
	return trimmed
}

func formulaViewProjectionKey(name string) string {
	return strings.ToLower(strings.TrimSpace(name))
}

// findFKForRelationship finds the FK field and target object for a relationship name.
// E.g., on any schema, "Parent" → ("ParentId", "ParentObject")
func findFKForRelationship(schema *models.ObjectMetadata, relName string) (fkField string, targetObj string) {
	if schema == nil {
		return "", ""
	}
	for _, f := range schema.Fields {
		if !constants.IsReferenceType(f.Type) {
			continue
		}
		target, ok := formulaFieldPrimaryTarget(f)
		if !ok {
			continue
		}
		for _, candidate := range relationshipTraversalNamesForField(f) {
			if strings.EqualFold(candidate, relName) {
				return f.APIName, target
			}
		}
	}
	return "", ""
}