Janady13/missionmatch / internal/platform/sourcing/ingest_bench.go

1package sourcing 2 3// Bench ingest extension. This file adds the WS9 bench loader path on top of the sourced lead ingest in 4// ingest.go, reusing its dedup and structured skills, certs, and agencies writes VERBATIM through the 5// shared writeStructuredCandidate helper. Three WS9 invariants are encoded here. 6// 7// Invariant 3, clearance stays unknown on import. The candidate is written through the SAME 8// UpsertExternalCandidate path, which hardcodes clearance_level 'unknown', clearance_poly 'none', and 9// clearance_status 'unknown'. IngestBenchRecord never threads an extracted clearance code to any clearance 10// column, so a resume regex can never score as a held clearance under the 0.25 clearance weight. A 11// clearance becomes known only later through an explicit org FSO provenance action. 12// 13// Invariant 6, embed is composed and run BEFORE the write transaction. IngestBenchRecord accepts a 14// PRECOMPUTED resume embedding vector. The embed network round trip happens in the CALLER, outside this 15// transaction, so a slow inference plane never holds the candidate row lock open across the embed timeout. 16// A nil vector commits the candidate WITHOUT an embedding, and the embed-backfill only-missing fallback 17// fills it later, so a skipped or failed embed never wedges a row. 18// 19// Invariant 2, mission tags are written through the APP role under RLS, never this migrator transaction. 20// IngestBenchRecord deliberately writes NO candidate_mission_tags row. TagCandidateMissions is a SEPARATE 21// function the import command calls AFTER the candidate commits, on the application pool inside 22// RunInTenantTx, so the RLS WITH CHECK validates the org on every tag insert. 23 24import ( 25 "context" 26 "fmt" 27 28 "github.com/google/uuid" 29 "github.com/jackc/pgx/v5" 30 pgvector "github.com/pgvector/pgvector-go" 31 32 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 33 "github.com/missionworkforce/missionmatch/internal/platform/extraction" 34 "github.com/missionworkforce/missionmatch/internal/platform/missionclass" 35 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 36) 37 38// BenchIngestResult is the outcome of ingesting one bench record: the resolved candidate id and whether 39// the row was a new insert or an enrich of an existing person (a dedup hit). The import command reports 40// these counts; New false is a dedup enrich, never a duplicate row. 41type BenchIngestResult struct { 42 CandidateID uuid.UUID 43 New bool // true when a new candidate was inserted, false when an existing person was enriched 44} 45 46// IngestBenchRecord resolves a bench person to a candidate and writes their structured workforce data, the 47// precomputed resume embedding, and one bench_import_log compliance row, ALL inside the one migrator 48// transaction the caller already opened (q is base.WithTx(tx)). It is the bench loader sibling of 49// IngestPerson: it reuses writeStructuredCandidate, so the dedup and the skills, certs, and agency writes 50// are byte identical to the sourced lead path. 51// 52// Clearance is never set here beyond the hardcoded 'unknown' the upsert writes (invariant 3). The embed 53// is NOT computed here: the caller composes the name free profile and calls inference.Client.Embed BEFORE 54// opening this transaction, then passes the finished vector(384) as embedding (invariant 6). A nil 55// embedding commits the candidate without a resume vector. One bench_import_log row is written per committed 56// record with the operator provided source, file_or_batch, and legal_basis. 57// 58// Mission tags are NOT written here (invariant 2): the caller writes candidate_mission_tags on the 59// application pool inside RunInTenantTx through TagCandidateMissions after this commits. 60func IngestBenchRecord(ctx context.Context, q *dbgen.Queries, orgID uuid.UUID, sp SourcedPerson, ext extraction.Extractor, embedding *pgvector.Vector, source, fileOrBatch, legalBasis string) (BenchIngestResult, error) { 61 candID, found, err := writeStructuredCandidate(ctx, q, orgID, sp, ext) 62 if err != nil { 63 return BenchIngestResult{}, err 64 } 65 66 // Invariant 6: write the precomputed vector in the SAME migrator transaction as the candidate, so the 67 // row and its embedding commit atomically. A nil vector skips the write entirely, leaving the row 68 // embeddable by the embed-backfill only-missing fallback; no embed network call happens here. 69 if embedding != nil { 70 if err := q.UpdateCandidateEmbedding(ctx, dbgen.UpdateCandidateEmbeddingParams{Embedding: *embedding, ID: candID}); err != nil { 71 return BenchIngestResult{}, fmt.Errorf("update candidate embedding: %w", err) 72 } 73 } 74 75 // One compliance row per committed record. candidate_id is set so an org can trace which bench person 76 // each row justifies. source is bench_file or bench_csv (never PDL); legal_basis is operator provided. 77 cid := candID 78 if err := q.InsertBenchImportLog(ctx, dbgen.InsertBenchImportLogParams{ 79 OrganizationID: orgID, 80 CandidateID: &cid, 81 Source: source, 82 FileOrBatch: fileOrBatch, 83 LegalBasis: legalBasis, 84 }); err != nil { 85 return BenchIngestResult{}, fmt.Errorf("insert bench import log: %w", err) 86 } 87 88 return BenchIngestResult{CandidateID: candID, New: !found}, nil 89} 90 91// candidateMissionTagWriter is the narrow port TagCandidateMissions writes through, so the classification is 92// unit testable with an injected fake and no database. The production implementation runs the write inside 93// RunInTenantTx on the application pool, exercising RLS on every tag row. 94type candidateMissionTagWriter interface { 95 upsertCandidateMissionTags(ctx context.Context, tags []dbgen.UpsertCandidateMissionTagParams) error 96} 97 98// TagCandidateMissions classifies one candidate's headline text and skill names with the deterministic 99// missionclass.Classify over the prebuilt AliasIndex, then upserts a candidate_mission_tag per Hit through 100// the APPLICATION pool inside RunInTenantTx (invariant 2). It is a SEPARATE function from IngestBenchRecord 101// and MUST NOT be called inside the migrator ingest transaction: the import command calls it AFTER the 102// candidate commits, mirroring cmd/tag-missions exactly. 103// 104// The org id stamped on every tag is read from the GUC context set by tenant.WithOrg, NEVER a caller 105// argument, so the RLS WITH CHECK validates the org on each insert and a tag can never land in another 106// tenant. The classifier is pure (no DB, no clock, no network), so the only DB touch is the tag write, 107// which the application role performs under RLS. 108func TagCandidateMissions(ctx context.Context, appDB *tenant.DB, candidateID uuid.UUID, headlineText string, skillNames []string, idx missionclass.AliasIndex) error { 109 return tagCandidateMissions(ctx, appDBTagWriter{db: appDB}, candidateID, headlineText, skillNames, idx) 110} 111 112// tagCandidateMissions is the inner, writer injected form. It does the pure classification and hands the 113// resulting params to the writer, so the test exercises the full classify to params mapping with a fake 114// writer and a fixed AliasIndex and asserts the upsert calls, with no DB. 115func tagCandidateMissions(ctx context.Context, w candidateMissionTagWriter, candidateID uuid.UUID, headlineText string, skillNames []string, idx missionclass.AliasIndex) error { 116 orgID, err := orgUUIDFromContext(ctx) 117 if err != nil { 118 return err 119 } 120 text := joinNonEmpty(append([]string{headlineText}, skillNames...)...) 121 var tags []dbgen.UpsertCandidateMissionTagParams 122 for _, h := range missionclass.Classify(text, "", "", idx) { 123 tags = append(tags, dbgen.UpsertCandidateMissionTagParams{ 124 OrganizationID: orgID, 125 CandidateID: candidateID, 126 MissionCode: h.Code, 127 Mechanism: h.Mechanism, 128 Evidence: h.Evidence, 129 }) 130 } 131 if err := w.upsertCandidateMissionTags(ctx, tags); err != nil { 132 return fmt.Errorf("sourcing: upsert candidate mission tags: %w", err) 133 } 134 return nil 135} 136 137// appDBTagWriter is the production candidateMissionTagWriter: it writes every tag inside ONE RunInTenantTx 138// as the application role, so RLS is exercised on each insert and the org comes from the validated GUC. 139type appDBTagWriter struct{ db *tenant.DB } 140 141func (w appDBTagWriter) upsertCandidateMissionTags(ctx context.Context, tags []dbgen.UpsertCandidateMissionTagParams) error { 142 if len(tags) == 0 { 143 return nil 144 } 145 return w.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 146 q := dbgen.New(tx) 147 for _, t := range tags { 148 if err := q.UpsertCandidateMissionTag(ctx, t); err != nil { 149 return err 150 } 151 } 152 return nil 153 }) 154} 155 156// orgUUIDFromContext reads the validated org id from the tenant context and parses it. A missing or 157// malformed org is a hard error: tagging refuses to run unscoped, mirroring RunInTenantTx fail closed. 158func orgUUIDFromContext(ctx context.Context) (uuid.UUID, error) { 159 raw, ok := tenant.OrgFromContext(ctx) 160 if !ok { 161 return uuid.Nil, tenant.ErrNoTenant 162 } 163 id, err := uuid.Parse(raw) 164 if err != nil { 165 return uuid.Nil, fmt.Errorf("sourcing: org id in context is not a uuid: %w", err) 166 } 167 return id, nil 168} 169 170// joinNonEmpty joins non empty parts with single spaces into the one blob the classifier reads. Empty parts 171// are dropped so a missing headline or skill never injects stray whitespace the word boundary matcher would 172// have to absorb. 173func joinNonEmpty(parts ...string) string { 174 out := "" 175 for _, p := range parts { 176 if p == "" { 177 continue 178 } 179 if out != "" { 180 out += " " 181 } 182 out += p 183 } 184 return out 185} 186