package analytics

import (
	"context"
	"fmt"
	"regexp"
	"strconv"
	"strings"

	pkgErrors "github.com/nexuscrm/backend/pkg/errors"
	"github.com/nexuscrm/backend/pkg/query"
	"github.com/nexuscrm/backend/pkg/utils"
	"github.com/nexuscrm/shared/pkg/models"
)

// executeJoinReport handles cross-object reports that require SQL JOINs.
// It builds a multi-table query from the JoinTemplate and executes it via raw SQL.
func (s *ReportService) executeJoinReport(
	ctx context.Context,
	def *models.ReportDefinition,
	tmpl *JoinTemplate,
	user *models.UserSession,
	runtimeFilters map[string]interface{},
	limit int,
) (*models.ReportResult, error) {
	// Build the query using the query builder.
	// The builder generates "FROM `table`" without alias, so the primary table
	// is referenced by its physical name (not a short alias) in the generated SQL.
	// JoinStep ON clauses use the PrimaryAlias, which we resolve to the table name.
	b := query.From(tmpl.PrimaryTable).SkipID()

	// Validate join types to prevent SQL injection via exported JoinTemplate struct.
	validJoinTypes := map[string]bool{"INNER JOIN": true, "LEFT JOIN": true}
	for _, j := range tmpl.Joins {
		if !validJoinTypes[j.JoinType] {
			return nil, pkgErrors.NewInternalError(fmt.Sprintf("invalid join type: %q", j.JoinType), nil)
		}
	}

	// Add JOINs with IsDeleted filtering on each joined table.
	for _, j := range tmpl.Joins {
		onClause := fmt.Sprintf("%s = %s AND `%s`.`IsDeleted` = 0",
			backtickJoinRef(j.OnLeft, tmpl), backtickJoinRef(j.OnRight, tmpl), j.Alias)
		b.Join(j.JoinType, j.Table, j.Alias, onClause)
	}

	// Resolve SELECT columns from the report definition.
	seen := make(map[string]bool)

	for _, col := range def.Columns {
		if seen[col] {
			continue
		}
		seen[col] = true

		qc, found := resolveQualifiedColumnDef(col, tmpl)
		if found {
			resultAlias := joinResultAlias(col, qc)
			b.AddSelectRaw(joinSelectExpression(qc, tmpl), resultAlias)
		} else {
			// Column not in template — skip unknown cross-object columns
			// rather than guessing the wrong table and causing SQL errors.
			// Use NULL placeholder so the column count stays consistent.
			b.AddSelectRaw("NULL", joinResultAlias(col, QualifiedColumn{}))
		}
	}

	// If no columns specified, select a useful default set.
	if len(def.Columns) == 0 {
		b.AddSelectRaw(fmt.Sprintf("`%s`.`Id`", tmpl.PrimaryTable), "Id")
	}

	// Always project the primary entity Id so the REST layer can dedupe
	// primary-side aggregates. JOIN fan-out produces one joined row per
	// child record, but Salesforce computes SUM/AVG/MIN/MAX of primary
	// fields once per primary entity; without this key the aggregates
	// silently inflate. Results are keyed as `__primary_id__` to avoid
	// colliding with any field the report actually selects.
	const primaryIdRowKey = "__primary_id__"
	b.AddSelectRaw(fmt.Sprintf("`%s`.`Id`", tmpl.PrimaryTable), primaryIdRowKey)

	// Resolve grouping fields.
	validGrouping := make([]string, 0, len(def.Grouping))
	for _, g := range def.Grouping {
		qc, found := resolveQualifiedColumnDef(g, tmpl)
		if found {
			if !seen[g] {
				b.AddSelectRaw(joinSelectExpression(qc, tmpl), joinResultAlias(g, qc))
				seen[g] = true
			}
			validGrouping = append(validGrouping, g)
		}
		// Skip unknown grouping columns — don't guess the wrong table.
	}

	// Build WHERE clauses from report filters (params added directly to builder).
	s.buildJoinWhere(b, def, tmpl)

	// Apply StandardDateFilter.
	if sdf := def.StandardDateFilter; sdf != nil && sdf.Column != "" {
		tblAlias, colName, found := resolveQualifiedColumn(sdf.Column, tmpl)
		if found {
			sqlAlias := resolveTableAlias(tblAlias, tmpl)
			qualifiedCol := fmt.Sprintf("`%s`.`%s`", sqlAlias, colName)
			startDate, endDate := s.normalizeDateBoundsForField(ctx, tableNameForAlias(tmpl, tblAlias), colName, sdf.StartDate, sdf.EndDate)
			if startDate != "" {
				b.Where(qualifiedCol+" >= ?", startDate)
			}
			if endDate != "" {
				b.Where(qualifiedCol+" <= ?", endDate)
			}
		}
	}

	// Apply runtime filters.
	if runtimeFilters != nil {
		for k, v := range runtimeFilters {
			if k == "__date_range" || k == "startDate" {
				continue
			}
			tblAlias, colName, found := resolveQualifiedColumn(k, tmpl)
			if !found {
				continue
			}
			sqlAlias := resolveTableAlias(tblAlias, tmpl)
			b.Where(fmt.Sprintf("`%s`.`%s` = ?", sqlAlias, colName), v)
		}
	}

	// IsDeleted on primary table.
	b.Where(fmt.Sprintf("`%s`.`IsDeleted` = 0", tmpl.PrimaryTable))

	// Limit to 2000 rows (SF behavior).
	b.Limit(limit)

	// Only apply DISTINCT when the report actually computes sums/avgs/etc
	// that would be inflated by JOIN fan-out. A plain "RowCount" aggregate
	// is a raw count of result rows and must match SF semantics, which do
	// NOT distinct the join output. Distinct'ing here silently dropped
	// rows with identical projected values — e.g. duplicate CampaignMember
	// entries on the same Campaign — producing an undercount vs SF.
	if needsDistinctForAggregates(def.Aggregates) {
		b.Distinct()
	}

	// Build and execute.
	qr := b.Build()
	sql := qr.SQL
	allParams := qr.Params

	if s.logger != nil {
		s.logger.Debug(fmt.Sprintf("executeJoinReport SQL: %s params: %v", sql, allParams))
	}
	rows, err := s.queryService.ExecuteRawSQL(ctx, sql, allParams, user)
	if err != nil {
		return nil, pkgErrors.NewInternalError(fmt.Sprintf("cross-object report query failed: %v", err), err)
	}

	// Separate COUNT query so Total Records reflects the true filtered total
	// rather than the LIMIT cap. Only runs when we hit the limit — otherwise
	// len(rows) is already the truth and we avoid the extra round-trip (and
	// the test-harness mock-call inflation).
	totalCount := len(rows)
	if len(rows) >= limit && limit > 0 {
		if innerSQL := stripLimitClause(sql); innerSQL != "" {
			countSQL := "SELECT COUNT(*) AS total FROM (" + innerSQL + ") _crt_count_sub"
			countRows, countErr := s.queryService.ExecuteRawSQL(ctx, countSQL, allParams, user)
			if countErr == nil && len(countRows) > 0 {
				if v, ok := countRows[0]["total"]; ok {
					totalCount = coerceInt(v)
				}
			}
		}
	}

	// Build result.
	resultDef := *def
	resultDef.Grouping = validGrouping

	// Map each aggregate field to its dedup row key. Primary-side fields
	// dedupe by the primary Id projected above; joined-side fields do not
	// dedupe (the join produces one row per joined entity, so the sum is
	// already correct). RowCount is not an aggregated field and never
	// appears here.
	dedupByAgg := buildAggregateDedupMap(def.Aggregates, tmpl)

	return &models.ReportResult{
		ReportMetadata:       resultDef,
		Rows:                 rows,
		TotalCount:           totalCount,
		AggregateDedupColumn: dedupByAgg,
	}, nil
}

