package ai

// G1 market and relationship graph. This file adds the federal capture intelligence traversals as FIXED,
// named, parameterized QueryIntents over the GLOBAL public market tables: usaspending_awards (the federal
// award cache), opportunity_records (the unified opportunity catalog), and sam_opportunities (the SAM
// solicitation feed with its dotted agency hierarchy). Where the G3 skill graph walks a curated adjacency
// recursively, the G1 market graph is mostly AGGREGATE reads: the competitor field, the incumbent of
// record, the recompete pipeline, and the agency buying pattern are all GROUP BY rollups, not recursion,
// because the data already carries the keys the question needs (the agency hierarchy is a dotted
// full_parent_path string, so a string split and a GROUP BY answers the buyer question without a recursive
// CTE, exactly as the determinism mandate prefers).
//
// The graph NEVER computes a match score, a band, or any employment decision number. A dollar share, a
// competitor rank, an estimated value band are MARKET aggregates, structure the capture team reads, and
// they are explicitly NOT match scores; the frozen engine in internal/modules/matching/engine stays the
// only scorer and is untouched. The model never authors this SQL: every statement is a static, reviewed
// constant, registered in the closed Registry, asserted by ValidateRegistry at boot, and re guarded by
// GuardReadOnlyQuery on every execution.
//
// Honesty mandates from the GraphRAG review, encoded in the SQL by construction:
//   - usaspending_awards carries NO psc, NO set_aside, and NO pop_state column, so the awards self join and
//     the competitor field key ONLY on (awarding_agency, naics_code). The finer (agency, naics, psc,
//     set_aside, pop_state) slices come from opportunity_records, which does carry that full tuple, and
//     those edges are labeled opportunity_records sourced so a reader never mistakes an award for a slice it
//     cannot support.
//   - every incumbent edge is labeled prime-of-record (a constant label column), never the staffing
//     subcontractor, because the sub award labor reality is absent from the public award data; the prime
//     recipient is the only honest incumbent the data supports.
//   - the agency hierarchy rollup splits the sam_opportunities full_parent_path dotted string and groups, it
//     does NOT use a recursive CTE, because the hierarchy is already a materialized path and recursion would
//     add nothing but a determinism risk.
//
// Decimal, not float. Every dollar figure (award_amount, estimated_value) and every share is Postgres
// numeric end to end and is scanned into shopspring/decimal in Go, NEVER float64 and NEVER cast through
// ::float8: a surfaced market figure is a decision adjacent number and the platform's decimal rule applies.
// The recency weight is a numeric step function (more recent award years weigh more), kept inside the
// numeric type so the weighted share stays exact.

