package ai

// teaching.go is the WS8 in tenant teaching producer (Phase 1, LIVE). It reads one tenant's reviewer craft
// (the slate decisions and the frozen match evidence) and folds it into the closed scrub.TeachingSignal
// vocabulary of contents free counts, so MEGAMIND can learn the CRAFT of the work (abstract patterns) and
// never a client fact. It is LIVE in the sense that it runs in production and persists the per tenant
// aggregates, but transmission to the inference plane stays gated off by TeachingEnabled until multiple
// live tenants exist, so nothing leaves the box yet.
//
// The producer never touches the engine math. It reads what the frozen engine already wrote (the decision,
// the sub scores, the gap summary, the matched skills) and counts it. It never changes a weight, a band
// cut, the semantic threshold, or any served score.
//
// Three correctness boundaries hold by construction:
//
//  1. The READ is RLS scoped. It runs inside RunInTenantTx on the NOBYPASSRLS app pool, so Postgres Row
//     Level Security restricts the join to the caller's own org. One tenant's producer can never read
//     another tenant's decisions.
//  2. The aggregate WRITE goes through the MIGRATOR pool seam, never the app role. teaching_signal_aggregates
//     is a global, migrator owned table the app role can only SELECT, so the live in tenant rows are written
//     through an injected Writer backed by the migrator pool.
//  3. The audit event runs through audit.Writer.OnTx INSIDE the same tenant tx that read the decisions, so
//     it commits with the read and carries the correct org. It carries counts only, no PII, no candidate id,
//     and no raw score.
//
// Everything score affecting is deferred to the two human step curation pipeline (Phase 4); this producer
// only aggregates evidence.

