package analytics

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

	"github.com/nexuscrm/backend/internal/features/metadata"
	"github.com/nexuscrm/backend/internal/infrastructure/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

// ResolveReportColumn turns a single entry from Report.Columns into
// (selectExpr, resultAlias, ok). ok==false means the column could not be
// resolved — the caller should emit NULL in its place. This function never
// panics and is pure with respect to its inputs (aside from DEBUG logging
// of unknown columns).
//
// Inputs may be nil:
//   - tmpl == nil: only the legacy alias map is consulted. Any dot-notation,
//     template-qualified, or polymorphic path will fail-soft with ok=false.
//   - md   == nil: FK traversal (three-segment cross-join) and polymorphic
//     bare-custom-field lookups are skipped; explicit Lead./Contact. paths
//     still resolve via the __LEAD_JOIN__ / __CONTACT_JOIN__ marker contract
//     (see below).
//
// Polymorphic marker contract with the wiring agent:
//
// When a polymorphic path resolves to a field that lives on the Lead or
// Contact side of CampaignMember.LeadOrContactId, this resolver cannot add
// a join to the template (the function is pure). Instead, it emits a
// synthesized select expression beginning with one of two marker prefixes:
//
//	"__LEAD_JOIN__:<field>"      e.g. "__LEAD_JOIN__:Company"
//	"__CONTACT_JOIN__:<field>"   e.g. "__CONTACT_JOIN__:AccountId"
//
// The wiring layer in report_service.go (added by a later PR) is expected
// to detect these prefixes, install the corresponding LEFT JOIN onto the
// JoinTemplate at execution time, and rewrite the select expression to the
// real qualified column. These marker strings are NOT dead code — they are
// the interface boundary between this pure resolver and the stateful
// wiring layer.
//
// SECURITY: every string that becomes part of a SQL identifier is validated
// against isSafeSQLIdent before it is emitted into a selectExpr or marker
// payload. Callers that save CRT Sections or Report.Columns are NOT trusted
// — an insider that can save a Report must not be able to inject SQL via
// the column path. Any failure path here returns ok=false (fail-soft).
func ResolveReportColumn(ctx context.Context, col string, tmpl *JoinTemplate, md *metadata.MetadataService) (selectExpr string, resultAlias string, ok bool) {
	return resolveReportColumnImpl(ctx, col, tmpl, newCRTMetadataAdapter(md))
}

func resolveReportColumnImpl(ctx context.Context, col string, tmpl *JoinTemplate, md crtMetadataLookup) (selectExpr string, resultAlias string, ok bool) {
	col = strings.TrimSpace(col)
	if col == "" {
		return "", "", false
	}

	// 1. Polymorphic CampaignMember_ContactOrLead.* — handled first because
	//    its prefix otherwise looks like a two-segment dot-notation path and
	//    a downstream .Lead / .Contact segment would be misclassified.
	if strings.HasPrefix(col, polyPrefix) {
		return resolveLeadOrContactPath(ctx, col, tmpl, md)
	}

	// 2. Flat legacy alias (no dot). This also covers SCREAMING_SNAKE.
	if !strings.Contains(col, ".") {
		if mapped, found := legacyAliasMap[col]; found {
			return buildSelectFromPath(mapped, tmpl, col)
		}
		// Fallthrough: treat as a bare field name on the template primary
		// table. The column name is user-controlled (Report.Columns is saved
		// by any user who can save a Report), so it MUST be validated before
		// being interpolated into a backticked identifier — otherwise an
		// embedded backtick defeats the quoting.
		if !isSafeSQLIdent(col) {
			logUnknownColumn(col, "flat-unsafe-ident")
			return "", "", false
		}
		if tmpl != nil {
			if qc, found := tmpl.Columns[col]; found {
				return qualifiedColumnToSelect(qc), deriveAlias(col), true
			}
			return fmt.Sprintf("`%s`.`%s`", tmpl.PrimaryAlias, col), deriveAlias(col), true
		}
		logUnknownColumn(col, "flat-no-template")
		return "", "", false
	}

	// 3. Dot-notation path.
	segments := strings.Split(col, ".")
	// Guard against things like "." or ".Foo" — empty segments.
	for _, s := range segments {
		if s == "" {
			logUnknownColumn(col, "empty-segment")
			return "", "", false
		}
	}

	if tmpl == nil {
		logUnknownColumn(col, "dot-no-template")
		return "", "", false
	}

	// Direct dot-key lookup in template columns (e.g. "Campaign.Name" keyed as-is).
	// The lookup value is a QualifiedColumn built by the synthesizer using
	// already-validated inputs; no extra check needed here.
	if qc, found := tmpl.Columns[col]; found {
		return qualifiedColumnToSelect(qc), deriveAlias(col), true
	}

	switch len(segments) {
	case 2:
		return resolveTwoSegment(segments[0], segments[1], tmpl, col)
	case 3:
		return resolveThreeSegment(ctx, segments[0], segments[1], segments[2], tmpl, md, col)
	default:
		// 4+ segment paths are out of scope per §2 non-goal #3.
		logUnknownColumn(col, "too-many-segments")
		return "", "", false
	}
}