import (
	"context"
	"errors"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/shopspring/decimal"

	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

// The registered names of the four G1 market intents. Each is a FIXED entry in the closed Registry in
// tag.go, so ValidateRegistry asserts it at boot and ExecuteIntent runs it through the same guard, tenant
// transaction, and audit posture as every other TAG intent.
const (
	IntentSegmentCompetitorField = "segment_competitor_field"
	IntentIncumbentOfRecord      = "incumbent_of_record"
	IntentRecompetePipeline      = "recompete_pipeline"
	IntentAgencyBuyingPattern    = "agency_buying_pattern"
)

// primeOfRecordLabel is the constant incumbent edge label. Every incumbent the G1 graph surfaces is the
// PRIME recipient of record, never the staffing subcontractor: the public award data carries the prime
// recipient only, so labeling the edge prime-of-record is the honest statement of what the data supports.
// It is an identifier constant, so its hyphen is allowed; it never appears in surfaced prose.
const primeOfRecordLabel = "prime-of-record"

// DefaultRecompeteWindowMonths is the default recompete window when a caller binds a non positive one: the
// 12 to 24 month horizon the capture pipeline watches.
const DefaultRecompeteWindowMonths = 24

// maxRecompeteWindowMonths bounds a caller bound window so an oversized one cannot turn the pipeline into an
// unbounded scan. A five year horizon is the meaningful ceiling for a recompete watch.
const maxRecompeteWindowMonths = 60

// DefaultCompetitorLookbackYears is the default recency lookback for the competitor field when a caller
// binds a non positive one: a five year award history is the standard capture baseline.
const DefaultCompetitorLookbackYears = 5

// maxCompetitorLookbackYears bounds the lookback so an oversized one cannot widen the scan unbounded.
const maxCompetitorLookbackYears = 15

// segmentCompetitorFieldSQL is the FIXED competitor field aggregate. Given an agency ($1), a NAICS ($2),
// and a recency lookback in years ($3), it returns, per recipient UEI, the recency weighted dollar share of
// awards in that (agency, naics) segment, ordered share DESC with a UEI tiebreaker, so a dominant share
// reads wired and a fragmented field reads open. The honesty mandate is enforced by construction:
//
//   - the segment key is ONLY (awarding_agency, naics_code), the only keys usaspending_awards carries, never
//     a psc or set_aside or pop_state the table does not have,
//   - is_galapagos labels the tenant's own past performance row, so the capture team sees where it already
//     stands in its own field,
//   - the recency weight is a numeric step function on the award year (the more recent the period of
//     performance end, the heavier the weight), kept in the numeric type so the weighted dollars and the
//     share stay exact decimal,
//   - the share is the recency weighted dollars over the segment total recency weighted dollars, computed in
//     numeric, never float; a window function over the segment supplies the denominator,
//   - the evidence is the contributing award count per recipient, so the narration can cite the specific
//     awards behind a share.
//
// Determinism: the ORDER BY ends with the recipient_uei tiebreaker, a total order over the grouped rows, so
// the same inputs yield byte identical rows.
const segmentCompetitorFieldSQL = `
WITH segment AS (
    SELECT
        a.recipient_uei,
        a.recipient_name,
        a.is_galapagos,
        a.award_amount,
        CASE
            WHEN a.pop_end_date >= (now()::date - ($3::int * 365)) THEN 1.00::numeric
            WHEN a.pop_end_date >= (now()::date - ((2 * $3)::int * 365)) THEN 0.50::numeric
            ELSE 0.25::numeric
        END AS recency_weight
    FROM usaspending_awards a
    WHERE a.awarding_agency = $1
      AND a.naics_code = $2
      AND a.award_amount IS NOT NULL
), weighted AS (
    SELECT
        recipient_uei,
        max(recipient_name)                               AS recipient_name,
        bool_or(is_galapagos)                             AS is_galapagos,
        count(*)                                          AS award_count,
        sum(award_amount * recency_weight)::numeric       AS weighted_dollars
    FROM segment
    GROUP BY recipient_uei
)
SELECT
    w.recipient_uei,
    w.recipient_name,
    w.is_galapagos,
    w.award_count,
    w.weighted_dollars,
    CASE
        WHEN sum(w.weighted_dollars) OVER () = 0 THEN 0::numeric
        ELSE (w.weighted_dollars / sum(w.weighted_dollars) OVER ())::numeric
    END AS dollar_share
FROM weighted w
ORDER BY dollar_share DESC, w.recipient_uei`

// incumbentOfRecordSQL is the FIXED incumbent walk. Given an opportunity_records id ($1), it walks to the
// predecessor award in usaspending_awards keyed on the SAME (agency, naics) the award carries plus the
// opportunity's dedupe_key matched against the award's PIID and id, returning the prime recipient UEI
// labeled prime-of-record with the matched award as evidence. It is a JOIN, not a recursion: the
// predecessor is one hop away on a shared key, so a recursive CTE would add nothing.
//
// The honesty mandate is enforced: the award side keys only on (awarding_agency = o.agency, naics_code =
// o.naics), the only keys the award carries, and the incumbent edge is the constant prime-of-record label,
// never the staffing sub. The dedupe_key match is the cross source link the opportunity catalog already
// computes (the normalized solicitation or award id), tried against the award's PIID and its generated id,
// so a recompete opportunity finds its own predecessor award. The most recent matching award (by period of
// performance end) is the incumbent of record; the ORDER BY ends with the award generated_internal_id
// tiebreaker for a byte stable single row pick.
const incumbentOfRecordSQL = `
SELECT
    o.id                                AS opportunity_id,
    o.title                             AS opportunity_title,
    o.agency                            AS agency,
    o.naics                             AS naics,
    a.recipient_uei                     AS incumbent_uei,
    a.recipient_name                    AS incumbent_name,
    'prime-of-record'                   AS incumbent_role,
    a.generated_internal_id             AS award_id,
    a.award_amount                      AS award_amount,
    a.pop_end_date                      AS award_pop_end_date
FROM opportunity_records o
JOIN usaspending_awards a
  ON a.awarding_agency = o.agency
 AND a.naics_code = o.naics
 AND o.dedupe_key <> ''
 AND (a.award_id = o.dedupe_key OR a.generated_internal_id = o.dedupe_key)
WHERE o.id = $1
ORDER BY a.pop_end_date DESC NULLS LAST, a.generated_internal_id`

// recompetePipelineSQL is the FIXED recompete pipeline. Given an agency ($1) and a window in months ($2),
// it returns the opportunity_records and the usaspending_awards (priced or not) whose pop_end_date falls
// inside the forward window, the takeaway pipeline a capture team watches, each row carrying its source so an award
// row and an opportunity row are never conflated. The two arms are UNIONed and the result is ordered by
// pop_end_date with a record id tiebreaker, a total order for byte stable output.
//
// Both arms key the agency the same honest way: opportunity_records on its agency column, awards on
// awarding_agency. The award arm labels the incumbent prime-of-record (the only incumbent the data
// supports); the opportunity arm carries its own incumbent_uei when the catalog recorded one. estimated_value
// and award_amount are numeric, scanned into decimal, never float.
const recompetePipelineSQL = `
SELECT
    'opportunity_records'         AS source_table,
    o.id::text                    AS record_id,
    o.title                       AS title,
    o.agency                      AS agency,
    o.naics                       AS naics,
    o.incumbent_uei               AS incumbent_uei,
    o.estimated_value             AS dollars,
    o.value_source                AS value_source,
    o.pop_end_date                AS pop_end_date
FROM opportunity_records o
WHERE o.agency = $1
  AND o.pop_end_date IS NOT NULL
  AND o.pop_end_date >= now()::date
  AND o.pop_end_date <= (now()::date + ($2::int * 30))
UNION
SELECT
    'usaspending_awards'          AS source_table,
    a.generated_internal_id       AS record_id,
    a.description                 AS title,
    a.awarding_agency             AS agency,
    a.naics_code                  AS naics,
    a.recipient_uei               AS incumbent_uei,
    a.award_amount                AS dollars,
    a.source                      AS value_source,
    a.pop_end_date                AS pop_end_date
FROM usaspending_awards a
WHERE a.awarding_agency = $1
  AND a.pop_end_date IS NOT NULL
  AND a.pop_end_date >= now()::date
  AND a.pop_end_date <= (now()::date + ($2::int * 30))
ORDER BY pop_end_date, record_id`

// agencyBuyingPatternSQL is the FIXED buyer pattern. Given an agency or department prefix ($1) and an
// optional NAICS filter ($2, empty string means every NAICS), it returns the NAICS and PSC distribution, the
// set_aside mix, and the estimated_value band for that buyer, the where to position view. The agency match
// uses the sam_opportunities full_parent_path DOTTED string: the buyer is matched when the first segment of
// the dotted path (the department, split on the first dot) equals the prefix OR the prefix is a leading
// segment of the path, so a department prefix rolls up all its sub agencies WITHOUT a recursive CTE.
//
// The distribution comes from opportunity_records (which carries the full naics, psc, set_aside,
// estimated_value tuple) joined to the agencies the path rollup identified, so the slice is honestly
// opportunity_records sourced. estimated_value is numeric; the band thresholds are numeric literals, so the
// banding stays exact decimal. The result groups by (naics, psc, set_aside, value_band) and orders by the
// opportunity count DESC with a total order tiebreaker on the grouped keys.
const agencyBuyingPatternSQL = `
WITH agency_match AS (
    SELECT DISTINCT s.department, s.sub_agency
    FROM sam_opportunities s
    WHERE split_part(s.full_parent_path, '.', 1) = $1
       OR s.full_parent_path = $1
       OR s.full_parent_path LIKE ($1 || '.%')
), banded AS (
    SELECT
        o.naics,
        o.psc,
        o.set_aside,
        o.estimated_value,
        CASE
            WHEN o.estimated_value IS NULL THEN 'unpriced'
            WHEN o.estimated_value < 1000000::numeric THEN 'under_1m'
            WHEN o.estimated_value < 10000000::numeric THEN '1m_to_10m'
            WHEN o.estimated_value < 100000000::numeric THEN '10m_to_100m'
            ELSE 'over_100m'
        END AS value_band
    FROM opportunity_records o
    WHERE ($2 = '' OR o.naics = $2)
      AND (
            o.agency = $1
         OR EXISTS (
                SELECT 1 FROM agency_match am
                WHERE am.department = o.agency OR am.sub_agency = o.agency
            )
      )
)
SELECT
    b.naics,
    b.psc,
    b.set_aside,
    b.value_band,
    count(*)                                  AS opportunity_count,
    sum(coalesce(b.estimated_value, 0))::numeric AS banded_value
FROM banded b
GROUP BY b.naics, b.psc, b.set_aside, b.value_band
ORDER BY opportunity_count DESC, b.naics, b.psc, b.set_aside, b.value_band`

// ClampRecompeteWindowMonths normalizes a requested recompete window to the bounded range: a non positive or
// missing window becomes the default 24 months, and a window above the ceiling is capped, so the pipeline
// scan is always bounded.
func ClampRecompeteWindowMonths(months int) int {
	if months <= 0 {
		return DefaultRecompeteWindowMonths
	}
	if months > maxRecompeteWindowMonths {
		return maxRecompeteWindowMonths
	}
	return months
}

// ClampCompetitorLookbackYears normalizes a requested competitor lookback to the bounded range.
func ClampCompetitorLookbackYears(years int) int {
	if years <= 0 {
		return DefaultCompetitorLookbackYears
	}
	if years > maxCompetitorLookbackYears {
		return maxCompetitorLookbackYears
	}
	return years
}

// CompetitorRow is one recipient in a competitor field: the recipient, whether it is the tenant's own past
// performance (is_galapagos), the count of contributing awards (the evidence), the recency weighted dollars,
// and the recency weighted dollar share, both decimal, never float. It is pure market structure: there is no
// match score and no employment decision number on it.
type CompetitorRow struct {
	RecipientUEI    string          `json:"recipient_uei"`
	RecipientName   string          `json:"recipient_name"`
	IsGalapagos     bool            `json:"is_galapagos"`
	AwardCount      int             `json:"award_count"`
	WeightedDollars decimal.Decimal `json:"weighted_dollars"`
	DollarShare     decimal.Decimal `json:"dollar_share"`
}

// SegmentCompetitorField runs the registered segment_competitor_field intent and scans the result into typed
// rows, carrying the weighted dollars and the share as decimal.Decimal (the numeric columns are scanned
// directly into decimal, never through float64). The agency and naics key the (agency, naics) segment the
// only way usaspending_awards supports; the lookback is clamped to the bounded range. The generic
// ExecuteIntent path remains available to the narration layer.
func SegmentCompetitorField(ctx context.Context, db *tenant.DB, agency, naics string, lookbackYears int) ([]CompetitorRow, error) {
	if db == nil {
		return nil, errors.New("ai: no database")
	}
	var out []CompetitorRow
	err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if err := GuardReadOnlyQuery(segmentCompetitorFieldSQL, ReadAllowlist); err != nil {
			return err
		}
		rows, err := tx.Query(ctx, segmentCompetitorFieldSQL, agency, naics, ClampCompetitorLookbackYears(lookbackYears))
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var r CompetitorRow
			if err := rows.Scan(&r.RecipientUEI, &r.RecipientName, &r.IsGalapagos, &r.AwardCount, &r.WeightedDollars, &r.DollarShare); err != nil {
				return err
			}
			out = append(out, r)
		}
		return rows.Err()
	})
	return out, err
}

