Janady13/missionmatch / internal/modules/matching/service.go

1package matching 2 3import ( 4 "context" 5 "encoding/json" 6 "errors" 7 "fmt" 8 "sort" 9 "time" 10 11 "github.com/google/uuid" 12 "github.com/jackc/pgx/v5" 13 "github.com/shopspring/decimal" 14 15 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 16 "github.com/missionworkforce/missionmatch/internal/modules/matching/intelligence" 17 "github.com/missionworkforce/missionmatch/internal/platform/audit" 18 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 19 "github.com/missionworkforce/missionmatch/internal/platform/extraction" 20 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 21) 22 23// ErrNoDatabase is returned when a DB-backed operation is attempted but no pool is configured. 24var ( 25 ErrNoDatabase = errors.New("matching: no database configured") 26 ErrNotFound = errors.New("matching: not found") 27 ErrBadRequest = errors.New("matching: bad request") 28 ErrNoExtractor = errors.New("matching: no extractor configured") 29) 30 31const defaultMaxCandidates = 500 32 33// Service is the slate-loop application service. It runs every DB operation inside a tenant 34// transaction (RLS-scoped) and delegates scoring to the pure engine. 35type Service struct { 36 db *tenant.DB 37 matcher engine.SkillMatcher // injected semantic skill matcher; may be nil (exact-only) 38 aud *audit.Writer // AEDT audit trail; may be nil (tests skip audit) 39 extractor extraction.Extractor // Mission Check resume/JD extractor; may be nil (Mission Check then 503) 40 clock func() time.Time // injectable for deterministic tests 41} 42 43// NewService builds the service. db may be nil (DB routes then return ErrNoDatabase). aud may be nil. 44func NewService(db *tenant.DB, matcher engine.SkillMatcher, aud *audit.Writer) *Service { 45 return &Service{db: db, matcher: matcher, aud: aud, clock: time.Now} 46} 47 48// WithExtractor injects the Mission Check structured extractor and returns the service for chaining. It 49// is a separate setter so the NewService signature (and its existing callers, including the integration 50// tests) stay unchanged. A service with no extractor returns ErrNoExtractor on the Mission Check paths. 51func (s *Service) WithExtractor(ext extraction.Extractor) *Service { 52 s.extractor = ext 53 return s 54} 55 56// auditOnTx records an AEDT event on the caller's tenant tx if an audit writer is configured. The 57// actor is the validated session principal; demographics are NEVER carried in an audit event. It 58// returns the write error so the caller fails closed: an AEDT score or decision that cannot be 59// recorded rolls back rather than being served unaudited (the audit package contract). 60func (s *Service) auditOnTx(ctx context.Context, tx pgx.Tx, e audit.Event) error { 61 if s.aud == nil { 62 return nil 63 } 64 if p, ok := tenant.UserFromContext(ctx); ok { 65 if id, err := uuid.Parse(p.UserID); err == nil { 66 e.ActorUserID = &id 67 } 68 } 69 return s.aud.OnTx(ctx, tx, e) 70} 71 72// servedScoreEvent builds the aedt.score.served event for a single served candidate score. It carries 73// the candidate, the tool, the score, and the band so the compliance pool can reconstruct a screening 74// stage adverse impact analysis by joining to eeo, and NOTHING about demographics. 75func servedScoreEvent(tool string, candidateID uuid.UUID, res engine.MatchResult) audit.Event { 76 payload, _ := json.Marshal(map[string]any{ 77 "tool": tool, 78 "score": res.Total.StringFixed(2), 79 "band": res.Band.Code(), 80 }) 81 return audit.Event{Action: "aedt.score.served", EntityType: "candidate", EntityID: candidateID.String(), NewValues: payload} 82} 83 84func (s *Service) orgID(ctx context.Context) (uuid.UUID, error) { 85 raw, ok := tenant.OrgFromContext(ctx) 86 if !ok { 87 return uuid.Nil, tenant.ErrNoTenant 88 } 89 id, err := uuid.Parse(raw) 90 if err != nil { 91 return uuid.Nil, fmt.Errorf("matching: invalid org id in context: %w", err) 92 } 93 return id, nil 94} 95 96// ScreenCandidate scores one candidate against one requirement (screenMATCH / POST /match). Read-only. 97func (s *Service) ScreenCandidate(ctx context.Context, candidateID, requirementID uuid.UUID) (engine.MatchResult, error) { 98 if s.db == nil { 99 return engine.MatchResult{}, ErrNoDatabase 100 } 101 asOf := s.clock() 102 var res engine.MatchResult 103 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 104 q := dbgen.New(tx) 105 er, err := loadRequirement(ctx, q, requirementID) 106 if err != nil { 107 return err 108 } 109 c, err := q.GetCandidate(ctx, candidateID) 110 if err != nil { 111 return fmt.Errorf("get candidate: %w", err) 112 } 113 ec, err := loadCandidate(ctx, q, c, asOf) 114 if err != nil { 115 return err 116 } 117 res = engine.ScoreCandidate(ctx, ec, er, s.matcher) 118 // A focused single candidate qualification is a screening stage AEDT output. Record it as a 119 // served score (candidate, tool, score, band, no demographics) so adverse impact at the 120 // screening stage stays visible even before any slate decision exists. Fail closed if it cannot 121 // be recorded. 122 return s.auditOnTx(ctx, tx, servedScoreEvent("screenMATCH.match", candidateID, res)) 123 }) 124 return res, err 125} 126 127// GenerateSlate scores every candidate against a requirement, ranks them, and persists the slate 128// plus the frozen decision records (slateMATCH / POST /slates/generate). One atomic transaction. 129func (s *Service) GenerateSlate(ctx context.Context, in GenerateSlateInput) (SlateResult, error) { 130 if s.db == nil { 131 return SlateResult{}, ErrNoDatabase 132 } 133 if in.Name == "" { 134 return SlateResult{}, errors.New("matching: slate name is required") 135 } 136 if in.MaxCandidates <= 0 { 137 in.MaxCandidates = defaultMaxCandidates 138 } 139 orgID, err := s.orgID(ctx) 140 if err != nil { 141 return SlateResult{}, err 142 } 143 asOf := s.clock() 144 var out SlateResult 145 err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 146 var e error 147 out, e = generateSlate(ctx, dbgen.New(tx), orgID, in, s.matcher, asOf) 148 if e != nil { 149 return e 150 } 151 payload, _ := json.Marshal(map[string]any{"requirement_id": in.RequirementID.String(), "members": len(out.Members)}) 152 return s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.slate.generated", EntityType: "slate", EntityID: out.SlateID.String(), NewValues: payload}) 153 }) 154 return out, err 155} 156 157// DecideSlateMember records the human approve or reject of a slate member. This is the human in the 158// loop control: the engine never decides; a person with slate authority does, and the action is 159// recorded with the actor and the time, plus an AEDT audit event. decision must be approved or 160// rejected (pending is the un decided default and cannot be set here). 161func (s *Service) DecideSlateMember(ctx context.Context, slateID, candidateID uuid.UUID, decision string) error { 162 if s.db == nil { 163 return ErrNoDatabase 164 } 165 if decision != "approved" && decision != "rejected" { 166 return ErrBadRequest 167 } 168 p, ok := tenant.UserFromContext(ctx) 169 if !ok { 170 return tenant.ErrNoTenant 171 } 172 decidedBy, err := uuid.Parse(p.UserID) 173 if err != nil { 174 return tenant.ErrNoTenant 175 } 176 return s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 177 q := dbgen.New(tx) 178 row, e := q.SetSlateMemberDecision(ctx, dbgen.SetSlateMemberDecisionParams{ 179 SlateID: slateID, CandidateID: candidateID, Decision: decision, DecidedBy: &decidedBy, 180 }) 181 if errors.Is(e, pgx.ErrNoRows) { 182 return ErrNotFound 183 } 184 if e != nil { 185 return e 186 } 187 payload, _ := json.Marshal(map[string]any{"decision": decision, "band": row.Band}) 188 return s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.slate_member.decided", EntityType: "candidate", EntityID: candidateID.String(), NewValues: payload}) 189 }) 190} 191 192// SlateMemberTrace returns the persisted per component trace, rationale, and gap summary for a frozen 193// decision, the audit and candidate facing explanation of what was assessed and why. 194func (s *Service) SlateMemberTrace(ctx context.Context, slateID, candidateID uuid.UUID) (dbgen.SlateMember, error) { 195 if s.db == nil { 196 return dbgen.SlateMember{}, ErrNoDatabase 197 } 198 var m dbgen.SlateMember 199 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 200 row, e := dbgen.New(tx).GetSlateMember(ctx, dbgen.GetSlateMemberParams{SlateID: slateID, CandidateID: candidateID}) 201 if errors.Is(e, pgx.ErrNoRows) { 202 return ErrNotFound 203 } 204 if e != nil { 205 return e 206 } 207 m = row 208 return nil 209 }) 210 return m, err 211} 212 213// ListCandidates / ListRequirements are the read endpoints (GET /candidates, /requirements). 214func (s *Service) ListCandidates(ctx context.Context, limit, offset int32) ([]dbgen.Candidate, error) { 215 if s.db == nil { 216 return nil, ErrNoDatabase 217 } 218 var out []dbgen.Candidate 219 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 220 var e error 221 out, e = dbgen.New(tx).ListCandidates(ctx, dbgen.ListCandidatesParams{Limit: limit, Offset: offset}) 222 return e 223 }) 224 return out, err 225} 226 227func (s *Service) ListRequirements(ctx context.Context, limit, offset int32) ([]dbgen.Requirement, error) { 228 if s.db == nil { 229 return nil, ErrNoDatabase 230 } 231 var out []dbgen.Requirement 232 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 233 var e error 234 out, e = dbgen.New(tx).ListRequirements(ctx, dbgen.ListRequirementsParams{Limit: limit, Offset: offset}) 235 return e 236 }) 237 return out, err 238} 239 240// GetCandidate returns one candidate by id, scoped to the caller's org by RLS. A missing id or one in 241// another org both surface as ErrNotFound: RLS makes the two indistinguishable, which is the intended 242// privacy floor (no cross org existence oracle). The by id read backs a deep linked evaluation, so the 243// page no longer depends on the candidate being inside the first list page. 244func (s *Service) GetCandidate(ctx context.Context, id uuid.UUID) (dbgen.Candidate, error) { 245 if s.db == nil { 246 return dbgen.Candidate{}, ErrNoDatabase 247 } 248 var out dbgen.Candidate 249 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 250 c, e := dbgen.New(tx).GetCandidate(ctx, id) 251 if errors.Is(e, pgx.ErrNoRows) { 252 return ErrNotFound 253 } 254 if e != nil { 255 return e 256 } 257 out = c 258 return nil 259 }) 260 return out, err 261} 262 263// GetRequirement returns one requirement by id, scoped to the caller's org by RLS. Missing and cross 264// org both surface as ErrNotFound, mirroring GetCandidate. 265func (s *Service) GetRequirement(ctx context.Context, id uuid.UUID) (dbgen.Requirement, error) { 266 if s.db == nil { 267 return dbgen.Requirement{}, ErrNoDatabase 268 } 269 var out dbgen.Requirement 270 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 271 rq, e := dbgen.New(tx).GetRequirement(ctx, id) 272 if errors.Is(e, pgx.ErrNoRows) { 273 return ErrNotFound 274 } 275 if e != nil { 276 return e 277 } 278 out = rq 279 return nil 280 }) 281 return out, err 282} 283 284// generateSlate is the pure orchestration over a Querier, testable with a fake, no live DB. 285func generateSlate(ctx context.Context, q dbgen.Querier, orgID uuid.UUID, in GenerateSlateInput, matcher engine.SkillMatcher, asOf time.Time) (SlateResult, error) { 286 er, err := loadRequirement(ctx, q, in.RequirementID) 287 if err != nil { 288 return SlateResult{}, err 289 } 290 cands, err := q.ListCandidates(ctx, dbgen.ListCandidatesParams{Limit: in.MaxCandidates, Offset: 0}) 291 if err != nil { 292 return SlateResult{}, fmt.Errorf("list candidates: %w", err) 293 } 294 295 type scored struct { 296 id uuid.UUID 297 res engine.MatchResult 298 } 299 results := make([]scored, 0, len(cands)) 300 for _, c := range cands { 301 ec, err := loadCandidate(ctx, q, c, asOf) 302 if err != nil { 303 return SlateResult{}, err 304 } 305 results = append(results, scored{id: c.ID, res: engine.ScoreCandidate(ctx, ec, er, matcher)}) 306 } 307 // Deterministic ranking: total score descending, candidate id ascending as the unique tiebreaker. 308 sort.SliceStable(results, func(i, j int) bool { 309 if d := results[i].res.Total.Cmp(results[j].res.Total); d != 0 { 310 return d > 0 311 } 312 return results[i].id.String() < results[j].id.String() 313 }) 314 315 slate, err := q.CreateSlate(ctx, dbgen.CreateSlateParams{ 316 OrganizationID: orgID, RequirementID: in.RequirementID, Name: in.Name, Status: "draft", CreatedBy: in.CreatedBy, 317 }) 318 if err != nil { 319 return SlateResult{}, fmt.Errorf("create slate: %w", err) 320 } 321 322 members := make([]SlateMemberView, 0, len(results)) 323 for i, sc := range results { 324 rank := int32(i + 1) 325 params, err := ToSlateMemberParams(orgID, slate.ID, sc.id, sc.res, &rank) 326 if err != nil { 327 return SlateResult{}, err 328 } 329 if _, err := q.AddSlateMember(ctx, params); err != nil { 330 return SlateResult{}, fmt.Errorf("add slate member: %w", err) 331 } 332 members = append(members, slateMemberView(sc.id.String(), int(rank), sc.res)) 333 } 334 return SlateResult{SlateID: slate.ID, RequirementID: in.RequirementID, Name: in.Name, Members: members}, nil 335} 336 337func loadRequirement(ctx context.Context, q dbgen.Querier, reqID uuid.UUID) (engine.Requirement, error) { 338 r, err := q.GetRequirement(ctx, reqID) 339 if err != nil { 340 return engine.Requirement{}, fmt.Errorf("get requirement: %w", err) 341 } 342 return reqFromRow(ctx, q, r) 343} 344 345func reqFromRow(ctx context.Context, q dbgen.Querier, r dbgen.Requirement) (engine.Requirement, error) { 346 rs, err := q.ListRequiredSkills(ctx, r.ID) 347 if err != nil { 348 return engine.Requirement{}, fmt.Errorf("list required skills: %w", err) 349 } 350 rc, err := q.ListRequiredCertifications(ctx, r.ID) 351 if err != nil { 352 return engine.Requirement{}, fmt.Errorf("list required certifications: %w", err) 353 } 354 return ToEngineRequirement(r, rs, rc) 355} 356 357// loadCandidateByID loads a candidate, its sub-lists, and the engine input in one place. It returns 358// the DB row (for clearance dates), the engine candidate, and the customer agencies. 359func loadCandidateByID(ctx context.Context, q dbgen.Querier, id uuid.UUID, asOf time.Time) (dbgen.Candidate, engine.Candidate, []string, error) { 360 c, err := q.GetCandidate(ctx, id) 361 if err != nil { 362 return dbgen.Candidate{}, engine.Candidate{}, nil, fmt.Errorf("get candidate: %w", err) 363 } 364 skills, err := q.ListCandidateSkills(ctx, id) 365 if err != nil { 366 return dbgen.Candidate{}, engine.Candidate{}, nil, fmt.Errorf("list candidate skills: %w", err) 367 } 368 certs, err := q.ListCandidateCertifications(ctx, id) 369 if err != nil { 370 return dbgen.Candidate{}, engine.Candidate{}, nil, fmt.Errorf("list candidate certifications: %w", err) 371 } 372 cust, err := q.ListCandidateCustomerExperience(ctx, id) 373 if err != nil { 374 return dbgen.Candidate{}, engine.Candidate{}, nil, fmt.Errorf("list candidate customer experience: %w", err) 375 } 376 ec, err := ToEngineCandidate(c, skills, certs, cust, asOf) 377 if err != nil { 378 return dbgen.Candidate{}, engine.Candidate{}, nil, err 379 } 380 return c, ec, cust, nil 381} 382 383// Screen returns the full screenMATCH™ intelligence view for one candidate vs one requirement, 384// the explainable score plus the Mission Readiness panels. Read only. 385func (s *Service) Screen(ctx context.Context, candidateID, requirementID uuid.UUID) (ScreenView, error) { 386 if s.db == nil { 387 return ScreenView{}, ErrNoDatabase 388 } 389 asOf := s.clock() 390 var view ScreenView 391 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 392 q := dbgen.New(tx) 393 er, err := loadRequirement(ctx, q, requirementID) 394 if err != nil { 395 return err 396 } 397 c, ec, cust, err := loadCandidateByID(ctx, q, candidateID, asOf) 398 if err != nil { 399 return err 400 } 401 res := engine.ScoreCandidate(ctx, ec, er, s.matcher) 402 view = assembleScreenView(ec, er, res, ClearanceFactsFrom(c, cust), asOf) 403 return s.auditOnTx(ctx, tx, servedScoreEvent("screenMATCH.screen", candidateID, res)) 404 }) 405 return view, err 406} 407 408// RankCandidates scores EVERY candidate in the org against one requirement and returns them ranked 409// (Mission Readiness descending), the Candidate Screening Queue (Screen 1). Read-only: no slate is 410// persisted (use GenerateSlate to freeze a slate). 411func (s *Service) RankCandidates(ctx context.Context, requirementID uuid.UUID) ([]RankedCandidateView, error) { 412 if s.db == nil { 413 return nil, ErrNoDatabase 414 } 415 asOf := s.clock() 416 var out []RankedCandidateView 417 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 418 q := dbgen.New(tx) 419 er, err := loadRequirement(ctx, q, requirementID) 420 if err != nil { 421 return err 422 } 423 cands, err := q.ListCandidates(ctx, dbgen.ListCandidatesParams{Limit: 500, Offset: 0}) 424 if err != nil { 425 return fmt.Errorf("list candidates: %w", err) 426 } 427 type scored struct { 428 view RankedCandidateView 429 total decimal.Decimal 430 } 431 ranked := make([]scored, 0, len(cands)) 432 for _, c := range cands { 433 ec, err := loadCandidate(ctx, q, c, asOf) 434 if err != nil { 435 return err 436 } 437 res := engine.ScoreCandidate(ctx, ec, er, s.matcher) 438 // The Candidate Screening Queue scores everyone for a requirement, a screening stage AEDT 439 // output. Record each as a served score so the screening stage adverse impact denominator is 440 // complete, not biased toward the focused single candidate screens. 441 if e := s.auditOnTx(ctx, tx, servedScoreEvent("screenMATCH.rank", c.ID, res)); e != nil { 442 return e 443 } 444 ranked = append(ranked, scored{view: rankedCandidateView(c, res), total: res.Total}) 445 } 446 sort.SliceStable(ranked, func(i, j int) bool { 447 if d := ranked[i].total.Cmp(ranked[j].total); d != 0 { 448 return d > 0 449 } 450 return ranked[i].view.FullName < ranked[j].view.FullName 451 }) 452 out = make([]RankedCandidateView, len(ranked)) 453 for i, sc := range ranked { 454 out[i] = sc.view 455 } 456 return nil 457 }) 458 return out, err 459} 460 461// OpportunitiesForCandidate scores one candidate against every open requirement in the org and 462// returns the best fits, ranked, the Opportunity Fit Engine™ ("best opportunities for this person"). 463func (s *Service) OpportunitiesForCandidate(ctx context.Context, candidateID uuid.UUID, limit int) ([]OpportunityFitView, error) { 464 if s.db == nil { 465 return nil, ErrNoDatabase 466 } 467 if limit <= 0 || limit > 100 { 468 limit = 10 469 } 470 asOf := s.clock() 471 var out []OpportunityFitView 472 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 473 q := dbgen.New(tx) 474 _, ec, _, err := loadCandidateByID(ctx, q, candidateID, asOf) 475 if err != nil { 476 return err 477 } 478 out, err = s.rankOpenOpportunities(ctx, tx, ec, limit, func(res engine.MatchResult) error { 479 // Scoring this person against each opportunity is a screening stage AEDT output; record each 480 // served score so the screening stage analysis is complete. 481 return s.auditOnTx(ctx, tx, servedScoreEvent("screenMATCH.opportunities", candidateID, res)) 482 }) 483 return err 484 }) 485 return out, err 486} 487 488// rankOpenOpportunities scores a ready engine.Candidate against every OPEN requirement, calls auditFn 489// for each served score (so the saved path and the ephemeral Mission Check path each record under their 490// own AEDT action), ranks total descending with the requirement id as the stable tiebreak, and 491// truncates to limit. It is the single ranking path shared by OpportunitiesForCandidate and Mission 492// Check Mode A, so an unsaved resume and a saved candidate with identical fields rank identically. 493func (s *Service) rankOpenOpportunities(ctx context.Context, tx pgx.Tx, ec engine.Candidate, limit int, auditFn func(res engine.MatchResult) error) ([]OpportunityFitView, error) { 494 q := dbgen.New(tx) 495 reqs, err := q.ListOpenRequirements(ctx, dbgen.ListOpenRequirementsParams{Limit: 500, Offset: 0}) 496 if err != nil { 497 return nil, fmt.Errorf("list open requirements: %w", err) 498 } 499 500 type scored struct { 501 view OpportunityFitView 502 total decimal.Decimal 503 } 504 ranked := make([]scored, 0, len(reqs)) 505 for _, rr := range reqs { 506 er, err := reqFromRow(ctx, q, rr) 507 if err != nil { 508 return nil, err 509 } 510 res := engine.ScoreCandidate(ctx, ec, er, s.matcher) 511 if err := auditFn(res); err != nil { 512 return nil, err 513 } 514 ranked = append(ranked, scored{ 515 view: OpportunityFitView{ 516 RequirementID: rr.ID, Title: rr.Title, CustomerAgency: deref(rr.CustomerAgency), 517 Score: res.Total.StringFixed(2), Band: res.Band.Code(), 518 Status: intelligence.MissionStatusLabel(res.Band), 519 }, 520 total: res.Total, 521 }) 522 } 523 sort.SliceStable(ranked, func(i, j int) bool { 524 if d := ranked[i].total.Cmp(ranked[j].total); d != 0 { 525 return d > 0 526 } 527 return ranked[i].view.RequirementID.String() < ranked[j].view.RequirementID.String() 528 }) 529 if len(ranked) > limit { 530 ranked = ranked[:limit] 531 } 532 out := make([]OpportunityFitView, len(ranked)) 533 for i, sc := range ranked { 534 out[i] = sc.view 535 } 536 return out, nil 537} 538 539func loadCandidate(ctx context.Context, q dbgen.Querier, c dbgen.Candidate, asOf time.Time) (engine.Candidate, error) { 540 skills, err := q.ListCandidateSkills(ctx, c.ID) 541 if err != nil { 542 return engine.Candidate{}, fmt.Errorf("list candidate skills: %w", err) 543 } 544 certs, err := q.ListCandidateCertifications(ctx, c.ID) 545 if err != nil { 546 return engine.Candidate{}, fmt.Errorf("list candidate certifications: %w", err) 547 } 548 cust, err := q.ListCandidateCustomerExperience(ctx, c.ID) 549 if err != nil { 550 return engine.Candidate{}, fmt.Errorf("list candidate customer experience: %w", err) 551 } 552 return ToEngineCandidate(c, skills, certs, cust, asOf) 553} 554