// polyPrefix is the exact polymorphic path prefix observed in the research
// inventory — it is a Report.Columns convention, not a CRT-metadata construct.
const polyPrefix = "CampaignMember_ContactOrLead."

// Marker prefixes for the contract described in the ResolveReportColumn doc.
const (
	leadJoinMarker    = "__LEAD_JOIN__:"
	contactJoinMarker = "__CONTACT_JOIN__:"
	// lookupJoinMarker is emitted by resolveThreeSegment when a 3-segment
	// path resolves to a parent/lookup traversal that needs a runtime
	// LEFT JOIN. Payload shape:
	//   __LOOKUP_JOIN__:<originAlias>|<fkCol>|<targetTable>|<targetField>
	// ExpandLookupJoinMarkers (in crt_polymorphic_joins.go) installs the
	// join with a deterministic alias and rewrites the marker to the real
	// `<alias>.<field>` form.
	lookupJoinMarker = "__LOOKUP_JOIN__:"
)

// legacyAliasMap seeds the flat-alias translation table from the research
// inventory's top flat codes. Entries here are the only ones verified against
// the real Sections[] / Report.Columns data; see §10 decision #2 — grow the
// seed only when a new unknown shows up in the debug log.
var legacyAliasMap = map[string]string{
	"CUST_NAME":  "Account.Name",
	"SUBJECT":    "Subject",
	"STATUS":     "Status",
	"CLOSE_DATE": "CloseDate",
	"FULL_NAME":  "Name",
}

// cmDenormalizedFields are the CampaignMember columns that mirror the resolved
// Lead/Contact side directly. The research data confirms these five plus
// CompanyOrAccount are present on campaignmember; we include Title for
// completeness (it is denormalized by SF on CampaignMember too).
var cmDenormalizedFields = map[string]struct{}{
	"FirstName":        {},
	"LastName":         {},
	"Email":            {},
	"Phone":            {},
	"Title":            {},
	"CompanyOrAccount": {},
}

// resolveLeadOrContactPath implements the §4.8 algorithm for the
// "CampaignMember_ContactOrLead." prefix. Every path that emits a marker or
// qualified column validates its SQL identifier against isSafeSQLIdent —
// this is the CRITICAL-1 fix for SQL injection via the marker payload.
func resolveLeadOrContactPath(ctx context.Context, col string, tmpl *JoinTemplate, md crtMetadataLookup) (string, string, bool) {
	suffix := strings.TrimPrefix(col, polyPrefix)
	if suffix == "" {
		logUnknownColumn(col, "poly-empty-suffix")
		return "", "", false
	}

	// The template must carry a campaignmember join step; if it does not,
	// there is no `cm` alias to project onto and the whole path is moot.
	cmAlias, hasCM := findTableAlias(tmpl, "campaignmember")
	if !hasCM {
		logUnknownColumn(col, "poly-no-campaignmember-join")
		return "", "", false
	}

	// Case 1: denormalized field on CampaignMember itself.
	if _, denorm := cmDenormalizedFields[suffix]; denorm {
		// suffix is a key in a constant map, but validate defensively so
		// future additions to cmDenormalizedFields cannot slip unsafe
		// identifiers through.
		if !isSafeSQLIdent(suffix) {
			logUnknownColumn(col, "poly-denorm-unsafe-ident")
			return "", "", false
		}
		return fmt.Sprintf("`%s`.`%s`", cmAlias, suffix), deriveAlias(col), true
	}

	// Case 2: explicit Lead./Contact. branch.
	if strings.HasPrefix(suffix, "Lead.") {
		field := strings.TrimPrefix(suffix, "Lead.")
		if !isSafeSQLIdent(field) {
			logUnknownColumn(col, "poly-lead-unsafe-ident")
			return "", "", false
		}
		return leadJoinMarker + field, deriveAlias(col), true
	}
	if strings.HasPrefix(suffix, "Contact.") {
		field := strings.TrimPrefix(suffix, "Contact.")
		if !isSafeSQLIdent(field) {
			logUnknownColumn(col, "poly-contact-unsafe-ident")
			return "", "", false
		}
		return contactJoinMarker + field, deriveAlias(col), true
	}

	// Case 3: bare field — heuristically probe both sides via metadata.
	if md == nil {
		logUnknownColumn(col, "poly-bare-no-metadata")
		return "", "", false
	}
	if !isSafeSQLIdent(suffix) {
		logUnknownColumn(col, "poly-bare-unsafe-ident")
		return "", "", false
	}
	onLead := fieldExistsOnObject(ctx, md, "lead", suffix)
	onContact := fieldExistsOnObject(ctx, md, "contact", suffix)
	switch {
	case onLead && !onContact:
		return leadJoinMarker + suffix, deriveAlias(col), true
	case onContact && !onLead:
		return contactJoinMarker + suffix, deriveAlias(col), true
	default:
		// Both or neither — fail-soft per §10 decision #3.
		logUnknownColumn(col, "poly-bare-ambiguous")
		return "", "", false
	}
}

