package analytics

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

	"github.com/nexuscrm/backend/internal/features/metadata"
	"github.com/nexuscrm/backend/pkg/logging"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// 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 *metadata.MetadataService
	logger   *logging.Logger

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

// NewFormulaViewService creates a new FormulaViewService
func NewFormulaViewService(db *sql.DB, metadata *metadata.MetadataService, logger *logging.Logger) *FormulaViewService {
	return &FormulaViewService{
		db:       db,
		metadata: metadata,
		logger:   logger,
		viewSet:  make(map[string]bool),
	}
}

// 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 {
	schema := s.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return fmt.Errorf("object %s not found", objectName)
	}

	// Collect active formula fields
	formulaFields := s.getActiveFormulaFields(schema)

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

	// Strategy: DSL engine → legacy SF translator
	var ddl string
	var err error

	s.logger.Info(fmt.Sprintf("Using DSL engine for %s (%d formula fields)", objectName, len(formulaFields)))
	ddl, err = s.buildViewFromDSL(ctx, objectName, schema, formulaFields)
	if err != nil {
		s.logger.Warn(fmt.Sprintf("DSL engine failed for %s, falling back to legacy translator: %v", objectName, err))
		ddl, err = s.buildViewFromTranslator(ctx, objectName, schema, formulaFields)
	}

	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,
) (string, error) {
	// Convert FieldMetadata to DSL FormulaDefinitions
	var defs []FormulaDefinition
	for _, f := range formulaFields {
		if f.Formula == nil || *f.Formula == "" {
			continue
		}
		// Try to parse — skip fields that can't be parsed by DSL
		if _, err := Parse(*f.Formula); err != nil {
			s.logger.Warn(fmt.Sprintf("DSL: skipping unparseable formula %s.%s: %v", objectName, f.APIName, err))
			continue
		}
		rt := "Text"
		if f.ReturnType != nil && string(*f.ReturnType) != "" {
			rt = string(*f.ReturnType)
		}
		defs = append(defs, FormulaDefinition{
			Name:       f.APIName,
			Expression: *f.Formula,
			ReturnType: rt,
		})
	}

	if len(defs) == 0 {
		return "", fmt.Errorf("no parseable DSL formulas for %s", objectName)
	}

	// Collect base columns (non-formula fields) from metadata
	baseSet := make(map[string]bool)
	var baseColumns []string
	var joins []JoinSpec
	usedAliases := make(map[string]int)

	// 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[def.Name] = true
	}

	for _, f := range schema.Fields {
		if f.Type == "formula" && f.Formula != nil && *f.Formula != "" {
			continue // skip formula fields — they're computed
		}
		if !baseSet[f.APIName] {
			baseColumns = append(baseColumns, f.APIName)
			baseSet[f.APIName] = true
		}

		// Build JOINs from reference fields
		if f.Type == "reference" && len(f.ReferenceTo) > 0 && f.ReferenceTo[0] != "" {
			targetTable := f.ReferenceTo[0]
			relName := deriveRelationshipName(f.APIName)
			alias := generateAlias(relName)

			if cnt, exists := usedAliases[alias]; exists {
				usedAliases[alias] = cnt + 1
				alias = fmt.Sprintf("%s%d", alias, cnt+1)
			} else {
				usedAliases[alias] = 1
			}

			joins = append(joins, JoinSpec{
				Alias:       alias,
				TargetTable: targetTable,
				SourceField: f.APIName,
				TargetField: "Id",
			})
		}
	}

	// Supplement base columns from INFORMATION_SCHEMA.COLUMNS 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).
	physTable := strings.ToLower(objectName)
	physRows, err := s.db.QueryContext(ctx,
		`SELECT COLUMN_NAME FROM information_schema.COLUMNS
		 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
		 ORDER BY ORDINAL_POSITION`, physTable)
	if err == nil {
		defer physRows.Close()
		for physRows.Next() {
			var colName string
			if physRows.Scan(&colName) == nil && !baseSet[colName] && !computedFormulaNames[colName] {
				baseColumns = append(baseColumns, colName)
				baseSet[colName] = true
			}
		}
	}

	// Validate JOINs: remove any targeting non-existent tables
	var validJoins []JoinSpec
	for _, j := range joins {
		var exists int
		s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.TABLES
			WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = ?`, strings.ToLower(j.TargetTable)).Scan(&exists)
		if exists > 0 {
			validJoins = append(validJoins, j)
		}
	}

	// Auto-generate AggJoins from AggregateExpr nodes in formula ASTs
	var aggJoins []AggregateJoinSpec
	aggIdx := 0
	for _, def := range defs {
		ast, err := Parse(def.Expression)
		if err != nil {
			continue
		}
		for _, agg := range extractAggregates(ast) {
			// Resolve child table name
			childTable := strings.ToLower(agg.ChildObject)
			var actualTable string
			s.db.QueryRowContext(ctx, `SELECT TABLE_NAME FROM information_schema.TABLES
				WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = ? LIMIT 1`, childTable).Scan(&actualTable)
			if actualTable == "" {
				continue
			}

			// Resolve FK: convention is ParentObjectId (e.g. AccountId on Opportunity)
			fk := objectName + "Id"
			// Verify FK column exists
			var fkExists int
			s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.COLUMNS
				WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
				actualTable, fk).Scan(&fkExists)
			if fkExists == 0 {
				continue
			}

			aggField := agg.ChildField
			if aggField == "" {
				aggField = "*"
			}

			alias := fmt.Sprintf("rsq%d", aggIdx)
			outputAlias := fmt.Sprintf("agg_%d", aggIdx)
			aggJoins = append(aggJoins, AggregateJoinSpec{
				Alias:       alias,
				ChildTable:  actualTable,
				AggFunc:     agg.Func,
				AggField:    aggField,
				ForeignKey:  fk,
				ParentKey:   "Id",
				OutputAlias: outputAlias,
			})
			aggIdx++
		}
	}

	// Resolve deep cross-table references (multi-hop JOINs like Owner.Profile.Name)
	deepJoins := s.resolveDeepJoinChains(ctx, objectName, schema, defs)
	if len(deepJoins) > 0 {
		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.
	fieldTypes := s.buildFieldTypeMap(ctx, schema, validJoins, deepJoins)
	cfg := ViewBuildConfig{
		ObjectName:    objectName,
		PhysicalTable: strings.ToLower(objectName),
		TableAlias:    "t",
		BaseColumns:   baseColumns,
		Formulas:      defs,
		Joins:         validJoins,
		AggJoins:      aggJoins,
		DeepJoins:     deepJoins,
		FieldTypes:    fieldTypes,
	}

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

// buildFieldTypeMap assembles the FieldType lookup the compiler consults to
// decide whether a BinaryOp between two operands is date-arithmetic. Keys:
//   - "<field>"         for fields on the primary schema
//   - "<object>.<field>" for fields reachable via a 1-hop JOIN
//   - "<a>.<b>.<c>"      for fields reachable via a multi-hop DeepJoin
//
// Unresolved keys default to empty string, which the compiler treats as
// unknown (safe fallback to the type-naive emission that existed before).
func (s *FormulaViewService) buildFieldTypeMap(
	ctx context.Context,
	schema *models.ObjectMetadata,
	joins []JoinSpec,
	deepJoins []DeepJoinChain,
) map[string]string {
	out := make(map[string]string)
	if schema != nil {
		for _, f := range schema.Fields {
			out[f.APIName] = effectiveFieldType(f)
		}
	}
	// 1-hop cross-refs: look up each joined object's schema by API name
	// (JOINs use TargetTable which may be lower-case physical table; we
	// need the logical object to match AST CrossRef.Object).
	for _, j := range joins {
		related := s.metadata.GetSchema(ctx, j.TargetTable)
		if related == nil {
			continue
		}
		// AST CrossRef.Object matches the relationship name's object token,
		// typically the PascalCase object API name. Record under both the
		// target-table lookup key and the object APIName for robustness.
		for _, f := range related.Fields {
			key := related.APIName + "." + f.APIName
			out[key] = effectiveFieldType(f)
		}
	}
	// Deep joins: type of the final field at the end of the multi-hop chain.
	for _, dj := range deepJoins {
		if len(dj.Parts) < 2 {
			continue
		}
		// The final hop's target table carries the field. Hops[last].TargetTable
		// holds it (Parts[last] is the final field, not an object).
		if len(dj.Hops) == 0 {
			continue
		}
		finalTable := dj.Hops[len(dj.Hops)-1].TargetTable
		related := s.metadata.GetSchema(ctx, finalTable)
		if related == nil {
			continue
		}
		for _, f := range related.Fields {
			if f.APIName == dj.FinalField {
				out[strings.Join(dj.Parts, ".")] = effectiveFieldType(f)
				break
			}
		}
	}
	return out
}

// effectiveFieldType returns the type the compiler should reason about when
// emitting SQL. For formula fields the raw Type is "formula", which tells the
// compiler nothing about operand shape; ReturnType (e.g. "date", "double") is
// the actual value type the formula produces. Without this, a formula like
// `Overall_End_Date__c - Overall_Start_Date__c` — where both operands are
// themselves date-valued formulas — falls back to MySQL's YYYYMMDD numeric
// subtraction instead of DATEDIFF.
func effectiveFieldType(f models.FieldMetadata) string {
	t := strings.ToLower(string(f.Type))
	if t == "formula" && f.ReturnType != nil {
		return strings.ToLower(string(*f.ReturnType))
	}
	return t
}

// buildViewFromTranslator generates View DDL using our SF formula → TiDB SQL translator.
// Used as fallback when DSL engine cannot parse certain formulas.
func (s *FormulaViewService) buildViewFromTranslator(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	formulaFields []models.FieldMetadata,
) (string, error) {
	baseFields := s.getBaseFields(schema)
	lookups := s.collectLookups(ctx, schema)
	layers := s.analyzeFormulaLayers(formulaFields)
	return s.buildViewDDLWithRollups(objectName, baseFields, formulaFields, lookups, layers, "", nil)
}

// 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()

	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
	}

	// 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
}

// ResolveTableName returns v_{object} if the object has an active view, otherwise returns the physical table 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 formula fields?
	hasActive := false
	for _, f := range schema.Fields {
		if f.FormulaMode != nil && *f.FormulaMode == "active" {
			hasActive = true
			break
		}
	}
	if !hasActive {
		return schema.APIName
	}

	// Check if view actually exists
	if s.ViewExists(ctx, schema.APIName) {
		return ViewName(schema.APIName)
	}

	return schema.APIName
}

// 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 {
		return "", fmt.Errorf("object %s not found", objectName)
	}

	formulaFields := s.getActiveFormulaFields(schema)

	if len(formulaFields) == 0 {
		return "", fmt.Errorf("object %s has no active formula fields", objectName)
	}

	baseFields := s.getBaseFields(schema)
	lookups := s.collectLookups(ctx, schema)
	layers := s.analyzeFormulaLayers(formulaFields)

	return s.buildViewDDLWithRollups(objectName, baseFields, formulaFields, lookups, layers, "", nil)
}

// RebuildAllViews rebuilds views for all objects that have active formula fields
func (s *FormulaViewService) RebuildAllViews(ctx context.Context) error {
	schemas := s.metadata.GetSchemas(ctx)
	var errs []string

	for _, schema := range schemas {
		formulaFields := s.getActiveFormulaFields(schema)
		if len(formulaFields) == 0 {
			continue
		}

		if err := s.RebuildView(ctx, schema.APIName); err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", schema.APIName, err))
			s.logger.Warn(fmt.Sprintf("Failed to rebuild view for %s: %v", schema.APIName, err))
		}
	}

	if len(errs) > 0 {
		return fmt.Errorf("view rebuild 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.
func (s *FormulaViewService) EnsureViews(ctx context.Context) (created []string, errs []string) {
	schemas := s.metadata.GetSchemas(ctx)

	for _, schema := range schemas {
		// Check if this object needs a View (has active formula fields)
		formulaFields := s.getActiveFormulaFields(schema)
		if len(formulaFields) == 0 {
			continue
		}

		// Skip if View already exists
		if s.ViewExists(ctx, schema.APIName) {
			continue
		}

		// Build the View
		if err := s.RebuildView(ctx, schema.APIName); err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", schema.APIName, err))
			s.logger.Warn(fmt.Sprintf("Failed to create view for %s: %v", schema.APIName, err))
		} else {
			created = append(created, schema.APIName)
			s.logger.Info(fmt.Sprintf("Created missing view v_%s", strings.ToLower(schema.APIName)))
		}
	}

	return created, errs
}

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

// getActiveFormulaFields returns formula fields with formulaMode=active AND non-empty formula
func (s *FormulaViewService) getActiveFormulaFields(schema *models.ObjectMetadata) []models.FieldMetadata {
	var result []models.FieldMetadata
	for _, f := range schema.Fields {
		if f.Formula == nil || *f.Formula == "" {
			continue
		}
		if f.FormulaMode == nil || *f.FormulaMode != "active" {
			continue
		}
		result = append(result, f)
	}
	return result
}

// getBaseFields returns non-formula fields (physical columns)
func (s *FormulaViewService) getBaseFields(schema *models.ObjectMetadata) []models.FieldMetadata {
	var result []models.FieldMetadata
	for _, f := range schema.Fields {
		isFormula := strings.EqualFold(string(f.Type), string(constants.FieldTypeFormula))
		if !isFormula {
			result = append(result, f)
		}
	}
	return result
}

// isPercentFormula returns true if this formula field's return type is "percent".
// SF percent fields store 0-100 in DB but formulas compute in 0-1 range.
// Views must multiply by 100 to match the physical column convention.
func isPercentFormula(f models.FieldMetadata) bool {
	return f.ReturnType != nil && strings.EqualFold(string(*f.ReturnType), string(constants.FieldTypePercent))
}

// wrapPercentExpr wraps a SQL expression with "* 100" if the field is a percent formula.
func wrapPercentExpr(sqlExpr string, f models.FieldMetadata) string {
	if isPercentFormula(f) {
		return fmt.Sprintf("(%s) * 100", sqlExpr)
	}
	return sqlExpr
}

// lookupJoin represents a LEFT JOIN needed for cross-table formula references
type lookupJoin struct {
	Alias       string // e.g. "o" for opportunity
	TargetTable string // e.g. "opportunity"
	SourceField string // e.g. "OpportunityId" (FK on source table)
	TargetField string // e.g. "Id" (PK on target table)
}

// collectLookups collects lookup relationships from the schema for cross-table JOINs.
// Returns a map keyed by both the target object name AND the SF relationship name
// (e.g., OwnerId → "User" and "Owner" both map to the same join).
func (s *FormulaViewService) collectLookups(ctx context.Context, schema *models.ObjectMetadata) map[string]lookupJoin {
	result := make(map[string]lookupJoin)

	for _, f := range schema.Fields {
		isLookup := strings.EqualFold(string(f.Type), string(constants.FieldTypeReference))
		if !isLookup || len(f.ReferenceTo) == 0 {
			continue
		}

		targetObj := f.ReferenceTo[0]
		// Derive the SF relationship name from the field name
		// OwnerId → Owner, AccountId → Account, Custom_Field__c → Custom_Field__r
		relName := deriveRelationshipName(f.APIName)

		alias := generateAlias(relName) // Use relationship name for alias (e.g., "own" for Owner)
		join := lookupJoin{
			Alias:       alias,
			TargetTable: targetObj,
			SourceField: f.APIName,
			TargetField: constants.FieldID,
		}

		result[targetObj] = join
		if relName != targetObj {
			result[relName] = join // Also key by relationship name
		}
	}

	return result
}

// deriveRelationshipName converts a lookup field name to its SF relationship name.
// OwnerId → Owner, AccountId → Account, CreatedById → CreatedBy,
// 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
}

// generateAlias creates a short alias for a table name
func generateAlias(tableName string) string {
	lower := strings.ToLower(tableName)
	// Common mappings
	switch {
	case strings.HasPrefix(lower, "opportunity"):
		return "o"
	case strings.HasPrefix(lower, "account"):
		return "a"
	case strings.HasPrefix(lower, "contact"):
		return "ct"
	case lower == "user":
		return "u"
	case lower == "owner":
		return "own"
	case lower == "createdby":
		return "cb"
	case lower == "lastmodifiedby":
		return "lmb"
	case strings.HasPrefix(lower, "lead"):
		return "ld"
	case strings.HasPrefix(lower, "quote"):
		return "qt"
	case strings.HasPrefix(lower, "order"):
		return "ord"
	case strings.HasPrefix(lower, "contract"):
		return "ctr"
	case strings.HasPrefix(lower, "campaign"):
		return "cmp"
	default:
		// Use first 3 chars of the lowercased name
		if len(lower) > 3 {
			return lower[:3]
		}
		return lower
	}
}

// resolveDeepJoinChains resolves multi-hop cross-table references (DeepCrossRef)
// into chained JOIN specifications by walking FK relationships through metadata.
// Example: Owner.Profile.Name on Account →
//
//	Hop 0: Account.OwnerId → User (alias: dj_own)
//	Hop 1: User.ProfileId → Profile (alias: dj_prof)
//	Final: dj_prof.Name
func (s *FormulaViewService) resolveDeepJoinChains(
	ctx context.Context,
	objectName string,
	schema *models.ObjectMetadata,
	defs []FormulaDefinition,
) []DeepJoinChain {
	// 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 {
			continue
		}
		for _, dcr := range extractDeepCrossRefs(ast) {
			key := strings.Join(dcr.Parts, ".")
			if !seen[key] {
				seen[key] = true
				allDCRs = append(allDCRs, dcr)
			}
		}
	}
	if len(allDCRs) == 0 {
		return nil
	}

	usedAliases := make(map[string]bool)
	var chains []DeepJoinChain
	for _, dcr := range allDCRs {
		chain, ok := s.resolveOneDeepChain(ctx, schema, dcr, usedAliases)
		if ok {
			chains = append(chains, chain)
		}
	}
	return chains
}

func (s *FormulaViewService) resolveOneDeepChain(
	ctx context.Context,
	baseSchema *models.ObjectMetadata,
	dcr DeepCrossRef,
	usedAliases map[string]bool,
) (DeepJoinChain, bool) {
	parts := dcr.Parts
	if len(parts) < 3 {
		return DeepJoinChain{}, false
	}

	chain := DeepJoinChain{
		Parts:      parts,
		FinalField: parts[len(parts)-1],
	}

	currentSchema := baseSchema
	sourceAlias := "t"

	// 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 == "" {
			s.logger.Warn(fmt.Sprintf("DeepJoin: cannot resolve hop %d '%s' on %s", i, relName, currentSchema.APIName))
			return DeepJoinChain{}, false
		}

		// Verify target table exists in TiDB
		var tableExists int
		s.db.QueryRowContext(ctx,
			`SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND LOWER(TABLE_NAME) = ?`,
			strings.ToLower(targetObj)).Scan(&tableExists)
		if tableExists == 0 {
			s.logger.Warn(fmt.Sprintf("DeepJoin: target table %s not found", targetObj))
			return DeepJoinChain{}, false
		}

		// Generate unique alias with "dj_" prefix
		alias := "dj_" + generateAlias(relName)
		for usedAliases[alias] {
			alias = alias + "x"
		}
		usedAliases[alias] = true

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

		// Advance to next hop: get schema for the target object
		if i < len(parts)-2 {
			nextSchema := s.metadata.GetSchema(ctx, targetObj)
			if nextSchema == nil {
				s.logger.Warn(fmt.Sprintf("DeepJoin: no metadata for %s", targetObj))
				return DeepJoinChain{}, false
			}
			currentSchema = nextSchema
		}
		sourceAlias = alias
	}

	chain.FinalAlias = chain.Hops[len(chain.Hops)-1].Alias
	return chain, true
}

// findFKForRelationship finds the FK field and target object for a relationship name.
// E.g., on User schema, "Profile" → ("ProfileId", "Profile")
// E.g., on Account schema, "Owner" → ("OwnerId", "User")
func findFKForRelationship(schema *models.ObjectMetadata, relName string) (fkField string, targetObj string) {
	for _, f := range schema.Fields {
		if string(f.Type) != "reference" || len(f.ReferenceTo) == 0 {
			continue
		}
		derived := deriveRelationshipName(f.APIName)
		if strings.EqualFold(derived, relName) {
			return f.APIName, f.ReferenceTo[0]
		}
	}
	return "", ""
}

// buildJoinAliasMap converts []lookupJoin to map[name]alias for TranslateFormula.
// Includes both the target table name and the relationship name derived from SourceField.
func buildJoinAliasMap(joins []lookupJoin) map[string]string {
	m := make(map[string]string)
	for _, j := range joins {
		m[j.TargetTable] = j.Alias
		// Also add the relationship name: OwnerId→Owner, AccountId→Account
		relName := deriveRelationshipName(j.SourceField)
		m[relName] = j.Alias
	}
	return m
}

// formulaLayer groups formula fields by their dependency depth
type formulaLayer struct {
	Depth  int
	Fields []models.FieldMetadata
}

// analyzeFormulaLayers performs topological sort on formula dependencies.
// Fields that only reference base fields go to L1, fields that reference L1 go to L2, etc.
func (s *FormulaViewService) analyzeFormulaLayers(formulaFields []models.FieldMetadata) []formulaLayer {
	// Build set of formula field names for dependency detection
	formulaNames := make(map[string]bool, len(formulaFields))
	for _, f := range formulaFields {
		formulaNames[strings.ToLower(f.APIName)] = true
	}

	// Compute depth for each formula field
	depths := make(map[string]int, len(formulaFields))
	for _, f := range formulaFields {
		depths[strings.ToLower(f.APIName)] = s.computeDepth(f, formulaFields, formulaNames, make(map[string]bool))
	}

	// Group by depth
	depthGroups := make(map[int][]models.FieldMetadata)
	maxDepth := 0
	for _, f := range formulaFields {
		d := depths[strings.ToLower(f.APIName)]
		depthGroups[d] = append(depthGroups[d], f)
		if d > maxDepth {
			maxDepth = d
		}
	}

	// Convert to sorted layers
	layers := make([]formulaLayer, 0, maxDepth+1)
	for d := 0; d <= maxDepth; d++ {
		if fields, ok := depthGroups[d]; ok {
			layers = append(layers, formulaLayer{Depth: d, Fields: fields})
		}
	}

	return layers
}

// computeDepth calculates the dependency depth for a formula field.
// A field that only references base fields has depth 0.
// A field that references another formula field of depth N has depth N+1.
func (s *FormulaViewService) computeDepth(field models.FieldMetadata, allFormulas []models.FieldMetadata, formulaNames map[string]bool, visited map[string]bool) int {
	key := strings.ToLower(field.APIName)
	if visited[key] {
		return 0 // Break circular reference
	}
	visited[key] = true

	if field.Formula == nil {
		return 0
	}
	expr := strings.ToLower(*field.Formula)

	maxDep := -1
	for _, other := range allFormulas {
		otherKey := strings.ToLower(other.APIName)
		if otherKey == key {
			continue
		}

		// Check if this formula references the other formula field
		if strings.Contains(expr, strings.ToLower(other.APIName)) {
			dep := s.computeDepth(other, allFormulas, formulaNames, visited)
			if dep > maxDep {
				maxDep = dep
			}
		}
	}

	if maxDep < 0 {
		return 0 // No formula dependencies → base layer
	}
	return maxDep + 1
}

// buildViewDDL generates the complete CREATE OR REPLACE VIEW DDL (no rollups).
func (s *FormulaViewService) buildViewDDL(
	objectName string,
	baseFields []models.FieldMetadata,
	formulaFields []models.FieldMetadata,
	lookups map[string]lookupJoin,
	layers []formulaLayer,
) (string, error) {
	return s.buildViewDDLWithRollups(objectName, baseFields, formulaFields, lookups, layers, "", nil)
}

// buildViewDDLWithRollups generates the complete CREATE OR REPLACE VIEW DDL
// using nested subqueries for dependency layering, plus rollup subquery JOINs.
func (s *FormulaViewService) buildViewDDLWithRollups(
	objectName string,
	baseFields []models.FieldMetadata,
	formulaFields []models.FieldMetadata,
	lookups map[string]lookupJoin,
	layers []formulaLayer,
	rollupJoinSQL string,
	rollupCols map[string]string, // parentFieldName → "rsq.`alias`"
) (string, error) {
	viewName := ViewName(objectName)
	tableAlias := "t"

	if len(layers) == 0 && len(rollupCols) == 0 {
		return "", fmt.Errorf("no formula layers or rollups to build")
	}

	// Sort base fields by name for deterministic DDL
	sort.Slice(baseFields, func(i, j int) bool {
		return baseFields[i].APIName < baseFields[j].APIName
	})

	// Collect which lookups are actually needed by formula expressions
	neededJoins := s.collectNeededJoins(formulaFields, lookups)

	// For zero formula layers but with rollups only
	if len(layers) == 0 {
		return s.buildRollupOnlyDDL(viewName, objectName, tableAlias, baseFields, neededJoins, rollupJoinSQL, rollupCols)
	}

	// For single layer without rollups: simple
	if len(layers) == 1 && len(rollupCols) == 0 {
		return s.buildSingleLayerDDL(viewName, objectName, tableAlias, layers[0].Fields, neededJoins)
	}

	return s.buildMultiLayerDDL(viewName, objectName, tableAlias, layers, neededJoins, baseFields, rollupJoinSQL, rollupCols)
}

// buildRollupOnlyDDL generates a View with only rollup subqueries, no formula layers.
func (s *FormulaViewService) buildRollupOnlyDDL(viewName, tableName, alias string, baseFields []models.FieldMetadata, joins []lookupJoin, rollupJoinSQL string, rollupCols map[string]string) (string, error) {
	if err := validateIdentifier(viewName, "view"); err != nil {
		return "", err
	}
	if err := validateIdentifier(tableName, "table"); err != nil {
		return "", err
	}
	var sb strings.Builder
	baseColList := buildBaseFieldList(alias, baseFields)
	sb.WriteString(fmt.Sprintf("%s `%s` AS\nSELECT\n    %s", CreateOrReplaceViewStmt, viewName, baseColList))

	for fieldName, colExpr := range rollupCols {
		sb.WriteString(fmt.Sprintf(",\n    COALESCE(%s, 0) AS `%s`", colExpr, fieldName))
	}

	sb.WriteString(fmt.Sprintf("\nFROM `%s` %s", tableName, alias))
	for _, j := range joins {
		sb.WriteString(fmt.Sprintf("\nLEFT JOIN `%s` %s ON %s.`%s` = %s.`%s`",
			j.TargetTable, j.Alias, alias, j.SourceField, j.Alias, j.TargetField))
	}
	sb.WriteString(rollupJoinSQL)

	return sb.String(), nil
}

// collectNeededJoins determines which lookup JOINs are actually referenced by formulas
func (s *FormulaViewService) collectNeededJoins(formulaFields []models.FieldMetadata, lookups map[string]lookupJoin) []lookupJoin {
	needed := make(map[string]lookupJoin)

	for _, f := range formulaFields {
		if f.Formula == nil {
			continue
		}
		expr := strings.ToLower(*f.Formula)

		for objName, join := range lookups {
			// Check if formula references the related object
			// SF format: ObjectName.FieldName or ObjectName__r.FieldName
			lowerObj := strings.ToLower(objName)
			relName := strings.TrimSuffix(lowerObj, "__c") + "__r"

			if strings.Contains(expr, lowerObj+".") || strings.Contains(expr, relName+".") {
				needed[objName] = join
			}
		}
	}

	// Sort for deterministic output, dedup by alias
	seen := make(map[string]bool)
	result := make([]lookupJoin, 0, len(needed))
	for _, j := range needed {
		if seen[j.Alias] {
			continue
		}
		seen[j.Alias] = true
		result = append(result, j)
	}
	sort.Slice(result, func(i, j int) bool {
		return result[i].TargetTable < result[j].TargetTable
	})

	return result
}

// buildBaseFieldList generates the explicit column list for base fields, replacing t.*
// to avoid duplicate column name errors when the physical table has formula snapshot columns.
func buildBaseFieldList(alias string, baseFields []models.FieldMetadata) string {
	cols := make([]string, len(baseFields))
	for i, f := range baseFields {
		cols[i] = fmt.Sprintf("%s.`%s`", alias, f.APIName)
	}
	return strings.Join(cols, ", ")
}

func (s *FormulaViewService) buildSingleLayerDDL(viewName, tableName, alias string, fields []models.FieldMetadata, joins []lookupJoin) (string, error) {
	if err := validateIdentifier(viewName, "view"); err != nil {
		return "", err
	}
	if err := validateIdentifier(tableName, "table"); err != nil {
		return "", err
	}
	var sb strings.Builder
	joinAliasMap := buildJoinAliasMap(joins)

	sb.WriteString(fmt.Sprintf("%s `%s` AS\nSELECT\n    %s.*", CreateOrReplaceViewStmt, viewName, alias))

	// Add formula expressions
	for _, f := range fields {
		sqlExpr, err := TranslateFormula(*f.Formula, alias, joinAliasMap)
		if err != nil {
			// Only fall back to physical column if the field is not an active formula
			// (active formulas have no physical column).
			if f.FormulaMode != nil && *f.FormulaMode == "active" {
				s.logger.Warn(fmt.Sprintf("TranslateFormula failed for active formula %s: %v — skipping (no physical column)", f.APIName, err))
				continue
			}
			s.logger.Warn(fmt.Sprintf("TranslateFormula fallback for %s: %v — using physical column", f.APIName, err))
			sb.WriteString(fmt.Sprintf(",\n    %s.`%s`", alias, f.APIName))
			continue
		}
		sqlExpr = wrapPercentExpr(sqlExpr, f)
		sb.WriteString(fmt.Sprintf(",\n    %s AS `%s`", sqlExpr, f.APIName))
	}

	sb.WriteString(fmt.Sprintf("\nFROM `%s` %s", tableName, alias))

	// Add JOINs
	for _, j := range joins {
		sb.WriteString(fmt.Sprintf("\nLEFT JOIN `%s` %s ON %s.`%s` = %s.`%s`",
			j.TargetTable, j.Alias, alias, j.SourceField, j.Alias, j.TargetField))
	}

	return sb.String(), nil
}

func (s *FormulaViewService) buildMultiLayerDDL(viewName, tableName, baseAlias string, layers []formulaLayer, joins []lookupJoin, baseFields []models.FieldMetadata, rollupJoinSQL string, rollupCols map[string]string) (string, error) {
	if err := validateIdentifier(viewName, "view"); err != nil {
		return "", err
	}
	if err := validateIdentifier(tableName, "table"); err != nil {
		return "", err
	}
	// Build from innermost (L0) to outermost (Ln)
	// L0: SELECT <explicit base cols>, <rollup cols>, <L0 formulas> FROM table t LEFT JOIN ... LEFT JOIN (rollup subquery)
	// L1: SELECT L0.*, <L1 formulas> FROM (L0) L0
	// Ln: SELECT Ln-1.*, <Ln formulas> FROM (Ln-1) Ln-1

	numLayers := len(layers)

	// Build innermost subquery (L0) — use explicit column list to avoid duplicates
	innerAlias := baseAlias
	var innerSQL strings.Builder

	joinAliasMap := buildJoinAliasMap(joins)

	// Use explicit base field list instead of t.* to avoid duplicate formula column names
	baseColList := buildBaseFieldList(innerAlias, baseFields)
	innerSQL.WriteString(fmt.Sprintf("SELECT\n        %s", baseColList))

	// Add rollup columns in L0
	for fieldName, colExpr := range rollupCols {
		innerSQL.WriteString(fmt.Sprintf(",\n        COALESCE(%s, 0) AS `%s`", colExpr, fieldName))
	}
	for _, f := range layers[0].Fields {
		sqlExpr, err := TranslateFormula(*f.Formula, innerAlias, joinAliasMap)
		if err != nil {
			if f.FormulaMode != nil && *f.FormulaMode == "active" {
				s.logger.Warn(fmt.Sprintf("TranslateFormula failed for active formula %s (L0): %v — skipping", f.APIName, err))
				continue
			}
			s.logger.Warn(fmt.Sprintf("TranslateFormula fallback for %s (L0): %v — using physical column", f.APIName, err))
			innerSQL.WriteString(fmt.Sprintf(",\n        %s.`%s`", innerAlias, f.APIName))
			continue
		}
		sqlExpr = wrapPercentExpr(sqlExpr, f)
		innerSQL.WriteString(fmt.Sprintf(",\n        %s AS `%s`", sqlExpr, f.APIName))
	}
	innerSQL.WriteString(fmt.Sprintf("\n    FROM `%s` %s", tableName, innerAlias))

	// JOINs only in innermost layer (they reference the physical table)
	for _, j := range joins {
		innerSQL.WriteString(fmt.Sprintf("\n    LEFT JOIN `%s` %s ON %s.`%s` = %s.`%s`",
			j.TargetTable, j.Alias, innerAlias, j.SourceField, j.Alias, j.TargetField))
	}

	// Rollup subquery JOINs
	if rollupJoinSQL != "" {
		innerSQL.WriteString(rollupJoinSQL)
	}

	currentSubquery := innerSQL.String()

	// Build middle layers (L1 to Ln-1)
	for i := 1; i < numLayers; i++ {
		layerAlias := fmt.Sprintf("L%d", i-1)
		var layerSQL strings.Builder

		layerSQL.WriteString(fmt.Sprintf("SELECT\n        %s.*", layerAlias))
		for _, f := range layers[i].Fields {
			sqlExpr, err := TranslateFormula(*f.Formula, layerAlias, joinAliasMap)
			if err != nil {
				if f.FormulaMode != nil && *f.FormulaMode == "active" {
					s.logger.Warn(fmt.Sprintf("TranslateFormula failed for active formula %s (L%d): %v — skipping", f.APIName, i, err))
					continue
				}
				s.logger.Warn(fmt.Sprintf("TranslateFormula fallback for %s (L%d): %v — using physical column", f.APIName, i, err))
				layerSQL.WriteString(fmt.Sprintf(",\n        %s.`%s`", layerAlias, f.APIName))
				continue
			}
			sqlExpr = wrapPercentExpr(sqlExpr, f)
			layerSQL.WriteString(fmt.Sprintf(",\n        %s AS `%s`", sqlExpr, f.APIName))
		}
		layerSQL.WriteString(fmt.Sprintf("\n    FROM (\n    %s\n    ) %s", currentSubquery, layerAlias))

		currentSubquery = layerSQL.String()
	}

	// Wrap in final CREATE VIEW
	outerAlias := fmt.Sprintf("L%d", numLayers-2)
	if numLayers == 2 {
		outerAlias = "L0"
	}

	var finalSQL strings.Builder
	finalSQL.WriteString(fmt.Sprintf("%s `%s` AS\nSELECT %s.*\nFROM (\n%s\n) %s",
		CreateOrReplaceViewStmt, viewName, outerAlias, currentSubquery, outerAlias))

	return finalSQL.String(), nil
}