//go:build integration

// WS9 bench ingestion, exercised against a REAL Postgres 16 + pgvector with RLS in force. It drives the
// two WS9 seams the controller validates on staging: sourcing.IngestBenchRecord on the migrator pool (the
// candidate, skills, the precomputed embed, and the bench_import_log row, all in one migrator tx) and
// sourcing.TagCandidateMissions on the app pool inside RunInTenantTx (the mission tag write under RLS). It
// does NOT shell out to the import-bench binary; it drives the library seams directly so the data layer
// properties are proven against live Postgres.
//
// The proven invariants are: INVARIANT 3, clearance stays 'unknown' on import so a resume regex can never
// score as held; the candidate skills resolve to canonical ontology rows (the Signals Intelligence case);
// resume_embedding is populated when a vector is passed and NULL when nil; the mission tags land under RLS
// and a second org cannot see them; one bench_import_log row per imported record carries the operator
// legal_basis and the bench source; and a re run of the same record ENRICHES rather than duplicating.
package integration

import (
	"context"
	"testing"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	pgvector "github.com/pgvector/pgvector-go"

	"github.com/missionworkforce/missionmatch/internal/platform/benchimport"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/embeddings"
	"github.com/missionworkforce/missionmatch/internal/platform/extraction"
	"github.com/missionworkforce/missionmatch/internal/platform/missionclass"
	"github.com/missionworkforce/missionmatch/internal/platform/sourcing"
)

// benchTestLegalBasis is the operator provided controller basis the test records against every imported
// record, mirroring the required -legal-basis flag. It is never the PDL string (invariant 7).
const benchTestLegalBasis = "client provided cleared bench, client is the data controller"

// importOneBenchRecord drives the WS9 ingest seams for one record exactly as the import command does: it
// runs sourcing.IngestBenchRecord in one migrator transaction (candidate, skills, the precomputed vector,
// and one bench_import_log row), commits, then tags the candidate's missions on the app pool inside
// RunInTenantTx. The embed vector is precomputed by the CALLER (invariant 6), so a nil vector commits the
// row without an embedding. It returns the resolved candidate id and whether the row was a new insert.
func importOneBenchRecord(t *testing.T, org uuid.UUID, rec benchimport.BenchRecord, vec *pgvector.Vector, source, fileOrBatch string, idx missionclass.AliasIndex) (uuid.UUID, bool) {
	t.Helper()
	ctx := context.Background()
	ext := extraction.NewDeterministic()
	sp := rec.ToSourcedPerson()

	tx, err := migrator.Begin(ctx)
	if err != nil {
		t.Fatalf("begin migrator tx: %v", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback(ctx)
		}
	}()

	out, err := sourcing.IngestBenchRecord(ctx, dbgen.New(tx), org, sp, ext, vec, source, fileOrBatch, benchTestLegalBasis)
	if err != nil {
		t.Fatalf("IngestBenchRecord: %v", err)
	}
	if err := tx.Commit(ctx); err != nil {
		t.Fatalf("commit migrator tx: %v", err)
	}
	committed = true

	// Invariant 2: tag the candidate's missions on the APP pool inside RunInTenantTx after the candidate
	// commits, with the org from WithOrg so the RLS WITH CHECK validates it. The headline carries the SIGINT
	// token for a record whose JobTitle resolves to an intelligence mission.
	headline := rec.JobTitle
	if err := sourcing.TagCandidateMissions(ctxOrg(org), appDB, out.CandidateID, headline, sp.Skills, idx); err != nil {
		t.Fatalf("TagCandidateMissions: %v", err)
	}
	return out.CandidateID, out.New
}

func benchImportLogCount(t *testing.T, org, candID uuid.UUID) int {
	t.Helper()
	var n int
	if err := migrator.QueryRow(context.Background(),
		"SELECT count(*) FROM bench_import_log WHERE organization_id=$1 AND candidate_id=$2", org, candID).Scan(&n); err != nil {
		t.Fatalf("bench_import_log count: %v", err)
	}
	return n
}