import (
	"context"
	"encoding/json"
	"sort"
	"time"

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

	"github.com/missionworkforce/missionmatch/internal/modules/matching/engine"
	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/inference"
	"github.com/missionworkforce/missionmatch/internal/platform/scrub"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

// init registers the live engine model version with the scrub boundary so a signal stamped with
// engine.ModelVersion is accepted, and a drifted version is refused the same way an unknown Pattern is. The
// version is pinned from the engine package (never a literal), the only module here allowed to import it.
func init() {
	scrub.RegisterModelVersion(engine.ModelVersion)
}

// teachingThresholds keeps the few derived classification rules in one place so the folding logic is a pure
// function of the stored evidence and the rules are visible in one spot for review.
const (
	// clearanceRequiredNone and clearanceRequiredUnknown are the two required_level values that mean no
	// clearance is required (matching engine.scoreClearance, which passes everyone below Confidential). Any
	// other required_level means a clearance was demanded, so a zero clearance_score on an approved member is
	// a reviewer override of a clearance gap.
	clearanceRequiredNone    = "none"
	clearanceRequiredUnknown = "unknown"
)

// DecisionRow is the contents free craft evidence for one slate member: the reviewer decision, the stored
// sub scores that reveal a clearance gap or an experience shortfall, the frozen gap summary, the required
// clearance level, and the matched skills evidence from the candidate's persisted evaluation. It is the
// shape ListSlateDecisionEvidenceForOrgRow already returns; the producer reads only these closed fields and
// never a name, an email, or a resume.
type DecisionRow = dbgen.ListSlateDecisionEvidenceForOrgRow

// DecisionReader is the RLS scoped read seam. The live implementation runs inside RunInTenantTx on the app
// pool; a test injects a fake returning fixture rows so the folding logic unit tests with no database.
type DecisionReader interface {
	ListDecisionEvidence(ctx context.Context) ([]DecisionRow, error)
}

// AggregateWriter is the migrator pool write seam for the global teaching_signal_aggregates table. The live
// implementation is backed by the migrator pool (the app role has SELECT only on this global table); a test
// injects a fake that records the upserts so the producer can be exercised with no database.
type AggregateWriter interface {
	UpsertTeachingSignalAggregate(ctx context.Context, arg dbgen.UpsertTeachingSignalAggregateParams) error
}

// AppReader is the app pool RunInTenantTx seam. *tenant.DB satisfies it. It is an interface so the audit on
// the same tenant tx can be exercised, while the folding logic itself unit tests through the DecisionReader.
type AppReader interface {
	RunInTenantTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error
}

// *tenant.DB is the live AppReader, a compile time assertion that the app pool satisfies the read seam.
var _ AppReader = (*tenant.DB)(nil)

// readerFromTx builds a DecisionReader bound to a tenant transaction, so the RLS scoped read runs on the
// same connection whose app.current_org_id GUC RunInTenantTx set.
type readerFromTx struct{ tx pgx.Tx }

// readerFromTx is the live DecisionReader: a compile time assertion that the tenant tx backed reader honors
// the seam the producer reads through.
var _ DecisionReader = readerFromTx{}

func (r readerFromTx) ListDecisionEvidence(ctx context.Context) ([]DecisionRow, error) {
	return dbgen.New(r.tx).ListSlateDecisionEvidenceForOrg(ctx)
}

// migratorAggregateWriter backs AggregateWriter with the migrator pool. It is a thin adapter so the
// producer depends on the interface, not the concrete dbgen DBTX, and a test can substitute a fake.
type migratorAggregateWriter struct{ db dbgen.DBTX }

func (w migratorAggregateWriter) UpsertTeachingSignalAggregate(ctx context.Context, arg dbgen.UpsertTeachingSignalAggregateParams) error {
	return dbgen.New(w.db).UpsertTeachingSignalAggregate(ctx, arg)
}

// NewMigratorAggregateWriter wraps a migrator scoped dbgen.DBTX (the migrator pool) as the global aggregate
// writer. The caller passes the migrator pool here, never the app pool, because the app role cannot write
// the global teaching_signal_aggregates table.
func NewMigratorAggregateWriter(migrator dbgen.DBTX) AggregateWriter {
	return migratorAggregateWriter{db: migrator}
}

// TeachingProducer folds one tenant's reviewer craft into scrub.TeachingSignal counts and persists the live
// in tenant aggregates. It is LIVE but transmission stays gated off by the inference client's TeachingEnabled
// flag, so calling Teach validates the boundary end to end (scrub.Marshal) without emitting anything off box.
type TeachingProducer struct {
	app     AppReader         // app pool RunInTenantTx, the RLS scoped read boundary
	writer  AggregateWriter   // migrator pool, the global aggregate write seam (never the app role)
	audit   *audit.Writer     // audit.Writer.OnTx on the tenant tx
	teacher *inference.Client // Teach, gated off by TeachingEnabled
	now     func() time.Time  // injectable clock so the window dates are deterministic in tests
}

// NewTeachingProducer wires the live producer. app is the app pool (RLS scoped reads), writer is backed by
// the MIGRATOR pool (global aggregate writes), auditW records the per tenant event on the tenant tx, and
// teacher is the inference client (Teach stays gated off by TeachingEnabled).
func NewTeachingProducer(app AppReader, writer AggregateWriter, auditW *audit.Writer, teacher *inference.Client) *TeachingProducer {
	return &TeachingProducer{app: app, writer: writer, audit: auditW, teacher: teacher, now: time.Now}
}

// Result reports what one AggregateForOrg run folded: how many slate decisions it read and the signals it
// produced. It carries counts only, never a candidate id or a score, so it is safe to log.
type Result struct {
	DecisionsRead int
	Signals       []scrub.TeachingSignal
}

// AggregateForOrg reads the org's slate decisions and folds them into per tenant craft aggregates. The org
// is taken from ctx (the validated tenant), never an argument.
//
// Order of operations:
//  1. Inside RunInTenantTx on the app pool (RLS scoped), read the decision evidence and emit the
//     teaching.signal.aggregated audit on the SAME tenant tx, so the audit commits with the read.
//  2. After the read commits, fold the rows into scrub.TeachingSignal counts (a pure function).
//  3. For each signal, call inference.Client.Teach (boundary checked, transmission gated off).
//  4. Persist each signal into teaching_signal_aggregates through the MIGRATOR pool writer, tenant_id set,
//     dp_applied false (the live in tenant rows). The app role cannot write this global table.
func (p *TeachingProducer) AggregateForOrg(ctx context.Context) (Result, error) {
	orgID, ok := tenant.OrgFromContext(ctx)
	if !ok {
		return Result{}, tenant.ErrNoTenant
	}

	// 1. RLS scoped read plus the audit on the SAME tenant tx (invariant 5: the per tenant event commits
	//    with the read it records and carries the correct org).
	var rows []DecisionRow
	err := p.app.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var rerr error
		rows, rerr = readerFromTx{tx: tx}.ListDecisionEvidence(ctx)
		if rerr != nil {
			return rerr
		}
		// Counts only in the audit payload: how many decisions this run aggregated. No candidate id, no
		// score, no name. The org is read from ctx by OnTx, never passed here.
		newVals, merr := json.Marshal(map[string]int{"decisions_aggregated": len(rows)})
		if merr != nil {
			return merr
		}
		return p.audit.OnTx(ctx, tx, audit.Event{
			Action:     "teaching.signal.aggregated",
			EntityType: "teaching_signal",
			NewValues:  newVals,
		})
	})
	if err != nil {
		return Result{}, err
	}

	// 2. Fold the evidence into the closed craft vocabulary (a pure function of the rows plus the window).
	signals := fold(rows, engine.ModelVersion)

	// 3 and 4. Exercise the scrub boundary through Teach (gated off, emits nothing) and persist the live in
	//          tenant aggregates through the migrator pool writer. Factored out so the teach and persist loop
	//          unit tests with fakes and no database, while the tenant tx read and audit above are covered by
	//          the integration test.
	if perr := p.persist(ctx, orgID, signals); perr != nil {
		return Result{}, perr
	}

	return Result{DecisionsRead: len(rows), Signals: signals}, nil
}