// IncumbentEdge is the predecessor award an opportunity recompetes against: the opportunity, the matched
// award, and the prime recipient UEI labeled prime-of-record (the only incumbent the public award data
// supports). award_amount is decimal, never float. It carries no score and no employment decision number.
type IncumbentEdge struct {
	OpportunityID    string           `json:"opportunity_id"`
	OpportunityTitle string           `json:"opportunity_title"`
	Agency           string           `json:"agency"`
	NAICS            string           `json:"naics"`
	IncumbentUEI     string           `json:"incumbent_uei"`
	IncumbentName    string           `json:"incumbent_name"`
	IncumbentRole    string           `json:"incumbent_role"`
	AwardID          string           `json:"award_id"`
	AwardAmount      *decimal.Decimal `json:"award_amount,omitempty"`
	AwardPopEndDate  string           `json:"award_pop_end_date,omitempty"`
}

// IncumbentOfRecord runs the registered incumbent_of_record intent and scans the matched predecessor awards
// into typed edges. Every returned edge is labeled prime-of-record by construction (the SQL emits the
// constant label), so a caller never mistakes the prime recipient for the staffing subcontractor the data
// does not carry. award_amount is scanned into decimal.
func IncumbentOfRecord(ctx context.Context, db *tenant.DB, opportunityID string) ([]IncumbentEdge, error) {
	if db == nil {
		return nil, errors.New("ai: no database")
	}
	var out []IncumbentEdge
	err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if err := GuardReadOnlyQuery(incumbentOfRecordSQL, ReadAllowlist); err != nil {
			return err
		}
		rows, err := tx.Query(ctx, incumbentOfRecordSQL, opportunityID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var (
				e      IncumbentEdge
				popEnd pgtype.Date
			)
			if err := rows.Scan(&e.OpportunityID, &e.OpportunityTitle, &e.Agency, &e.NAICS,
				&e.IncumbentUEI, &e.IncumbentName, &e.IncumbentRole, &e.AwardID, &e.AwardAmount, &popEnd); err != nil {
				return err
			}
			e.AwardPopEndDate = isoDate(popEnd)
			out = append(out, e)
		}
		return rows.Err()
	})
	return out, err
}

