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
}

// polyJoinAliases match the convention the column-path resolver's marker
// prefixes assume.
const (
	polyLeadAlias    = "l"
	polyContactAlias = "ct"
	polyLeadTable    = "lead"
	polyContactTable = "contact"
)

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

// setPolyWarnSinkForTest installs a test-only sink and returns a restore
// function.
func setPolyWarnSinkForTest(fn func(format string, args ...interface{})) func() {
	polyWarnSinkMu.Lock()
	prev := polyWarnSink
	polyWarnSink = fn
	polyWarnSinkMu.Unlock()
	return func() {
		polyWarnSinkMu.Lock()
		polyWarnSink = prev
		polyWarnSinkMu.Unlock()
	}
}

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

// ExpandPolymorphicMarkers inspects resolved for any __LEAD_JOIN__: /
// __CONTACT_JOIN__: markers the column-path resolver emitted. When at least
// one marker is present it installs the corresponding LEFT JOIN onto tmpl
// (at most once per side) and rewrites the offending selectExpr values to
// the real `alias`.`col` form. When markers are present but the template
// lacks a campaignmember join step the markers are left as-is and a WARN is
// logged — the caller will emit NULL for those columns via the usual
// ok==false path.
//
// The function mutates tmpl (appending joins) and returns a fresh slice of
// resolvedReportColumn with the rewritten selectExpr values. Inputs that do
// not carry any marker are passed through untouched.
func ExpandPolymorphicMarkers(tmpl *JoinTemplate, resolved []resolvedReportColumn) []resolvedReportColumn {
	if len(resolved) == 0 {
		return resolved
	}

	needsLead, needsContact := false, false
	for _, rc := range resolved {
		if !rc.ok {
			continue
		}
		if strings.HasPrefix(rc.selectExpr, leadJoinMarker) {
			needsLead = true
		}
		if strings.HasPrefix(rc.selectExpr, contactJoinMarker) {
			needsContact = true
		}
	}

	if !needsLead && !needsContact {
		return resolved
	}

	cmAlias, hasCM := findTableAlias(tmpl, "campaignmember")
	if !hasCM {
		emitPolyWarn("template has no campaignmember join, leaving %d markers unresolved", countMarkers(resolved))
		return resolved
	}

	// Install joins only once, only for the sides we actually need, and
	// only when they are not already present. The execution layer tolerates
	// repeated joins badly (ambiguous alias), so this must be idempotent.
	if needsLead && !hasJoinAlias(tmpl, polyLeadAlias) {
		tmpl.Joins = append(tmpl.Joins, JoinStep{
			Table:    polyLeadTable,
			Alias:    polyLeadAlias,
			JoinType: "LEFT JOIN",
			OnLeft:   fmt.Sprintf("%s.Id", polyLeadAlias),
			OnRight:  fmt.Sprintf("%s.LeadOrContactId", cmAlias),
		})
	}
	if needsContact && !hasJoinAlias(tmpl, polyContactAlias) {
		tmpl.Joins = append(tmpl.Joins, JoinStep{
			Table:    polyContactTable,
			Alias:    polyContactAlias,
			JoinType: "LEFT JOIN",
			OnLeft:   fmt.Sprintf("%s.Id", polyContactAlias),
			OnRight:  fmt.Sprintf("%s.LeadOrContactId", cmAlias),
		})
	}

	out := make([]resolvedReportColumn, len(resolved))
	for i, rc := range resolved {
		out[i] = rc
		if !rc.ok {
			continue
		}
		switch {
		case strings.HasPrefix(rc.selectExpr, leadJoinMarker):
			field := strings.TrimPrefix(rc.selectExpr, leadJoinMarker)
			// CRITICAL-1 defense-in-depth: the column-path resolver already
			// validates the payload before emitting a marker, but a future
			// regression (new code path that constructs a marker directly)
			// must not bypass SQL identifier validation. Refuse to expand
			// unsafe markers — the resolved column falls back to NULL.
			if !isSafeSQLIdent(field) {
				emitPolyWarn("rejecting unsafe lead marker payload, col=%q", rc.col)
				out[i].ok = false
				out[i].selectExpr = ""
				continue
			}
			out[i].selectExpr = fmt.Sprintf("`%s`.`%s`", polyLeadAlias, field)
		case strings.HasPrefix(rc.selectExpr, contactJoinMarker):
			field := strings.TrimPrefix(rc.selectExpr, contactJoinMarker)
			if !isSafeSQLIdent(field) {
				emitPolyWarn("rejecting unsafe contact marker payload, col=%q", rc.col)
				out[i].ok = false
				out[i].selectExpr = ""
				continue
			}
			out[i].selectExpr = fmt.Sprintf("`%s`.`%s`", polyContactAlias, field)
		}
	}
	return out
}

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

func countMarkers(resolved []resolvedReportColumn) int {
	n := 0
	for _, rc := range resolved {
		if !rc.ok {
			continue
		}
		if strings.HasPrefix(rc.selectExpr, leadJoinMarker) || strings.HasPrefix(rc.selectExpr, contactJoinMarker) {
			n++
		}
	}
	return n
}

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