// Package eeo computes adverse impact reporting over the segregated eeo schema. It is the ONLY code
// path that reads demographic data, and it runs exclusively on the compliance database role through a
// separate pool. The application and scoring roles have no grant on the eeo schema, so demographics
// can never reach a score; they flow one way, into this report. The platform PRODUCES the four fifths
// data for the deploying employer's independent audit; it does not itself constitute the audit.
package eeo

import (
	"context"
	"errors"
	"time"

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

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

// ErrNoCompliance is returned when no compliance pool is configured, so adverse impact reporting is
// unavailable (the endpoint returns 503 rather than silently producing an empty, misleading report).
var ErrNoCompliance = errors.New("eeo: compliance database not configured")

// fourFifths is the 0.80 impact ratio threshold (the EEOC Uniform Guidelines four fifths rule). A
// group whose selection rate divided by the highest group rate falls below this is flagged.
var fourFifths = decimal.RequireFromString("0.80")

// Service computes adverse impact over the compliance pool. The pool authenticates as the compliance
// role and is the only pool with eeo access. Every query runs inside a tenant transaction so RLS
// scopes both the eeo and the outcome tables to the caller's org.
type Service struct{ compliance *tenant.DB }

// NewService builds the service. compliance may be nil (then AdverseImpact returns ErrNoCompliance).
func NewService(compliance *tenant.DB) *Service { return &Service{compliance: compliance} }

// candRow is one candidate's demographic group and whether they were selected at a stage.
type candRow struct {
	gender        string // display label, "unknown" when not provided
	race          string
	genderUnknown bool // true only when the value is genuinely missing, not a real value named "unknown"
	raceUnknown   bool
	selected      bool
}

// AdverseImpact builds the report for the caller's org: a selection stage analysis over recorded slate
// decisions and a screening stage analysis over served scores, each with four fifths impact ratios.
func (s *Service) AdverseImpact(ctx context.Context, now time.Time) (Report, error) {
	if !s.compliance.Ready() {
		return Report{}, ErrNoCompliance
	}
	rep := Report{
		GeneratedAt: now.UTC().Format(time.RFC3339),
		Source:      "Voluntary self identification demographic data in the segregated eeo schema, joined to recorded AEDT outcomes. Read by the compliance role only.",
		Method:      "Four fifths rule (EEOC Uniform Guidelines). Selection rate per demographic group divided by the highest group rate. A ratio below 0.80 is flagged. The unknown category is reported as a count and excluded from the ratio comparison. The platform supplies this data for an independent audit run by the deploying employer.",
	}
	err := s.compliance.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		sel, e := s.stageRows(ctx, tx, selectionQuery)
		if e != nil {
			return e
		}
		scr, e := s.stageRows(ctx, tx, screeningQuery)
		if e != nil {
			return e
		}
		from, to, e := dateRange(ctx, tx)
		if e != nil {
			return e
		}
		selNoDemo, e := countQuery(ctx, tx, selectionNoDemoQuery)
		if e != nil {
			return e
		}
		scrNoDemo, e := countQuery(ctx, tx, screeningNoDemoQuery)
		if e != nil {
			return e
		}
		rep.DataFrom, rep.DataTo = from, to
		rep.Selection = analyze("selection (recorded slate decisions)", sel, selNoDemo)
		rep.Screening = analyze("screening (served scores)", scr, scrNoDemo)
		return nil
	})
	if err != nil {
		return Report{}, err
	}
	return rep, nil
}