// TestBenchImportIngestsUnderRLS proves the WS9 ingest seams end to end on live Postgres: a structured
// record with a Signals Intelligence skill and a SIGINT title plus a resume fixture both become candidates
// with the clearance unknown invariant held, the skill resolves to a canonical ontology row, the embed
// vector lands only when passed, the mission tag is RLS scoped, and one compliance row is written per
// record with the bench source and the operator legal basis.
func TestBenchImportIngestsUnderRLS(t *testing.T) {
	ctx := context.Background()
	org, otherOrg := seedOrg(t), seedOrg(t)
	// An empty AliasIndex: the SIGINT tag comes from the literal keyword in the title, so the test does not
	// depend on the curated alias rows being present in the database under test (mirrors the WS7 tag test).
	idx := missionclass.NewAliasIndex(map[string]string{})

	// A structured record with a Signals Intelligence skill and a SIGINT analyst title, embedded with a
	// deterministic vector so resume_embedding is populated. The clearance code is carried for triage only
	// and must never reach candidates.clearance_level (invariant 3).
	emb := embeddings.NewDeterministic(384)
	rawVec, _ := emb.Embed(ctx, "SIGINT Analyst | Signals Intelligence, Python, Splunk")
	pv := pgvector.NewVector(rawVec)
	csvRec := benchimport.BenchRecord{
		FullName:      "Sam Signals",
		Location:      "Honolulu, HI",
		JobTitle:      "SIGINT Analyst",
		Skills:        []string{"Signals Intelligence", "Python", "Splunk"},
		Years:         8,
		ClearanceCode: "ts_sci", // triage only, never written to candidates.clearance_level
		ExternalID:    "bench-csv-" + uuid.NewString(),
		Email:         "sam.signals." + uuid.NewString()[:8] + "@example.com",
	}
	csvID, csvNew := importOneBenchRecord(t, org, csvRec, &pv, "bench_csv", "bench_roster.csv", idx)
	if !csvNew {
		t.Fatalf("expected the CSV record to insert a new candidate")
	}

	// A second record from a resume fixture path, ingested WITHOUT a vector so resume_embedding stays NULL,
	// proving the embed skip path commits the row (the embed-backfill only-missing fallback fills it later).
	resumeRec := benchimport.BenchRecord{
		FullName:   "Rita Resume",
		Location:   "Albuquerque, NM",
		JobTitle:   "Cyber Security Engineer",
		Skills:     []string{"AWS", "Kubernetes"},
		Years:      6,
		ExternalID: "bench-file-" + uuid.NewString(),
	}
	resumeID, resumeNew := importOneBenchRecord(t, org, resumeRec, nil, "bench_file", "/bench/resumes/rita.pdf", idx)
	if !resumeNew {
		t.Fatalf("expected the resume record to insert a new candidate")
	}

	// ---- candidate rows: full_name set, clearance_level 'unknown' (INVARIANT 3) ----
	for _, tc := range []struct {
		id   uuid.UUID
		name string
	}{{csvID, "Sam Signals"}, {resumeID, "Rita Resume"}} {
		var fullName, clearance string
		if err := migrator.QueryRow(ctx,
			"SELECT full_name, clearance_level FROM candidates WHERE id=$1", tc.id).Scan(&fullName, &clearance); err != nil {
			t.Fatalf("read candidate %s: %v", tc.id, err)
		}
		if fullName != tc.name {
			t.Errorf("candidate full_name = %q, want %q", fullName, tc.name)
		}
		if clearance != "unknown" {
			t.Errorf("INVARIANT 3 BREACH: candidate %s clearance_level = %q, must stay 'unknown' on import", tc.id, clearance)
		}
	}

	// ---- candidate_skills resolved to canonical ontology skills (Signals Intelligence) ----
	skills := skillNamesFor(t, csvID)
	if !skills["Signals Intelligence"] {
		t.Errorf("the Signals Intelligence skill did not resolve to a candidate_skills ontology row; got %v", skills)
	}
	for _, want := range []string{"Python", "Splunk"} {
		if !skills[want] {
			t.Errorf("structured skill %q not written to candidate_skills; got %v", want, skills)
		}
	}
	// The skill is a real row in the global ontology skills table, reachable for matching.
	var skillN int
	if err := migrator.QueryRow(ctx, "SELECT count(*) FROM skills WHERE name='Signals Intelligence'").Scan(&skillN); err != nil {
		t.Fatalf("count Signals Intelligence skill: %v", err)
	}
	if skillN == 0 {
		t.Errorf("the Signals Intelligence ontology skill row was not created")
	}

	// ---- resume_embedding populated when a vector is passed, NULL when nil ----
	var hasVec bool
	if err := migrator.QueryRow(ctx, "SELECT resume_embedding IS NOT NULL FROM candidates WHERE id=$1", csvID).Scan(&hasVec); err != nil {
		t.Fatalf("read csv embedding presence: %v", err)
	}
	if !hasVec {
		t.Errorf("resume_embedding must be populated when a vector is passed (invariant 6 in tx write)")
	}
	var resumeHasVec bool
	if err := migrator.QueryRow(ctx, "SELECT resume_embedding IS NOT NULL FROM candidates WHERE id=$1", resumeID).Scan(&resumeHasVec); err != nil {
		t.Fatalf("read resume embedding presence: %v", err)
	}
	if resumeHasVec {
		t.Errorf("resume_embedding must be NULL when a nil vector is passed (the embed skip path)")
	}

	// ---- candidate_mission_tags present after TagCandidateMissions, RLS scoped ----
	var code, mechanism string
	if err := appDB.RunInTenantTx(ctxOrg(org), func(ctx context.Context, tx pgx.Tx) error {
		return tx.QueryRow(ctx,
			"SELECT mission_code, mechanism FROM candidate_mission_tags WHERE candidate_id=$1 AND mission_code='SIGINT'",
			csvID).Scan(&code, &mechanism)
	}); err != nil {
		t.Fatalf("read SIGINT tag for the bench candidate: %v", err)
	}
	if code != "SIGINT" || mechanism != missionclass.MechKeyword {
		t.Errorf("SIGINT tag = code %q mechanism %q, want SIGINT/keyword", code, mechanism)
	}
	// A second org MUST NOT see the tag: a count under its session is zero (RLS default deny).
	var foreignN int
	if err := appDB.RunInTenantTx(ctxOrg(otherOrg), func(ctx context.Context, tx pgx.Tx) error {
		return tx.QueryRow(ctx, "SELECT count(*) FROM candidate_mission_tags WHERE candidate_id=$1", csvID).Scan(&foreignN)
	}); err != nil {
		t.Fatalf("foreign org tag count: %v", err)
	}
	if foreignN != 0 {
		t.Fatalf("ISOLATION BREACH: a second org saw %d of the bench candidate's mission tags (want 0)", foreignN)
	}

	// ---- one bench_import_log row per imported record with the legal basis and the bench source ----
	if n := benchImportLogCount(t, org, csvID); n != 1 {
		t.Errorf("expected exactly one bench_import_log row for the csv record, got %d", n)
	}
	if n := benchImportLogCount(t, org, resumeID); n != 1 {
		t.Errorf("expected exactly one bench_import_log row for the resume record, got %d", n)
	}
	var basis, source string
	if err := migrator.QueryRow(ctx,
		"SELECT legal_basis, source FROM bench_import_log WHERE candidate_id=$1", csvID).Scan(&basis, &source); err != nil {
		t.Fatalf("read csv bench_import_log: %v", err)
	}
	if basis != benchTestLegalBasis {
		t.Errorf("bench_import_log legal_basis = %q, want the operator basis %q", basis, benchTestLegalBasis)
	}
	if source != "bench_csv" {
		t.Errorf("bench_import_log source = %q, want bench_csv", source)
	}
	var resumeSource string
	if err := migrator.QueryRow(ctx,
		"SELECT source FROM bench_import_log WHERE candidate_id=$1", resumeID).Scan(&resumeSource); err != nil {
		t.Fatalf("read resume bench_import_log: %v", err)
	}
	if resumeSource != "bench_file" {
		t.Errorf("resume bench_import_log source = %q, want bench_file", resumeSource)
	}

	// ---- dedup: re run the SAME csv record, assert it ENRICHES (no duplicate candidate) ----
	before := candidateOrgCount(t, org)
	reID, reNew := importOneBenchRecord(t, org, csvRec, &pv, "bench_csv", "bench_roster.csv", idx)
	if reNew {
		t.Fatalf("dedup failed: re running the same record reported a NEW candidate, not an enrich")
	}
	if reID != csvID {
		t.Fatalf("dedup resolved to a different candidate %s, want the original %s", reID, csvID)
	}
	if after := candidateOrgCount(t, org); after != before {
		t.Fatalf("dedup failed: candidate count went %d to %d (a duplicate was created)", before, after)
	}
	// The re run wrote a second compliance row for the same candidate (one per imported record), proving the
	// log is append friendly and audits every import event, while the candidate row was enriched not cloned.
	if n := benchImportLogCount(t, org, csvID); n != 2 {
		t.Errorf("expected two bench_import_log rows after the dedup re run (one per import), got %d", n)
	}
}