// buildAggregateDedupMap walks the report's aggregate definitions and returns
// a lookup from canonical aggregate field name to the SObject row key the
// REST layer should dedupe on when summing. Only primary-entity fields get
// a dedup key — joined-side aggregates sum raw rows directly.
func buildAggregateDedupMap(aggs []models.ReportAggregate, tmpl *JoinTemplate) map[string]string {
	if len(aggs) == 0 || tmpl == nil {
		return nil
	}
	out := map[string]string{}
	for _, a := range aggs {
		if a.Field == "" {
			continue
		}
		qc, found := resolveQualifiedColumnDef(a.Field, tmpl)
		if !found {
			continue
		}
		// SelectExpr-only entries (decorate-installed polymorphic/lookup
		// markers) carry the join side implicitly via the expression; those
		// live on joined aliases and do not dedupe.
		if qc.TableAlias == "" {
			continue
		}
		if qc.TableAlias == tmpl.PrimaryAlias {
			out[a.Field] = "__primary_id__"
		}
	}
	return out
}

// stripLimitClause removes a trailing "LIMIT N" from a SQL string. Returns
// empty if the shape doesn't match — caller falls back to len(rows).
func stripLimitClause(sql string) string {
	idx := strings.LastIndex(strings.ToUpper(sql), " LIMIT ")
	if idx < 0 {
		return sql
	}
	return sql[:idx]
}