// PipelineRow is one recompete the capture team watches: the source table it came from (so an award is never
// conflated with an opportunity), the record, the agency, the incumbent UEI, the dollars (decimal, never
// float), and the period of performance end that puts it in the window. It carries no score.
type PipelineRow struct {
	SourceTable  string           `json:"source_table"`
	RecordID     string           `json:"record_id"`
	Title        string           `json:"title"`
	Agency       string           `json:"agency"`
	NAICS        string           `json:"naics"`
	IncumbentUEI string           `json:"incumbent_uei"`
	Dollars      *decimal.Decimal `json:"dollars,omitempty"`
	ValueSource  string           `json:"value_source"`
	PopEndDate   string           `json:"pop_end_date"`
}

// RecompetePipeline runs the registered recompete_pipeline intent for an agency and a window in months,
// scanning the dollars into decimal. The window is clamped to the bounded range so the scan is always
// bounded.
func RecompetePipeline(ctx context.Context, db *tenant.DB, agency string, windowMonths int) ([]PipelineRow, error) {
	if db == nil {
		return nil, errors.New("ai: no database")
	}
	var out []PipelineRow
	err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if err := GuardReadOnlyQuery(recompetePipelineSQL, ReadAllowlist); err != nil {
			return err
		}
		rows, err := tx.Query(ctx, recompetePipelineSQL, agency, ClampRecompeteWindowMonths(windowMonths))
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var (
				r      PipelineRow
				popEnd pgtype.Date
			)
			if err := rows.Scan(&r.SourceTable, &r.RecordID, &r.Title, &r.Agency, &r.NAICS,
				&r.IncumbentUEI, &r.Dollars, &r.ValueSource, &popEnd); err != nil {
				return err
			}
			r.PopEndDate = isoDate(popEnd)
			out = append(out, r)
		}
		return rows.Err()
	})
	return out, err
}