// TestBenchImportDistinctRecordsWithoutExternalID is the regression guard for the empty external id collapse.
// A bench file usually carries NO external id, and the candidates partial unique index is
// (organization_id, external_id) WHERE external_id IS NOT NULL. Passing the external id as an empty STRING
// (not NULL) made every external id free record conflict on (org, ”) and collapse onto one row. The fix
// passes NULL when empty so identity is resolved only by the email then name plus location dedup. This test
// imports two distinct people with no external id and asserts two distinct candidates, not one.
func TestBenchImportDistinctRecordsWithoutExternalID(t *testing.T) {
	org := seedOrg(t)
	idx := missionclass.NewAliasIndex(nil)

	recA := benchimport.BenchRecord{
		FullName: "Distinct Alpha", Location: "Honolulu HI", JobTitle: "Intelligence Analyst",
		Skills: []string{"SIGINT"}, Email: "distinct.alpha@example.com",
	}
	recB := benchimport.BenchRecord{
		FullName: "Distinct Bravo", Location: "Albuquerque NM", JobTitle: "Systems Engineer",
		Skills: []string{"Kubernetes"}, Email: "distinct.bravo@example.com",
	}

	idA, newA := importOneBenchRecord(t, org, recA, nil, "bench_csv", "roster.csv", idx)
	idB, newB := importOneBenchRecord(t, org, recB, nil, "bench_csv", "roster.csv", idx)

	if !newA || !newB {
		t.Fatalf("two external id free records with distinct identity must both be NEW inserts, got newA=%v newB=%v", newA, newB)
	}
	if idA == idB {
		t.Fatalf("REGRESSION: two external id free bench records collapsed onto one candidate %s (the empty external_id partial index bug)", idA)
	}
	if n := candidateOrgCount(t, org); n != 2 {
		t.Fatalf("expected two distinct candidates from two external id free records, got %d", n)
	}
}