// persist runs the per signal teach and write loop: for each folded signal it calls inference.Client.Teach
// (which validates the scrub boundary and, with TeachingEnabled false, transmits nothing) and then upserts
// the live in tenant aggregate through the migrator pool writer with tenant_id set and dp_applied false. It
// touches no app pool and no tenant tx, so it is safe to unit test with a fake writer and a gated off teacher.
func (p *TeachingProducer) persist(ctx context.Context, orgID string, signals []scrub.TeachingSignal) error {
	tenantUUID, perr := parseOrgUUID(orgID)
	if perr != nil {
		return perr
	}
	start, end := p.window()
	for _, sig := range signals {
		if p.teacher != nil {
			if terr := p.teacher.Teach(ctx, sig); terr != nil {
				// Teach validates the boundary (scrub.Marshal). A failure means the producer built a signal
				// the boundary refuses, which is a bug to surface, not to swallow. Transmission is gated off.
				return terr
			}
		}
		if werr := p.writer.UpsertTeachingSignalAggregate(ctx, dbgen.UpsertTeachingSignalAggregateParams{
			TenantID:     &tenantUUID, // live in tenant row: tenant_id is set
			DpApplied:    false,       // not the cross tenant post DP path
			Pattern:      sig.Pattern,
			Band:         string(sig.Band),
			Outcome:      string(sig.Outcome),
			ModelVersion: sig.ModelVersion,
			Observations: int32(sig.Observations),
			Successes:    int32(sig.Successes),
			WindowStart:  pgDate(start),
			WindowEnd:    pgDate(end),
		}); werr != nil {
			return werr
		}
	}
	return nil
}

// foldKey identifies one aggregate bucket: a craft pattern at a similarity band for a decision outcome. The
// producer counts observations and successes per key, then turns each key into one scrub.TeachingSignal.
type foldKey struct {
	pattern string
	band    scrub.SimilarityBand
	outcome scrub.Outcome
}

type foldCount struct {
	observations int
	successes    int
}