// BuyingPatternRow is one slice of an agency's buying pattern: the NAICS, PSC, set_aside, and value band of
// the buyer's opportunities, the count in that slice, and the banded value (decimal, never float). It is
// opportunity_records sourced, the only table that carries the full slice tuple. It carries no score.
type BuyingPatternRow struct {
	NAICS            string          `json:"naics"`
	PSC              string          `json:"psc"`
	SetAside         string          `json:"set_aside"`
	ValueBand        string          `json:"value_band"`
	OpportunityCount int             `json:"opportunity_count"`
	BandedValue      decimal.Decimal `json:"banded_value"`
}

// AgencyBuyingPattern runs the registered agency_buying_pattern intent for an agency or department prefix and
// an optional NAICS filter, scanning the banded value into decimal. The agency rollup is the full_parent_path
// string split, NOT a recursive CTE.
func AgencyBuyingPattern(ctx context.Context, db *tenant.DB, agencyPrefix, naics string) ([]BuyingPatternRow, error) {
	if db == nil {
		return nil, errors.New("ai: no database")
	}
	var out []BuyingPatternRow
	err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if err := GuardReadOnlyQuery(agencyBuyingPatternSQL, ReadAllowlist); err != nil {
			return err
		}
		rows, err := tx.Query(ctx, agencyBuyingPatternSQL, agencyPrefix, naics)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var r BuyingPatternRow
			if err := rows.Scan(&r.NAICS, &r.PSC, &r.SetAside, &r.ValueBand, &r.OpportunityCount, &r.BandedValue); err != nil {
				return err
			}
			out = append(out, r)
		}
		return rows.Err()
	})
	return out, err
}

// isoDate renders a nullable Postgres date as a stable ISO 8601 calendar date string, so an evidence date
// marshals deterministically and a null date renders as the empty string. The dollar columns above scan
// straight into *decimal.Decimal (the same path the awards repo uses), so no numeric ever passes through
// float64.
func isoDate(d pgtype.Date) string {
	if !d.Valid {
		return ""
	}
	return d.Time.Format("2006-01-02")
}