// coerceInt handles the variety of numeric types ScanRowsToSObjects may emit
// for a COUNT(*) result (int64 from MySQL, []byte from TiDB driver, etc.).
func coerceInt(v interface{}) int {
	switch x := v.(type) {
	case int:
		return x
	case int32:
		return int(x)
	case int64:
		return int(x)
	case float64:
		return int(x)
	case []byte:
		var n int
		for _, b := range x {
			if b < '0' || b > '9' {
				return n
			}
			n = n*10 + int(b-'0')
		}
		return n
	case string:
		var n int
		for _, b := range []byte(x) {
			if b < '0' || b > '9' {
				return n
			}
			n = n*10 + int(b-'0')
		}
		return n
	}
	return 0
}

func joinSelectExpression(qc QualifiedColumn, tmpl *JoinTemplate) string {
	if qc.SelectExpr != "" {
		return qc.SelectExpr
	}

	sqlAlias := resolveTableAlias(qc.TableAlias, tmpl)
	return fmt.Sprintf("`%s`.`%s`", sqlAlias, qc.Column)
}

// buildJoinWhere adds WHERE clauses for report filters using the JOIN template.
// All params are added directly to the builder via b.Where(). When the report
// definition carries a BooleanFilter containing an OR, filters are combined
// via that expression; otherwise each filter is ANDed into the builder.
func (s *ReportService) buildJoinWhere(b *query.Builder, def *models.ReportDefinition, tmpl *JoinTemplate) {
	filters := def.Filters
	if len(filters) == 0 {
		return
	}

	// Pre-compile every filter into a (clause, params) pair. We always need
	// this for the AND path too, so we only walk the list once.
	type compiled struct {
		clause string
		params []interface{}
		ok     bool
	}
	clauses := make([]compiled, len(filters))
	for i, f := range filters {
		clause, params, ok := s.compileJoinFilterClause(f, tmpl)
		clauses[i] = compiled{clause: clause, params: params, ok: ok}
	}

	boolExpr := strings.TrimSpace(def.BooleanFilter)
	hasOR := boolExpr != "" && containsWordOR(boolExpr)

	if hasOR {
		// Substitute numbered tokens with the compiled clauses. Walk from
		// highest to lowest so "12" isn't eaten by "1". Params are collected
		// in substitution order; since expr only contains operators,
		// parentheses and numbers, the resulting parameter order mirrors a
		// left-to-right walk of the tree after substitution.
		expr := boolExpr
		// Collect params in token order (left-to-right walk of the original
		// expression), not substitution order, so `?` placeholders line up.
		type slot struct {
			idx    int // 1-based
			pos    int
			length int
		}
		tokRe := regexp.MustCompile(`\b\d+\b`)
		matches := tokRe.FindAllStringIndex(expr, -1)
		var slots []slot
		for _, m := range matches {
			n, err := strconv.Atoi(expr[m[0]:m[1]])
			if err != nil || n < 1 || n > len(filters) {
				continue
			}
			slots = append(slots, slot{idx: n, pos: m[0], length: m[1] - m[0]})
		}
		// Abort compilation if any referenced filter failed to compile.
		for _, sl := range slots {
			if !clauses[sl.idx-1].ok {
				hasOR = false
				break
			}
		}
		if hasOR {
			// Build new expression + ordered params by scanning left-to-right.
			var out strings.Builder
			var orderedParams []interface{}
			last := 0
			for _, sl := range slots {
				out.WriteString(expr[last:sl.pos])
				out.WriteString("(")
				out.WriteString(clauses[sl.idx-1].clause)
				out.WriteString(")")
				orderedParams = append(orderedParams, clauses[sl.idx-1].params...)
				last = sl.pos + sl.length
			}
			out.WriteString(expr[last:])
			combined := replaceSQLBoolOperators(out.String())
			b.Where(combined, orderedParams...)
			return
		}
	}

	// Default AND path.
	for _, c := range clauses {
		if !c.ok {
			continue
		}
		b.Where(c.clause, c.params...)
	}
}