// fold is the pure folding logic, a function of the stored decision evidence and the model version. It has
// no DB, no clock, and no I/O, so it unit tests with fixture rows.
//
// It folds into the four closed scrub.Patterns. Which patterns are derivable from the stored evidence today:
//
//   - adjacent_skill_substitution_outcome: DERIVABLE. candidate_evaluations.matched_skills carries each
//     required to candidate SkillMatch with its Similarity and an Exact flag (engine SkillMatch, no json
//     tags so the keys are the Go field names). A match with Matched true and Exact false is an adjacent or
//     semantic substitution. The Band comes from the cosine bucket of its Similarity, the Outcome from the
//     slate decision. A member with no stored evaluation (LEFT join NULL) contributes nothing here, which is
//     correct, the producer never fabricates a substitution it cannot see.
//
//   - clearance_gap_override_outcome: DERIVABLE. A clearance was required when required_level is neither
//     none nor unknown (matching engine.scoreClearance). A zero clearance_score on such a member is the
//     engine recording an unmet clearance, so a member with a required clearance and clearance_score zero is
//     a clearance gap; the reviewer decision generalizes whether they overrode it. The Band is unknown,
//     because clearance carries no similarity.
//
//   - experience_shortfall_outcome: DERIVABLE. The engine writes experience_score below 100 when the
//     candidate has fewer years than required (engine.scoreExperience). An experience shortfall is therefore
//     a member with experience_score below 100; the decision generalizes how reviewers treat it. Note the
//     engine does NOT add an experience entry to gap_summary (it appends only clearance, skill, and cert
//     gaps), so the score is the authoritative signal, not the gap summary text. The Band is unknown.
//
//   - reranker_label_agreement: ZERO TODAY. No reranker label and no reranker agreement is persisted with a
//     slate decision or a candidate evaluation. The inference reranker (inference.Client.Rerank) is not
//     stored alongside the frozen decision, so there is no stored evidence to fold. The producer derives
//     nothing for this pattern and emits no signal for it, rather than fabricate a count. When a reranker
//     label is persisted in a later phase, this fold gains a real branch.
func fold(rows []DecisionRow, modelVersion string) []scrub.TeachingSignal {
	counts := map[foldKey]*foldCount{}

	add := func(pattern string, band scrub.SimilarityBand, outcome scrub.Outcome, success bool) {
		k := foldKey{pattern: pattern, band: band, outcome: outcome}
		c := counts[k]
		if c == nil {
			c = &foldCount{}
			counts[k] = c
		}
		c.observations++
		if success {
			c.successes++
		}
	}

	for _, row := range rows {
		outcome, ok := outcomeFromDecision(row.Decision)
		if !ok {
			continue // an out of vocabulary decision is skipped rather than folded into a wrong bucket
		}
		// A success is the reviewer approving the candidate despite (or with) the observed evidence; it is
		// the positive class a craft signal generalizes over.
		approved := outcome == scrub.OutcomeApproved

		// adjacent_skill_substitution_outcome: one observation per non exact matched skill, banded by its
		// similarity. A member with no stored evaluation contributes none.
		for _, sm := range parseMatchedSkills(row.MatchedSkills) {
			if sm.Matched && !sm.Exact {
				add("adjacent_skill_substitution_outcome", bandFromSimilarity(sm.Similarity), outcome, approved)
			}
		}

		// clearance_gap_override_outcome: a required clearance the engine scored zero is a clearance gap.
		if clearanceRequired(row.RequiredLevel) && row.ClearanceScore.IsZero() {
			add("clearance_gap_override_outcome", scrub.BandUnknown, outcome, approved)
		}

		// experience_shortfall_outcome: an experience_score below 100 is a years shortfall.
		if row.ExperienceScore.LessThan(decimal.NewFromInt(100)) {
			add("experience_shortfall_outcome", scrub.BandUnknown, outcome, approved)
		}

		// reranker_label_agreement: no stored evidence today, intentionally folds nothing (see the doc above).
	}

	return signalsFromCounts(counts, modelVersion)
}

// signalsFromCounts turns the folded buckets into scrub.TeachingSignals in a deterministic order (by
// pattern, then band, then outcome), so the producer output and the persisted rows are stable across runs.
func signalsFromCounts(counts map[foldKey]*foldCount, modelVersion string) []scrub.TeachingSignal {
	keys := make([]foldKey, 0, len(counts))
	for k := range counts {
		keys = append(keys, k)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].pattern != keys[j].pattern {
			return keys[i].pattern < keys[j].pattern
		}
		if keys[i].band != keys[j].band {
			return keys[i].band < keys[j].band
		}
		return keys[i].outcome < keys[j].outcome
	})
	out := make([]scrub.TeachingSignal, 0, len(keys))
	for _, k := range keys {
		c := counts[k]
		out = append(out, scrub.TeachingSignal{
			Pattern:      k.pattern,
			Band:         k.band,
			Outcome:      k.outcome,
			Observations: c.observations,
			Successes:    c.successes,
			ModelVersion: modelVersion,
		})
	}
	return out
}