func resolveTwoSegment(prefix, field string, tmpl *JoinTemplate, original string) (string, string, bool) {
	alias, ok := findTableAlias(tmpl, prefix)
	if !ok {
		logUnknownColumn(original, "unknown-table-prefix")
		return "", "", false
	}
	if !isSafeSQLIdent(field) {
		logUnknownColumn(original, "two-seg-unsafe-field")
		return "", "", false
	}
	return fmt.Sprintf("`%s`.`%s`", alias, field), deriveAlias(original), true
}

func resolveThreeSegment(ctx context.Context, obj, rel, field string, tmpl *JoinTemplate, md crtMetadataLookup, original string) (string, string, bool) {
	alias, ok := findTableAlias(tmpl, obj)
	if !ok {
		logUnknownColumn(original, "three-seg-unknown-root")
		return "", "", false
	}
	if md == nil {
		logUnknownColumn(original, "three-seg-no-metadata")
		return "", "", false
	}
	// Walk the FK: look up `rel` on obj's schema as a field name first, then
	// as a relationship name (e.g. "Parent" → ParentId with
	// RelationshipName=="Parent"). Confirm it resolves to a reference and the
	// target schema actually has `field`. Fail-soft on any mismatch.
	schema := md.GetSchema(ctx, obj)
	if schema == nil {
		logUnknownColumn(original, "three-seg-root-schema-missing")
		return "", "", false
	}
	fkField := persistence.FindField(schema, rel)
	if fkField == nil || len(fkField.ReferenceTo) == 0 {
		fkField = findLookupByRelationshipName(schema, rel)
	}
	if fkField == nil || len(fkField.ReferenceTo) == 0 {
		logUnknownColumn(original, "three-seg-not-a-reference")
		return "", "", false
	}
	targetObj := fkField.ReferenceTo[0]
	targetSchema := md.GetSchema(ctx, targetObj)
	if targetSchema == nil {
		logUnknownColumn(original, "three-seg-target-schema-missing")
		return "", "", false
	}
	if persistence.FindField(targetSchema, field) == nil {
		logUnknownColumn(original, "three-seg-target-field-missing")
		return "", "", false
	}
	// Validate every string that will reach SQL interpolation. All three come
	// from metadata but defense-in-depth is cheap.
	if !isSafeSQLIdent(fkField.APIName) || !isSafeSQLIdent(field) || !isSafeSQLIdent(targetObj) {
		logUnknownColumn(original, "three-seg-unsafe-ident")
		return "", "", false
	}
	// Emit a __LOOKUP_JOIN__ marker so ExpandLookupJoinMarkers (called from
	// decorate) can install the actual join and rewrite the expression to a
	// real qualified column. Marker payload:
	//   originAlias : alias of the root object in the current template
	//   fkCol       : FK column name on originAlias
	//   targetTable : physical table to join
	//   field       : column on the target table
	marker := fmt.Sprintf("%s%s|%s|%s|%s", lookupJoinMarker, alias, fkField.APIName, strings.ToLower(targetObj), field)
	return marker, deriveAlias(original), true
}

// findLookupByRelationshipName returns the first field on schema whose
// RelationshipName matches relName (case-insensitive). This lets dot-notation
// paths like "Campaign.Parent.Name" locate ParentId via SF relationship-name
// conventions rather than raw field names.
func findLookupByRelationshipName(schema *models.ObjectMetadata, relName string) *models.FieldMetadata {
	if schema == nil || relName == "" {
		return nil
	}
	for i := range schema.Fields {
		f := &schema.Fields[i]
		if f.RelationshipName == nil || *f.RelationshipName == "" {
			continue
		}
		if strings.EqualFold(*f.RelationshipName, relName) && len(f.ReferenceTo) > 0 {
			return f
		}
	}
	return nil
}

