//go:build integration

package integration

import (
	"context"
	"encoding/json"
	"testing"

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

	"github.com/missionworkforce/missionmatch/internal/modules/ai"
	"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"
)

// TestTeachingProducerAggregatesInTenantRows proves the WS8 Phase 1 producer end to end against real
// Postgres: the read is RLS scoped (it sees only the caller org's slate decisions, never a second org's),
// the audit event commits on the same tenant tx, and the per tenant aggregates land in the global
// teaching_signal_aggregates table with tenant_id set and dp_applied false. It seeds TWO orgs with distinct
// reviewer craft; running the producer for org A must fold only org A's evidence.
func TestTeachingProducerAggregatesInTenantRows(t *testing.T) {
	// Org A: a clearance gap override (required secret, clearance_score 0, approved) plus an adjacent skill
	// substitution stored in the evaluation (a non exact match at cosine 0.91, a high band), approved.
	orgA := seedOrg(t)
	_ = seedTeachingMember(t, orgA, teachingSeed{
		requiredLevel:  "secret",
		clearanceScore: "0",
		experience:     "100",
		decision:       "approved",
		matchedSkills:  []teachSeedSkill{{similarity: "0.91", exact: false, matched: true}},
	})

	// Org B: a completely different shape (no clearance required, an experience shortfall, rejected). If the
	// producer's read were not RLS scoped, org A's run would fold this too and the counts would be wrong.
	orgB := seedOrg(t)
	_ = seedTeachingMember(t, orgB, teachingSeed{
		requiredLevel:  "none",
		clearanceScore: "100",
		experience:     "20",
		decision:       "rejected",
		matchedSkills:  nil,
	})

	// A teaching client with TeachingEnabled false: Teach validates the boundary but transmits nothing.
	teacher, err := inference.New(inference.Config{BaseURL: "http://127.0.0.1:0", TeachingEnabled: false})
	if err != nil {
		t.Fatalf("inference client: %v", err)
	}
	// The aggregate writer is backed by the MIGRATOR pool (the global table the app role cannot write).
	writer := ai.NewMigratorAggregateWriter(migrator)
	prod := ai.NewTeachingProducer(appDB, writer, audit.New(appDB), teacher)

	res, err := prod.AggregateForOrg(ctxOrg(orgA))
	if err != nil {
		t.Fatalf("AggregateForOrg(orgA): %v", err)
	}
	// RLS scoped read: org A has exactly one slate member, so exactly one decision was read, never org B's.
	if res.DecisionsRead != 1 {
		t.Fatalf("DecisionsRead = %d, want 1 (the read must be RLS scoped to org A only)", res.DecisionsRead)
	}

	// Org A's evidence folds two patterns: clearance_gap_override_outcome (unknown band, approved) and
	// adjacent_skill_substitution_outcome (high band, approved). Both must be persisted for org A.
	rows := teachingAggregatesForTenant(t, orgA)
	if len(rows) != len(res.Signals) || len(rows) == 0 {
		t.Fatalf("persisted %d aggregate rows for org A, produced %d signals", len(rows), len(res.Signals))
	}
	sawClearance, sawSubstitution := false, false
	for _, r := range rows {
		if r.TenantID == nil || *r.TenantID != orgA {
			t.Fatalf("aggregate tenant_id = %v, want org A %v (a live in tenant row)", r.TenantID, orgA)
		}
		if r.DpApplied {
			t.Fatal("aggregate dp_applied = true, want false for a live in tenant row")
		}
		if r.ModelVersion != engine.ModelVersion {
			t.Fatalf("aggregate model_version = %q, want engine.ModelVersion %q", r.ModelVersion, engine.ModelVersion)
		}
		switch r.Pattern {
		case "clearance_gap_override_outcome":
			sawClearance = true
			if r.Band != "unknown" || r.Outcome != "approved" || r.Observations != 1 || r.Successes != 1 {
				t.Fatalf("clearance aggregate = %+v, want unknown/approved/1/1", r)
			}
		case "adjacent_skill_substitution_outcome":
			sawSubstitution = true
			if r.Band != "high" || r.Outcome != "approved" || r.Observations != 1 || r.Successes != 1 {
				t.Fatalf("substitution aggregate = %+v, want high/approved/1/1", r)
			}
		}
	}
	if !sawClearance || !sawSubstitution {
		t.Fatalf("missing expected aggregates: clearance=%v substitution=%v", sawClearance, sawSubstitution)
	}

	// Org B has no aggregates, proving the producer never wrote for an org it was not run on.
	if got := teachingAggregatesForTenant(t, orgB); len(got) != 0 {
		t.Fatalf("org B has %d aggregate rows, want 0 (the producer ran only for org A)", len(got))
	}

	// The teaching.signal.aggregated audit event committed on org A's tenant tx, carrying counts only.
	var auditCount int
	var newVals []byte
	if err := appDB.RunInTenantTx(ctxOrg(orgA), func(ctx context.Context, tx pgx.Tx) error {
		return tx.QueryRow(ctx,
			`SELECT count(*), max(new_values::text) FROM audit_logs
			   WHERE organization_id=$1 AND action='teaching.signal.aggregated'`,
			orgA).Scan(&auditCount, &newVals)
	}); err != nil {
		t.Fatalf("count teaching audit: %v", err)
	}
	if auditCount != 1 {
		t.Fatalf("want 1 teaching.signal.aggregated audit for org A, got %d", auditCount)
	}
	var payload map[string]int
	if err := json.Unmarshal(newVals, &payload); err != nil {
		t.Fatalf("audit payload not the counts only shape: %v (%s)", err, newVals)
	}
	if payload["decisions_aggregated"] != 1 {
		t.Fatalf("audit decisions_aggregated = %d, want 1", payload["decisions_aggregated"])
	}

	// A rerun is idempotent (ON CONFLICT upserts in place), so the row count does not grow.
	if _, err := prod.AggregateForOrg(ctxOrg(orgA)); err != nil {
		t.Fatalf("AggregateForOrg(orgA) rerun: %v", err)
	}
	if got := teachingAggregatesForTenant(t, orgA); len(got) != len(rows) {
		t.Fatalf("rerun changed the aggregate row count from %d to %d (upsert must be idempotent)", len(rows), len(got))
	}
}

