package analytics

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

	pkgErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"golang.org/x/text/language"
	"golang.org/x/text/language/display"
)

var englishRegionNames = display.English.Regions()

// 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 the neutral date filter after the dialect has mapped any product
	// date-filter shape onto concrete bounds.
	if df := def.EffectiveDateFilter(); df != nil && df.Column != "" {
		tblAlias, colName, found := resolveQualifiedColumn(df.Column, tmpl)
		if found {
			sqlAlias := resolveTableAlias(tblAlias, tmpl)
			qualifiedCol := fmt.Sprintf("`%s`.`%s`", sqlAlias, colName)
			startDate, endDate := s.normalizeDateBoundsForField(ctx, tableNameForAlias(tmpl, tblAlias), colName, df.StartDate, df.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.DisplayMode = reportDisplayModeForExecution(resultDef.EffectiveDisplayMode(), validGrouping)
	resultDef.NormalizeBoundaryAliases()
	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
// installed report dialect returns a combined SQL expression, use it; 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}
	}

	fragments := make([]contract.ReportFilterExpressionFragment, 0, len(clauses))
	for i, clause := range clauses {
		if !clause.ok {
			continue
		}
		fragments = append(fragments, contract.ReportFilterExpressionFragment{
			Index:      i + 1,
			Expression: clause.clause,
			Params:     clause.params,
		})
	}
	if expr, ok := s.reportDefinitionDialect().BuildFilterExpression(contract.ReportFilterExpressionRequest{
		Language:      contract.ReportFilterExpressionLanguageSQL,
		FilterLogic:   def.EffectiveFilterLogic(),
		BooleanFilter: def.EffectiveFilterLogic(),
		Filters:       filters,
		Fragments:     fragments,
	}); ok {
		b.Where(expr.Expression, expr.Params...)
		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
}

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

func lookupCountryName(code string) string {
	normalized := strings.ToUpper(strings.TrimSpace(code))
	if len(normalized) != 2 {
		return ""
	}
	region, err := language.ParseRegion(normalized)
	if err != nil || !region.IsCountry() {
		return ""
	}
	return englishRegionNames.Name(region)
}

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

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

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
	}

	return base
}