// buildSelectFromPath turns a legacy-alias-mapped path like "Account.Name" into
// the correct qualified select expression relative to tmpl. Falls back to a
// bare column on the primary alias when tmpl is nil or the prefix is unknown.
func buildSelectFromPath(mapped string, tmpl *JoinTemplate, original string) (string, string, bool) {
	if strings.Contains(mapped, ".") {
		parts := strings.SplitN(mapped, ".", 2)
		if tmpl != nil {
			if alias, ok := findTableAlias(tmpl, parts[0]); ok {
				return fmt.Sprintf("`%s`.`%s`", alias, parts[1]), deriveAlias(original), true
			}
		}
		// Legacy alias maps to a cross-object path but the template does not
		// carry that table — project onto the primary alias using only the
		// leaf column. This matches the old sfColumnToField fallback.
		if tmpl != nil {
			return fmt.Sprintf("`%s`.`%s`", tmpl.PrimaryAlias, parts[1]), deriveAlias(original), true
		}
		return "", "", false
	}
	if tmpl != nil {
		return fmt.Sprintf("`%s`.`%s`", tmpl.PrimaryAlias, mapped), deriveAlias(original), true
	}
	// Without a template we still return the mapped column as a bare name
	// so callers operating on a single-table shape can still use it.
	return fmt.Sprintf("`%s`", mapped), deriveAlias(original), true
}

func findTableAlias(tmpl *JoinTemplate, name string) (string, bool) {
	if tmpl == nil {
		return "", false
	}
	if strings.EqualFold(name, tmpl.PrimaryTable) || strings.EqualFold(name, tmpl.PrimaryAlias) {
		return tmpl.PrimaryAlias, true
	}
	for _, j := range tmpl.Joins {
		if strings.EqualFold(name, j.Table) || strings.EqualFold(name, j.Alias) {
			return j.Alias, true
		}
	}
	return "", false
}

func fieldExistsOnObject(ctx context.Context, md crtMetadataLookup, obj, field string) bool {
	if md == nil {
		return false
	}
	schema := md.GetSchema(ctx, obj)
	return persistence.FindField(schema, field) != nil
}

func qualifiedColumnToSelect(qc QualifiedColumn) string {
	if qc.SelectExpr != "" {
		return qc.SelectExpr
	}
	return fmt.Sprintf("`%s`.`%s`", qc.TableAlias, qc.Column)
}

// deriveAlias produces a stable result-alias string for an incoming column
// path. The goal is to round-trip the original path through the response so
// the frontend can key back into it.
func deriveAlias(col string) string {
	return strings.ReplaceAll(col, ".", "_")
}

// unknownColumnSink is the DEBUG-channel fallback for §10 decision #2
// ("minimal seed + log unknowns"). It is a package-level function that
// tests may swap through setUnknownColumnSinkForTest (HIGH-7 race fix);
// production callers always go through the default stdlib-log sink. The
// mutex protects concurrent reads/writes so a property test swapping the
// sink cannot race a production resolve on another goroutine.
var (
	unknownColumnSinkMu sync.RWMutex
	unknownColumnSink   = defaultUnknownColumnSink
)

func defaultUnknownColumnSink(col, reason string) {
	// Keep the log one-line + trimmed — callers run this per report column
	// and we do not want to flood stderr in hot paths.
	if len(col) > 120 {
		col = col[:120] + "…"
	}
	log.Printf("crt_column_path: unresolved col=%q reason=%s", col, reason)
}

// setUnknownColumnSinkForTest installs a test-only sink and returns a
// restore function. Call via `defer restore()`. This is the only supported
// way to swap the sink.
func setUnknownColumnSinkForTest(fn func(col, reason string)) func() {
	unknownColumnSinkMu.Lock()
	prev := unknownColumnSink
	unknownColumnSink = fn
	unknownColumnSinkMu.Unlock()
	return func() {
		unknownColumnSinkMu.Lock()
		unknownColumnSink = prev
		unknownColumnSinkMu.Unlock()
	}
}

func logUnknownColumn(col, reason string) {
	unknownColumnSinkMu.RLock()
	sink := unknownColumnSink
	unknownColumnSinkMu.RUnlock()
	if sink != nil {
		sink(col, reason)
	}
}