// ---- seed helpers (write across tenants via the BYPASSRLS migrator) ----

type teachSeedSkill struct {
	similarity string
	exact      bool
	matched    bool
}

type teachingSeed struct {
	requiredLevel  string
	clearanceScore string
	experience     string
	decision       string
	matchedSkills  []teachSeedSkill
}

// seedTeachingMember seeds one requirement, one slate, one decided slate member, and (when matchedSkills is
// set) one candidate_evaluations row carrying the match evidence the producer reads. It returns the
// candidate id. Everything is written through the migrator so the seed can span orgs.
func seedTeachingMember(t *testing.T, org uuid.UUID, s teachingSeed) uuid.UUID {
	t.Helper()
	ctx := context.Background()
	q := dbgen.New(migrator)

	req, err := q.CreateRequirement(ctx, dbgen.CreateRequirementParams{
		OrganizationID: org, Title: "Teaching Role", RequiredLevel: s.requiredLevel, RequiredPoly: "none",
	})
	if err != nil {
		t.Fatalf("seed requirement: %v", err)
	}
	cand := seedCandidate(t, org, "none", "active", true)

	slate, err := q.CreateSlate(ctx, dbgen.CreateSlateParams{
		OrganizationID: org, RequirementID: req.ID, Name: "Teaching Slate", Status: "in_review",
	})
	if err != nil {
		t.Fatalf("seed slate: %v", err)
	}

	mv := engine.ModelVersion
	if _, err := q.AddSlateMember(ctx, dbgen.AddSlateMemberParams{
		OrganizationID: org, SlateID: slate.ID, CandidateID: cand.ID,
		TotalScore: dec0("80"), Band: "yellow",
		ClearanceScore: dec0(s.clearanceScore), SkillsScore: dec0("100"), CertsScore: dec0("100"),
		CustomerScore: dec0("100"), ExperienceScore: dec0(s.experience), LocationScore: dec0("100"),
		Explanation: "seeded", Trace: []byte("[]"), GapSummary: []byte("[]"), ModelVersion: &mv,
	}); err != nil {
		t.Fatalf("seed slate member: %v", err)
	}
	if _, err := q.SetSlateMemberDecision(ctx, dbgen.SetSlateMemberDecisionParams{
		SlateID: slate.ID, CandidateID: cand.ID, Decision: s.decision,
	}); err != nil {
		t.Fatalf("seed decision: %v", err)
	}

	if len(s.matchedSkills) > 0 {
		matched := make([]map[string]any, 0, len(s.matchedSkills))
		for _, sk := range s.matchedSkills {
			matched = append(matched, map[string]any{
				"RequiredSkillID": "", "RequiredSkillName": "skill", "CandidateSkillID": "",
				"CandidateName": "", "Similarity": sk.similarity, "Exact": sk.exact, "Matched": sk.matched,
			})
		}
		mj, err := json.Marshal(matched)
		if err != nil {
			t.Fatalf("marshal matched skills: %v", err)
		}
		if err := q.UpsertCandidateEvaluation(ctx, dbgen.UpsertCandidateEvaluationParams{
			OrganizationID: org, CandidateID: cand.ID, RequirementID: req.ID,
			TotalScore: dec0("80"), Band: "yellow",
			ClearanceScore: dec0(s.clearanceScore), SkillsScore: dec0("100"), CertsScore: dec0("100"),
			CustomerScore: dec0("100"), ExperienceScore: dec0(s.experience), LocationScore: dec0("100"),
			Explanation: "seeded", MatchedSkills: mj, GapSummary: []byte("[]"), ModelVersion: engine.ModelVersion,
		}); err != nil {
			t.Fatalf("seed evaluation: %v", err)
		}
	}
	return cand.ID
}

// teachingAggregatesForTenant reads the global teaching_signal_aggregates rows for one tenant through the
// migrator (the table is global, so the migrator sees all rows; we filter by tenant_id in the query).
func teachingAggregatesForTenant(t *testing.T, tenantID uuid.UUID) []dbgen.TeachingSignalAggregate {
	t.Helper()
	rows, err := migrator.Query(context.Background(),
		`SELECT id, tenant_id, dp_applied, pattern, band, outcome, model_version, observations, successes,
		        window_start, window_end, created_at
		   FROM teaching_signal_aggregates WHERE tenant_id=$1
		   ORDER BY pattern, band, outcome`, tenantID)
	if err != nil {
		t.Fatalf("query aggregates: %v", err)
	}
	defer rows.Close()
	var out []dbgen.TeachingSignalAggregate
	for rows.Next() {
		var r dbgen.TeachingSignalAggregate
		if err := rows.Scan(&r.ID, &r.TenantID, &r.DpApplied, &r.Pattern, &r.Band, &r.Outcome,
			&r.ModelVersion, &r.Observations, &r.Successes, &r.WindowStart, &r.WindowEnd, &r.CreatedAt); err != nil {
			t.Fatalf("scan aggregate: %v", err)
		}
		out = append(out, r)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("rows err: %v", err)
	}
	return out
}

// dec0 builds a decimal for a seed value, failing the test on a malformed literal.
func dec0(s string) decimal.Decimal { return decimal.RequireFromString(s) }