// compileJoinFilterClause resolves a single report filter into a SQL fragment
// and its bound parameters, using the JOIN template to pick the correct
// qualified column reference. Returns (_, _, false) when the filter should
// be skipped (unknown column, unresolvable).
func (s *ReportService) compileJoinFilterClause(f models.ReportFilter, tmpl *JoinTemplate) (string, []interface{}, bool) {
	var qualifiedCol string
	if qc, ok := tmpl.Columns[f.Column]; ok && qc.SelectExpr != "" {
		qualifiedCol = qc.SelectExpr
	} else {
		tblAlias, colName, found := resolveQualifiedColumn(f.Column, tmpl)
		if !found {
			return "", nil, false
		}
		sqlAlias := resolveTableAlias(tblAlias, tmpl)
		qualifiedCol = fmt.Sprintf("`%s`.`%s`", sqlAlias, colName)
	}

	op := "="
	switch f.Operator {
	case "equals", "eq":
		op = "="
	case "notEqual", "neq":
		op = "!="
	case "contains":
		op = "LIKE"
	case "notContain", "doesNotContain":
		op = "NOT LIKE"
	case "greaterThan", "gt":
		op = ">"
	case "greaterOrEqual", "gte":
		op = ">="
	case "lessThan", "lt":
		op = "<"
	case "lessOrEqual", "lte":
		op = "<="
	}

	val := f.Value
	if strings.EqualFold(val, "true") {
		val = "1"
	} else if strings.EqualFold(val, "false") {
		val = "0"
	}

	// Country-code → full-name mapping for CAMPAIGN_MEMBER COUNTRY_CODE.
	if strings.EqualFold(f.Column, "COUNTRY_CODE") && len(val) == 2 {
		if name := lookupCountryName(val); name != "" {
			val = name
		}
	}

	// Strip currency prefix for numeric comparisons. SF stores currency filter
	// values as "USD 500" (ISO code + space + amount). When the operator is a
	// numeric compare, MySQL coerces "USD 500" to 0, silently dropping the
	// filter. Extract the trailing numeric portion.
	if isNumericOp(op) {
		if n, ok := stripCurrencyPrefix(val); ok {
			val = n
		}
	}

	// Multi-value IN handling.
	if (op == "=" || op == "!=") && strings.Contains(val, ",") {
		parts := strings.Split(val, ",")
		placeholders := make([]string, len(parts))
		vals := make([]interface{}, len(parts))
		for j, p := range parts {
			placeholders[j] = "?"
			vals[j] = strings.TrimSpace(p)
		}
		inOp := "IN"
		if op == "!=" {
			inOp = "NOT IN"
		}
		return fmt.Sprintf("%s %s (%s)", qualifiedCol, inOp, strings.Join(placeholders, ",")), vals, true
	}

	if f.Operator == "contains" || f.Operator == "notContain" || f.Operator == "doesNotContain" {
		val = fmt.Sprintf("%%%s%%", val)
	}

	switch f.Operator {
	case "notContain", "doesNotContain", "notEqual", "neq":
		return fmt.Sprintf("(%s %s ? OR %s IS NULL)", qualifiedCol, op, qualifiedCol), []interface{}{val}, true
	default:
		return qualifiedCol + " " + op + " ?", []interface{}{val}, true
	}
}