// matchedSkill mirrors the closed subset of engine.SkillMatch the producer needs. The engine type has no
// json tags, so the persisted jsonb keys are the Go field names; these tags pin those names so the producer
// is independent of the engine package's json encoding and reads only the evidence fields it folds.
type matchedSkill struct {
	Similarity decimal.Decimal `json:"Similarity"`
	Exact      bool            `json:"Exact"`
	Matched    bool            `json:"Matched"`
}

// parseMatchedSkills decodes the candidate_evaluations.matched_skills jsonb. A NULL column (a slate member
// never screened through screenMATCH) or an empty array yields no skills, so the producer folds nothing for
// such a member rather than guess.
func parseMatchedSkills(raw []byte) []matchedSkill {
	if len(raw) == 0 {
		return nil
	}
	var skills []matchedSkill
	if err := json.Unmarshal(raw, &skills); err != nil {
		return nil // unreadable evidence is treated as absent, never fabricated
	}
	return skills
}

// outcomeFromDecision maps the slate_members.decision vocabulary to the scrub.Outcome vocabulary. They are
// the same three labels by construction (the scrub Outcome constants were taken verbatim from
// slate_members.decision), so this is a checked passthrough that refuses an unexpected label.
func outcomeFromDecision(decision string) (scrub.Outcome, bool) {
	switch decision {
	case "approved":
		return scrub.OutcomeApproved, true
	case "rejected":
		return scrub.OutcomeRejected, true
	case "pending":
		return scrub.OutcomePending, true
	default:
		return "", false
	}
}

// bandFromSimilarity buckets a cosine similarity into the coarse scrub.SimilarityBand, the public
// quantization the teaching boundary speaks in (never the raw score). The cut points match the scrub band
// documentation: exact at 1.00, high 0.90 to 0.99, medium 0.80 to 0.89, low below 0.80. A negative or
// missing similarity is unknown.
func bandFromSimilarity(sim decimal.Decimal) scrub.SimilarityBand {
	switch {
	case sim.GreaterThanOrEqual(decimal.NewFromInt(1)):
		return scrub.BandExact
	case sim.GreaterThanOrEqual(decimal.RequireFromString("0.90")):
		return scrub.BandHigh
	case sim.GreaterThanOrEqual(decimal.RequireFromString("0.80")):
		return scrub.BandMedium
	case sim.GreaterThanOrEqual(decimal.Zero):
		return scrub.BandLow
	default:
		return scrub.BandUnknown
	}
}

// clearanceRequired reports whether the requirement demanded a clearance, matching engine.scoreClearance
// which passes everyone when the required level is below Confidential (the none and unknown labels).
func clearanceRequired(requiredLevel string) bool {
	return requiredLevel != clearanceRequiredNone && requiredLevel != clearanceRequiredUnknown
}

// window returns the observation window the producer stamps on each aggregate. v1 aggregates the full
// stored history into one open ended window ending today, so a rerun refreshes the same window row in place
// (the upsert key includes window_start). The window dates carry no client fact, only the run date.
func (p *TeachingProducer) window() (start, end time.Time) {
	now := p.now().UTC()
	// A fixed epoch start keeps window_start stable across reruns so the upsert key is stable; the end moves
	// to the run date. This is the all history window for v1, not a rolling window.
	start = time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)
	end = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
	return start, end
}

// pgDate wraps a date for the date columns. The producer always sets a valid date, never NULL.
func pgDate(t time.Time) pgtype.Date {
	return pgtype.Date{Time: t, Valid: true}
}

// parseOrgUUID parses the validated org id from context into a uuid for the tenant_id column. The org
// originates from RunInTenantTx's validated context, so a parse failure is a programming error, not client
// input.
func parseOrgUUID(orgID string) (uuid.UUID, error) {
	return uuid.Parse(orgID)
}