package analytics

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
)

// resolvedReportColumn is the canonical intermediate representation used
// between ResolveReportColumn (the pure column-path resolver) and the SQL
// projection loop in executeJoinReport. Wave 3 wiring builds one of these
// per report column, optionally expands polymorphic markers, and feeds the
// result back into the execution path.
type resolvedReportColumn struct {
	col         string // original Report.Columns entry
	selectExpr  string // SQL expression ready for AddSelectRaw
	resultAlias string // row-key alias the frontend will read
	ok          bool   // false → emit NULL placeholder
}

// polyWarnSink is the WARN-level log function for diagnostics; tests swap
// it through setPolyWarnSinkForTest (HIGH-7 race fix). Production goes
// through the stdlib logger for parity with the existing column-path
// resolver. The mutex protects concurrent reads/writes against a test
// swap happening in parallel to a production resolve.
var (
	polyWarnSinkMu sync.RWMutex
	polyWarnSink   = defaultPolyWarnSink
)

func defaultPolyWarnSink(format string, args ...interface{}) {
	log.Printf("crt_polymorphic_joins: "+format, args...)
}

func emitPolyWarn(format string, args ...interface{}) {
	polyWarnSinkMu.RLock()
	sink := polyWarnSink
	polyWarnSinkMu.RUnlock()
	if sink != nil {
		sink(format, args...)
	}
}

// ExpandPolymorphicMarkers is retained as a no-op compatibility stage. PR4
// removed CRM-specific Lead/Contact marker emission from the column resolver;
// polymorphic report behavior must be represented through generic CRT metadata
// rather than object-name branches.
func ExpandPolymorphicMarkers(tmpl *JoinTemplate, resolved []resolvedReportColumn) []resolvedReportColumn {
	return resolved
}

// ExpandLookupJoinMarkers installs runtime LEFT JOINs for every
// __LOOKUP_JOIN__ marker emitted by resolveThreeSegment and rewrites those
// markers to real `alias`.`field` references. The marker payload shape is
// documented at the lookupJoinMarker constant.
//
// Aliases are minted deterministically from the targetTable name so
// repeated references to the same FK deduplicate to one join. The caller
// holds the mutated template; safe to call multiple times.
func ExpandLookupJoinMarkers(tmpl *JoinTemplate, resolved []resolvedReportColumn) []resolvedReportColumn {
	if tmpl == nil || len(resolved) == 0 {
		return resolved
	}
	// aliasByKey deduplicates by (originAlias, fkCol, targetTable) — two
	// markers pointing at the same parent row join exactly once.
	aliasByKey := map[string]string{}

	out := make([]resolvedReportColumn, len(resolved))
	for i, rc := range resolved {
		out[i] = rc
		if !rc.ok || !strings.HasPrefix(rc.selectExpr, lookupJoinMarker) {
			continue
		}
		payload := strings.TrimPrefix(rc.selectExpr, lookupJoinMarker)
		parts := strings.Split(payload, "|")
		if len(parts) != 4 {
			emitPolyWarn("malformed lookup marker, col=%q payload=%q", rc.col, payload)
			out[i].ok = false
			out[i].selectExpr = ""
			continue
		}
		originAlias, fkCol, targetTable, targetField := parts[0], parts[1], parts[2], parts[3]
		// Defense-in-depth: every payload segment must be a clean SQL ident
		// before we interpolate it. The resolver already validated these,
		// but a future regression in marker construction must not get a
		// free pass to the query builder.
		if !isSafeSQLIdent(originAlias) || !isSafeSQLIdent(fkCol) || !isSafeSQLIdent(targetTable) || !isSafeSQLIdent(targetField) {
			emitPolyWarn("rejecting unsafe lookup marker, col=%q", rc.col)
			out[i].ok = false
			out[i].selectExpr = ""
			continue
		}
		key := originAlias + "|" + fkCol + "|" + targetTable
		alias, seen := aliasByKey[key]
		if !seen {
			alias = mintLookupAlias(tmpl, targetTable)
			aliasByKey[key] = alias
			tmpl.Joins = append(tmpl.Joins, JoinStep{
				Table:    targetTable,
				Alias:    alias,
				JoinType: "LEFT JOIN",
				OnLeft:   fmt.Sprintf("%s.Id", alias),
				OnRight:  fmt.Sprintf("%s.%s", originAlias, fkCol),
			})
		}
		out[i].selectExpr = fmt.Sprintf("`%s`.`%s`", alias, targetField)
	}
	return out
}

// mintLookupAlias returns a short alias that doesn't collide with any
// existing alias on tmpl. Uses "lj<n>" so the result is stable across runs
// with the same join order.
func mintLookupAlias(tmpl *JoinTemplate, targetTable string) string {
	n := 1
	for {
		candidate := fmt.Sprintf("lj%d", n)
		if !hasJoinAlias(tmpl, candidate) && candidate != tmpl.PrimaryAlias {
			return candidate
		}
		n++
	}
}

func hasJoinAlias(tmpl *JoinTemplate, alias string) bool {
	if tmpl == nil {
		return false
	}
	for _, j := range tmpl.Joins {
		if j.Alias == alias {
			return true
		}
	}
	return false
}

// decorateTemplateForCRT runs each report column AND filter column through
// ResolveReportColumn, expands polymorphic markers, and installs the resolved
// expressions into tmpl.Columns so both executeJoinReport's projection loop
// and buildJoinWhere's filter loop find them by their original key.
//
// HIGH-6: this function has a documented side effect on tmpl — it mutates
// tmpl.Joins (polymorphic join installation) and tmpl.Columns (one entry
// per successfully resolved report column, keyed by the original
// Report.Columns entry). executeJoinReport relies on that side effect;
// do not call this from a code path that expects an immutable tmpl.
// The single return value is the same tmpl pointer for call-site
// readability.
func decorateTemplateForCRT(ctx context.Context, tmpl *JoinTemplate, columns []string, filterCols []string, md crtMetadataLookup) *JoinTemplate {
	if tmpl == nil {
		return nil
	}
	// Merge projection columns + filter columns into one batch so the
	// polymorphic join expansion pass walks them together and we install
	// each lead/contact join at most once across the whole report.
	all := make([]resolvedReportColumn, 0, len(columns)+len(filterCols))
	for _, col := range columns {
		expr, alias, ok := resolveReportColumnImpl(ctx, col, tmpl, md)
		all = append(all, resolvedReportColumn{col: col, selectExpr: expr, resultAlias: alias, ok: ok})
	}
	for _, col := range filterCols {
		if col == "" {
			continue
		}
		// Skip filter columns already resolved as projection columns to
		// avoid duplicate work and redundant map entries.
		if _, dup := tmplHas(all, col); dup {
			continue
		}
		expr, alias, ok := resolveReportColumnImpl(ctx, col, tmpl, md)
		all = append(all, resolvedReportColumn{col: col, selectExpr: expr, resultAlias: alias, ok: ok})
	}
	all = ExpandPolymorphicMarkers(tmpl, all)
	all = ExpandLookupJoinMarkers(tmpl, all)

	if tmpl.Columns == nil {
		tmpl.Columns = make(map[string]QualifiedColumn)
	}
	for _, rc := range all {
		if !rc.ok {
			continue
		}
		tmpl.Columns[rc.col] = QualifiedColumn{
			SelectExpr:  rc.selectExpr,
			ResultAlias: rc.resultAlias,
		}
	}
	return tmpl
}

func tmplHas(resolved []resolvedReportColumn, col string) (int, bool) {
	for i, rc := range resolved {
		if rc.col == col {
			return i, true
		}
	}
	return 0, false
}