// isNumericOp reports whether the resolved SQL operator is a numeric
// comparison that should not be string-compared when the RHS is a currency
// literal.
func isNumericOp(op string) bool {
	switch op {
	case ">", "<", ">=", "<=":
		return true
	}
	return false
}

// currencyPrefixRe matches leading ISO-4217-ish currency codes (3 uppercase
// letters) or common currency symbols optionally followed by a space, then a
// numeric value (with optional commas / decimal / sign). Returned numeric
// string is what MySQL can compare against a DECIMAL column.
var currencyPrefixRe = regexp.MustCompile(`^\s*(?:[A-Z]{3}\s*|[$€£¥₩₹])\s*([-+]?\d[\d,]*(?:\.\d+)?)\s*$`)

// stripCurrencyPrefix returns the numeric portion of a currency-formatted
// filter value (e.g. "USD 500" → "500", "$1,234.56" → "1234.56"). Returns
// (_, false) when the value doesn't match the pattern — the caller then
// uses the original string.
func stripCurrencyPrefix(s string) (string, bool) {
	m := currencyPrefixRe.FindStringSubmatch(s)
	if m == nil {
		return "", false
	}
	// Drop thousands separators so MySQL can parse the literal.
	return strings.ReplaceAll(m[1], ",", ""), true
}

// containsWordOR reports whether a BooleanFilter expression contains a
// top-level OR token (not inside a literal). A plain substring match would
// false-positive on words like "OPERATOR".
func containsWordOR(s string) bool {
	upper := strings.ToUpper(s)
	for i := 0; i <= len(upper)-2; i++ {
		if upper[i:i+2] != "OR" {
			continue
		}
		prevOk := i == 0 || !isIdentByte(upper[i-1])
		nextOk := i+2 == len(upper) || !isIdentByte(upper[i+2])
		if prevOk && nextOk {
			return true
		}
	}
	return false
}

// replaceSQLBoolOperators rewrites SF-style uppercase AND/OR tokens to SQL
// keywords (which are also AND/OR — so this is a no-op for uppercase but
// normalizes any mixed-case variants to canonical form). Quoted string
// literals are left alone.
func replaceSQLBoolOperators(s string) string {
	var b strings.Builder
	inSingle, inDouble := false, false
	i := 0
	for i < len(s) {
		ch := s[i]
		if ch == '\'' && !inDouble {
			inSingle = !inSingle
			b.WriteByte(ch)
			i++
			continue
		}
		if ch == '"' && !inSingle {
			inDouble = !inDouble
			b.WriteByte(ch)
			i++
			continue
		}
		if inSingle || inDouble {
			b.WriteByte(ch)
			i++
			continue
		}
		prevIsIdent := i > 0 && isIdentByte(s[i-1])
		if !prevIsIdent && i+2 <= len(s) && strings.EqualFold(s[i:i+2], "OR") && (i+2 == len(s) || !isIdentByte(s[i+2])) {
			b.WriteString("OR")
			i += 2
			continue
		}
		if !prevIsIdent && i+3 <= len(s) && strings.EqualFold(s[i:i+3], "AND") && (i+3 == len(s) || !isIdentByte(s[i+3])) {
			b.WriteString("AND")
			i += 3
			continue
		}
		b.WriteByte(ch)
		i++
	}
	return b.String()
}

// needsDistinctForAggregates returns true only when the report's aggregates
// would be inflated by JOIN fan-out. RowCount is a raw row count and must
// not be distinct'd (SF semantics). SUM/AVG/MIN/MAX of primary-entity fields
// would be inflated by child JOINs, so they warrant DISTINCT.
func needsDistinctForAggregates(aggs []models.ReportAggregate) bool {
	for _, a := range aggs {
		if strings.EqualFold(a.Type, "RowCount") || a.Type == "" && a.Field == "" {
			continue
		}
		return true
	}
	return false
}