func (s *Service) stageRows(ctx context.Context, tx pgx.Tx, query string) ([]candRow, error) {
	rows, err := tx.Query(ctx, query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []candRow
	for rows.Next() {
		var r candRow
		if err := rows.Scan(&r.gender, &r.race, &r.genderUnknown, &r.raceUnknown, &r.selected); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

func countQuery(ctx context.Context, tx pgx.Tx, query string) (int, error) {
	var n int
	err := tx.QueryRow(ctx, query).Scan(&n)
	return n, err
}

func dateRange(ctx context.Context, tx pgx.Tx) (string, string, error) {
	var from, to *time.Time
	err := tx.QueryRow(ctx, `SELECT min(created_at), max(created_at) FROM slate_members`).Scan(&from, &to)
	if err != nil {
		return "", "", err
	}
	fmtT := func(t *time.Time) string {
		if t == nil {
			return ""
		}
		return t.UTC().Format("2006-01-02")
	}
	return fmtT(from), fmtT(to), nil
}

// genderRace is the intersectional map key. Keying on a struct, not a delimiter joined string, means a
// demographic value that happens to contain the display separator can never collide across cells.
type genderRace struct{ gender, race string }

// analyze rolls candidate rows into the three required cuts (by gender, by race or ethnicity, and the
// intersectional cross tab) plus the unknown count. Whether a group is a known comparison group comes
// from explicit per dimension flags carried out of SQL, never from overloading a sentinel string, so a
// real demographic value of "unknown" is treated as known data. noDemographics is the count of decided
// candidates with no consented demographics at all, surfaced to disclose consent coverage.
func analyze(stage string, rows []candRow, noDemographics int) AnalysisSection {
	byGender := map[string]*tally{}
	byRace := map[string]*tally{}
	byBoth := map[genderRace]*tally{}
	unknown := 0
	for _, r := range rows {
		add(byGender, r.gender, !r.genderUnknown, r.selected)
		add(byRace, r.race, !r.raceUnknown, r.selected)
		add(byBoth, genderRace{r.gender, r.race}, !r.genderUnknown && !r.raceUnknown, r.selected)
		if r.genderUnknown || r.raceUnknown {
			unknown++
		}
	}
	bothStats := make([]groupStat, 0, len(byBoth))
	for k, t := range byBoth {
		bothStats = append(bothStats, groupStat{label: k.gender + " / " + k.race, known: t.known, considered: t.considered, selected: t.selected})
	}
	return AnalysisSection{
		Stage:               stage,
		TotalN:              len(rows),
		UnknownCount:        unknown,
		NoDemographicsCount: noDemographics,
		ByGender:            impactRows(statsFrom(byGender)),
		ByRace:              impactRows(statsFrom(byRace)),
		Intersectional:      impactRows(bothStats),
	}
}

type tally struct {
	considered, selected int
	known                bool
}

type groupStat struct {
	label                string
	known                bool
	considered, selected int
}

// add tallies a candidate into a group keyed by any comparable type (a gender or race label, or the
// intersectional struct).
func add[K comparable](m map[K]*tally, key K, known, selected bool) {
	t := m[key]
	if t == nil {
		t = &tally{known: known}
		m[key] = t
	}
	t.considered++
	if selected {
		t.selected++
	}
}

func statsFrom(m map[string]*tally) []groupStat {
	out := make([]groupStat, 0, len(m))
	for label, t := range m {
		out = append(out, groupStat{label: label, known: t.known, considered: t.considered, selected: t.selected})
	}
	return out
}

// impactRows computes the selection rate per group and the four fifths impact ratio against the
// highest rate among KNOWN groups. A four fifths comparison needs at least two known groups: with
// fewer, the lone known group would self reference to a passing 1.0000 and mask a real disparity, so
// every known row emits n/a instead. Unknown groups are listed with their rate but are never the
// reference and are never flagged, since the unknown bucket is a count, not a comparison group.
func impactRows(groups []groupStat) []ImpactRow {
	ref := decimal.Zero
	knownCount := 0
	rates := make([]decimal.Decimal, len(groups))
	for i, g := range groups {
		r := decimal.Zero
		if g.considered > 0 {
			r = decimal.NewFromInt(int64(g.selected)).DivRound(decimal.NewFromInt(int64(g.considered)), 6)
		}
		rates[i] = r
		if g.known {
			knownCount++
			if r.GreaterThan(ref) {
				ref = r
			}
		}
	}
	validRef := knownCount >= 2 && ref.GreaterThan(decimal.Zero)
	out := make([]ImpactRow, 0, len(groups))
	for i, g := range groups {
		row := ImpactRow{Group: g.label, Considered: g.considered, Selected: g.selected, Rate: rates[i].StringFixed(4)}
		if g.known && validRef {
			ratio := rates[i].DivRound(ref, 6)
			row.ImpactRatio = ratio.StringFixed(4)
			row.Flagged = ratio.LessThan(fourFifths)
		} else {
			row.ImpactRatio = "n/a"
		}
		out = append(out, row)
	}
	sortRows(out)
	return out
}

// selectionQuery: the latest consented demographics per candidate joined to their recorded slate
// decisions. A candidate is selected if approved on any slate. RLS scopes both tables to the org.
const selectionQuery = `
WITH latest AS (
    SELECT DISTINCT ON (candidate_id) candidate_id,
           COALESCE(NULLIF(gender, ''), 'unknown')         AS gender,
           COALESCE(NULLIF(race_ethnicity, ''), 'unknown') AS race,
           (NULLIF(gender, '') IS NULL)                     AS gender_unknown,
           (NULLIF(race_ethnicity, '') IS NULL)             AS race_unknown
    FROM eeo.candidate_demographics
    WHERE consent_given = true
    ORDER BY candidate_id, collected_at DESC
)
SELECT l.gender, l.race, l.gender_unknown, l.race_unknown, bool_or(sm.decision = 'approved') AS selected
FROM latest l
JOIN slate_members sm ON sm.candidate_id = l.candidate_id
GROUP BY l.candidate_id, l.gender, l.race, l.gender_unknown, l.race_unknown`

// selectionNoDemoQuery counts decided candidates with no consented demographics, so the report
// discloses how many recorded decisions are outside the four fifths analysis for want of self ID.
const selectionNoDemoQuery = `
SELECT count(DISTINCT sm.candidate_id)
FROM slate_members sm
WHERE sm.decision IN ('approved', 'rejected')
  AND NOT EXISTS (
      SELECT 1 FROM eeo.candidate_demographics d
      WHERE d.candidate_id = sm.candidate_id AND d.consent_given = true)`

// screeningQuery: the latest consented demographics joined to served score events. A candidate is
// screened in if any served score was band green. audit_logs is outside RLS, so the org filter is
// explicit on the GUC.
const screeningQuery = `
WITH latest AS (
    SELECT DISTINCT ON (candidate_id) candidate_id,
           COALESCE(NULLIF(gender, ''), 'unknown')         AS gender,
           COALESCE(NULLIF(race_ethnicity, ''), 'unknown') AS race,
           (NULLIF(gender, '') IS NULL)                     AS gender_unknown,
           (NULLIF(race_ethnicity, '') IS NULL)             AS race_unknown
    FROM eeo.candidate_demographics
    WHERE consent_given = true
    ORDER BY candidate_id, collected_at DESC
),
served AS (
    SELECT (entity_id)::uuid AS candidate_id,
           bool_or((new_values->>'band') = 'green') AS screened_in
    FROM audit_logs
    WHERE action = 'aedt.score.served' AND entity_type = 'candidate'
      AND organization_id = NULLIF(current_setting('app.current_org_id', true), '')::uuid
      AND entity_id IS NOT NULL
    GROUP BY entity_id
)
SELECT l.gender, l.race, l.gender_unknown, l.race_unknown, bool_or(s.screened_in) AS selected
FROM latest l
JOIN served s ON s.candidate_id = l.candidate_id
GROUP BY l.candidate_id, l.gender, l.race, l.gender_unknown, l.race_unknown`

// screeningNoDemoQuery counts candidates with a served score event but no consented demographics.
const screeningNoDemoQuery = `
SELECT count(DISTINCT (entity_id)::uuid)
FROM audit_logs a
WHERE a.action = 'aedt.score.served' AND a.entity_type = 'candidate' AND a.entity_id IS NOT NULL
  AND a.organization_id = NULLIF(current_setting('app.current_org_id', true), '')::uuid
  AND NOT EXISTS (
      SELECT 1 FROM eeo.candidate_demographics d
      WHERE d.candidate_id = (a.entity_id)::uuid AND d.consent_given = true)`