// countryCodeToName is a minimal ISO-3166 alpha-2 → full-name mapping used
// to translate CAMPAIGN_MEMBER COUNTRY_CODE filters when the physical
// campaignmember.Country column holds the name rather than the code. Only
// the countries that actually appear in live SF reports need entries —
// growing the map on demand is preferable to a 250-row static table.
var countryCodeToName = map[string]string{
	"JP": "Japan",
	"US": "United States",
	"CN": "China",
	"KR": "South Korea",
	"TW": "Taiwan",
	"HK": "Hong Kong",
	"SG": "Singapore",
	"IN": "India",
	"ID": "Indonesia",
	"TH": "Thailand",
	"VN": "Vietnam",
	"MY": "Malaysia",
	"PH": "Philippines",
	"AU": "Australia",
	"NZ": "New Zealand",
	"GB": "United Kingdom",
	"DE": "Germany",
	"FR": "France",
	"ES": "Spain",
	"IT": "Italy",
	"CA": "Canada",
	"BR": "Brazil",
	"MX": "Mexico",
}

func lookupCountryName(code string) string {
	return countryCodeToName[strings.ToUpper(code)]
}

func tableNameForAlias(tmpl *JoinTemplate, alias string) string {
	if tmpl == nil {
		return ""
	}
	if strings.EqualFold(alias, tmpl.PrimaryAlias) {
		return tmpl.PrimaryTable
	}
	for _, join := range tmpl.Joins {
		if strings.EqualFold(alias, join.Alias) {
			return join.Table
		}
	}
	return ""
}

// backtickRef converts "alias.Column" to "`alias`.`Column`" for safe SQL references.
func backtickRef(ref string) string {
	if idx := strings.Index(ref, "."); idx >= 0 {
		return fmt.Sprintf("`%s`.`%s`", ref[:idx], ref[idx+1:])
	}
	return fmt.Sprintf("`%s`", ref)
}

// backtickJoinRef is like backtickRef but resolves the primary alias to the table name,
// because the query builder's FROM clause doesn't use an alias.
func backtickJoinRef(ref string, tmpl *JoinTemplate) string {
	if idx := strings.Index(ref, "."); idx >= 0 {
		alias := ref[:idx]
		col := ref[idx+1:]
		// If the alias matches the primary alias, use the physical table name instead.
		if alias == tmpl.PrimaryAlias {
			return fmt.Sprintf("`%s`.`%s`", tmpl.PrimaryTable, col)
		}
		return fmt.Sprintf("`%s`.`%s`", alias, col)
	}
	return fmt.Sprintf("`%s`", ref)
}

// resolveTableAlias returns the SQL-usable reference for a table alias:
// primary alias → physical table name (because FROM has no alias), others → as-is.
func resolveTableAlias(alias string, tmpl *JoinTemplate) string {
	if alias == tmpl.PrimaryAlias {
		return tmpl.PrimaryTable
	}
	return alias
}

// sfColumnDisplayName converts an SF column reference to a display-friendly name.
// "CAMPAIGN_NAME" → "Name" (uses resolveSFColumn fallback)
// "Lead.Region__c" → "Region__c"
// "CampaignMember.Lead_Status__c" → "Lead_Status__c"
func sfColumnDisplayName(sfCol string) string {
	if idx := strings.Index(sfCol, "."); idx >= 0 {
		return sfCol[idx+1:]
	}
	return resolveSFColumn(sfCol)
}

func joinResultAlias(sfCol string, qc QualifiedColumn) string {
	if qc.ResultAlias != "" {
		return qc.ResultAlias
	}

	base := strings.TrimSuffix(sfCol, ".CONVERT")
	if idx := strings.LastIndex(base, "."); idx >= 0 {
		return base[idx+1:]
	}

	if pascal := utils.SnakeToPascal(base); pascal != base {
		return pascal
	}

	if mapped := resolveSFColumn(base); mapped != base {
		return mapped
	}

	return base
}