Janady13/missionmatch / internal/modules/recruiting/pipeline_service.go

1package recruiting 2 3import ( 4 "context" 5 "encoding/json" 6 "errors" 7 "strings" 8 "time" 9 10 "github.com/google/uuid" 11 "github.com/jackc/pgx/v5" 12 "github.com/jackc/pgx/v5/pgconn" 13 "github.com/shopspring/decimal" 14 15 "github.com/missionworkforce/missionmatch/internal/platform/audit" 16 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 17) 18 19const appSelect = ` 20 a.id::text, a.requisition_id::text, r.title, cl.name, 21 a.candidate_id::text, c.full_name, c.clearance_level, c.available, 22 a.stage, a.furthest_stage_ord, a.disposition_code, a.disposition_reason, 23 a.owner_user_id::text, u.full_name, 24 a.source_first_touch, a.source_last_touch, 25 a.stage_changed_at, a.created_at, a.updated_at` 26 27const appFrom = ` 28 FROM applications a 29 JOIN candidates c ON c.organization_id = a.organization_id AND c.id = a.candidate_id 30 JOIN requisitions r ON r.organization_id = a.organization_id AND r.id = a.requisition_id 31 LEFT JOIN clients cl ON cl.organization_id = r.organization_id AND cl.id = r.client_id 32 LEFT JOIN users u ON u.organization_id = a.organization_id AND u.id = a.owner_user_id` 33 34func scanApplication(row scannable, now time.Time) (Application, error) { 35 var a Application 36 var reqTitle, clientName, ownerID, ownerName, dispCode, dispReason *string 37 var furthest int 38 var changed, created, updated time.Time 39 40 if err := row.Scan( 41 &a.ID, &a.RequisitionID, &reqTitle, &clientName, 42 &a.CandidateID, &a.CandidateName, &a.ClearanceLevel, &a.Available, 43 &a.Stage, &furthest, &dispCode, &dispReason, 44 &ownerID, &ownerName, 45 &a.SourceFirstTouch, &a.SourceLastTouch, 46 &changed, &created, &updated, 47 ); err != nil { 48 return Application{}, err 49 } 50 a.RequisitionTitle, a.ClientName = str(reqTitle), str(clientName) 51 a.OwnerUserID, a.OwnerName = str(ownerID), str(ownerName) 52 a.DispositionCode, a.DispositionReason = str(dispCode), str(dispReason) 53 a.StageLabel = stageLabels[a.Stage] 54 if furthest >= 1 && furthest <= len(ladder) { 55 a.FurthestStage = ladder[furthest-1] 56 a.FurthestStageLabel = stageLabels[a.FurthestStage] 57 } 58 a.StageChangedAt, a.CreatedAt, a.UpdatedAt = ts(changed), ts(created), ts(updated) 59 a.DaysInStage = days(stageAge(changed, now)) 60 return a, nil 61} 62 63// Submit adds a candidate to a job order's pipeline. 64// 65// The unique constraint on (organization_id, requisition_id, candidate_id) is the authority on 66// duplicates, not a pre check. Two recruiters submitting the same person to the same role at the same 67// moment is exactly the case a SELECT then INSERT would get wrong, and the constraint cannot. 68func (s *Service) Submit(ctx context.Context, reqID uuid.UUID, in CreateApplicationInput) (Application, error) { 69 if s.db == nil { 70 return Application{}, ErrNoDatabase 71 } 72 candID, err := uuid.Parse(strings.TrimSpace(in.CandidateID)) 73 if err != nil { 74 return Application{}, &ValidationError{Field: "candidate_id", Message: "must be a candidate id"} 75 } 76 stage := strings.TrimSpace(in.Stage) 77 if stage == "" { 78 stage = stageSourced 79 } 80 // An application cannot BEGIN in a terminal state. A rejection with no preceding consideration is not 81 // a pipeline record, and it would enter the adverse impact denominator without ever being a real 82 // candidacy. 83 if !isStage(stage) || isTerminalStage(stage) { 84 return Application{}, &ValidationError{Field: "stage", Message: "a submission starts on the ladder, not at an outcome"} 85 } 86 reason, err := clean("reason", in.Reason, maxReqNote) 87 if err != nil { 88 return Application{}, err 89 } 90 touch, err := clean("source_touch", in.SourceTouch, 120) 91 if err != nil { 92 return Application{}, err 93 } 94 actor := actorPtr(ctx) 95 if actor == nil { 96 return Application{}, tenant.ErrNoTenant 97 } 98 99 // Scanned as a uuid.UUID rather than as text. The id comes from the database, so parsing it back 100 // could only ever fail on something impossible, but uuid.MustParse turns that impossible case into a 101 // panic in a request path, which CLAUDE.md forbids. Two of those panics sat AFTER the commit, where a 102 // panic would destroy the response for work that is already durably written and cannot be retried 103 // safely. Scanning the native type removes the parse, and with it the question. 104 var newID uuid.UUID 105 err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 106 // The job order must be live. Submitting into a cancelled or closed requisition records work 107 // against a role that no longer exists, and it would inflate the funnel of a dead order. 108 var status string 109 if e := tx.QueryRow(ctx, 110 `SELECT status FROM requisitions WHERE id = $1 AND deleted_at IS NULL`, reqID).Scan(&status); e != nil { 111 if errors.Is(e, pgx.ErrNoRows) { 112 return ErrNotFound 113 } 114 return translatePG(e) 115 } 116 if status == reqClosed || status == reqCancelled { 117 return &ValidationError{Field: "requisition_id", 118 Message: "this job order is " + status + ", so candidates cannot be submitted to it"} 119 } 120 121 // Insert at the default stage, then move it with an event if a later start was asked for. The 122 // stage column is writable only through the event log, so this is the only correct order. 123 e := tx.QueryRow(ctx, ` 124 INSERT INTO applications 125 (organization_id, requisition_id, candidate_id, owner_user_id, created_by, 126 source_first_touch, source_last_touch) 127 VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1, $2, $3, $3, $4, $4) 128 RETURNING id`, reqID, candID, actor, touch).Scan(&newID) 129 if e != nil { 130 return translatePG(e) 131 } 132 if err := insertStageEvent(ctx, tx, newID, stageSourced, stageSourced, "", reason, actor, true); err != nil { 133 return err 134 } 135 aid := newID 136 if stage != stageSourced { 137 if err := insertStageEvent(ctx, tx, newID, stageSourced, stage, "", reason, actor, false); err != nil { 138 return err 139 } 140 } 141 // Captured, not logged. The flag is raised inside captureActivity and cleared immediately, and 142 // this is unreachable from HTTP, so a human cannot produce a row that claims to be captured. 143 if e := captureActivity(ctx, tx, activityRow{ 144 actor: actor, kind: "submission", direction: "internal", outcome: "logged", 145 candidateID: &candID, applicationID: &aid, requisitionID: &reqID, 146 body: "Added to the pipeline at " + stage, 147 }); e != nil { 148 return e 149 } 150 nv, _ := json.Marshal(map[string]string{"application_id": newID.String(), "candidate_id": candID.String(), "stage": stage}) 151 return s.audit.OnTx(ctx, tx, audit.Event{ 152 Action: "application.created", ActorUserID: actor, 153 EntityType: "application", EntityID: newID.String(), NewValues: nv, 154 }) 155 }) 156 if err != nil { 157 return Application{}, err 158 } 159 return s.applicationRow(ctx, newID) 160} 161 162// MoveStage advances, reverses or disposes an application. 163// 164// Almost every rule here is enforced by the database, deliberately. from_stage must match reality, the 165// event cannot predate the current stage, the high water mark cannot fall, a terminal stage needs a 166// coded reason, and placements cannot exceed the openings. Each of those is a race if it is checked in 167// application code and a fact if it is checked under the row lock the trigger holds, so this method 168// validates only what is purely a matter of vocabulary and lets the database decide the rest. 169func (s *Service) MoveStage(ctx context.Context, id uuid.UUID, in MoveStageInput) (Application, error) { 170 if s.db == nil { 171 return Application{}, ErrNoDatabase 172 } 173 to := strings.TrimSpace(in.ToStage) 174 if !isStage(to) { 175 return Application{}, &ValidationError{Field: "to_stage", Message: "is not a permitted stage"} 176 } 177 expected := strings.TrimSpace(in.ExpectedStage) 178 // Required, not optional. It is the concurrency token: the database refuses the write unless the 179 // application is still where the caller last saw it, which is what stops two recruiters moving the 180 // same person from different screens. 181 if !isStage(expected) { 182 return Application{}, &ValidationError{Field: "expected_stage", Message: "is required and must be the stage you last saw"} 183 } 184 if expected == to { 185 return Application{}, &ValidationError{Field: "to_stage", Message: "the application is already at that stage"} 186 } 187 reason, err := clean("reason", in.Reason, maxReqNote) 188 if err != nil { 189 return Application{}, err 190 } 191 code := strings.TrimSpace(in.DispositionCode) 192 if isTerminalStage(to) { 193 if code == "" { 194 return Application{}, &ValidationError{Field: "disposition_code", 195 Message: "say why this candidate left the process"} 196 } 197 if err := validDisposition(code, to); err != nil { 198 return Application{}, err 199 } 200 } else if code != "" { 201 return Application{}, &ValidationError{Field: "disposition_code", 202 Message: "a reason code belongs only to a rejection or a withdrawal"} 203 } 204 205 actor := actorPtr(ctx) 206 err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 207 if err := insertStageEvent(ctx, tx, id, expected, to, code, reason, actor, false); err != nil { 208 return err 209 } 210 nv, _ := json.Marshal(map[string]string{"to": to, "code": code, "reason": reason}) 211 return s.audit.OnTx(ctx, tx, audit.Event{ 212 Action: "application.stage_changed", ActorUserID: actor, 213 EntityType: "application", EntityID: id.String(), NewValues: nv, 214 }) 215 }) 216 if err != nil { 217 return Application{}, err 218 } 219 return s.applicationRow(ctx, id) 220} 221 222// insertStageEvent writes the event. The trigger validates it, moves the row and measures the duration, 223// so this inserts and interprets the failure rather than duplicating any of that logic. 224// 225// first marks the opening event, whose from_stage is the same as its to_stage. The table's own CHECK 226// forbids from_stage = to_stage, so the opening event is written with a NULL predecessor and the trigger 227// fills it in. 228func insertStageEvent(ctx context.Context, tx pgx.Tx, id uuid.UUID, from, to, code, reason string, actor *uuid.UUID, first bool) error { 229 var fromArg any 230 if from != "" && !first { 231 fromArg = from 232 } 233 var codeArg any 234 if code != "" { 235 codeArg = code 236 } 237 _, err := tx.Exec(ctx, ` 238 INSERT INTO application_stage_events 239 (organization_id, application_id, from_stage, to_stage, disposition_code, reason, 240 actor_user_id, actor_name) 241 VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1, $2, $3, $4, $5, $6, 242 COALESCE((SELECT full_name FROM users WHERE id = $6), ''))`, 243 id, fromArg, to, codeArg, reason, actor) 244 if err != nil { 245 return translateStageErr(err) 246 } 247 return nil 248} 249 250// translateStageErr maps the trigger's raised conditions onto typed domain errors. The trigger is the 251// only place these rules can be checked without a race, so its messages are load bearing and this 252// mapping is what turns them into a status the caller can act on. 253func translateStageErr(err error) error { 254 var pg *pgconn.PgError 255 if !errors.As(err, &pg) { 256 return translatePG(err) 257 } 258 switch { 259 case pg.Code == "40001": // the trigger's stage conflict 260 return ErrConflict 261 case pg.Code == "22007": // an out of order event 262 return &ValidationError{Field: "occurred_at", Message: "that change is older than the current stage"} 263 // The live placement guard from 0055 fires on a stage move, so its message has to be understood 264 // here as well as in translateAssignmentErr. Without this case the trigger's instruction, which 265 // tells the user exactly what to do, was replaced by "failed a database constraint". 266 case pg.Code == "23514" && strings.Contains(pg.Message, "live placement"): 267 return &ValidationError{Field: "to_stage", Message: pg.Message} 268 case pg.Code == "23514" && strings.Contains(pg.Message, "opening"): 269 return &ValidationError{Field: "to_stage", Message: pg.Message} 270 case pg.Code == "23514" && strings.Contains(pg.Message, "reason code"): 271 return &ValidationError{Field: "disposition_code", Message: "say why this candidate left the process"} 272 case pg.Code == "23505": 273 return &DuplicateError{Field: "candidate_id"} 274 case pg.Code == "42501": 275 // The stage guard fired, which means something tried to write stage outside the event log. That 276 // is a defect in this service, not bad input. 277 return err 278 } 279 return translatePG(err) 280} 281 282// Board assembles one requisition's pipeline. 283// 284// The funnel is computed from high water marks in Go rather than by a GROUP BY on the current stage, 285// because only the high water mark yields a monotone series. A GROUP BY on stage answers "who is here 286// now", which is the board's columns, and those two questions have different answers on purpose. 287func (s *Service) Board(ctx context.Context, reqID uuid.UUID) (PipelineBoard, error) { 288 if s.db == nil { 289 return PipelineBoard{}, ErrNoDatabase 290 } 291 board := PipelineBoard{ 292 Columns: []StageColumn{}, Funnel: []FunnelStage{}, 293 Conversion: []Conversion{}, Dispositions: []DispositionBreakdown{}, 294 } 295 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 296 if e := tx.QueryRow(ctx, 297 `SELECT id::text, title, openings FROM requisitions WHERE id = $1 AND deleted_at IS NULL`, reqID). 298 Scan(&board.RequisitionID, &board.RequisitionTitle, &board.Openings); e != nil { 299 if errors.Is(e, pgx.ErrNoRows) { 300 return ErrNotFound 301 } 302 return translatePG(e) 303 } 304 305 now := time.Now().UTC() 306 rows, e := tx.Query(ctx, `SELECT`+appSelect+appFrom+` 307 WHERE a.requisition_id = $1 308 ORDER BY a.furthest_stage_ord DESC, a.stage_changed_at ASC, a.id ASC`, reqID) 309 if e != nil { 310 return translatePG(e) 311 } 312 defer rows.Close() 313 314 byStage := map[string][]Application{} 315 var marks []int 316 dispCount := map[string]DispositionBreakdown{} 317 for rows.Next() { 318 a, e := scanApplication(rows, now) 319 if e != nil { 320 return translatePG(e) 321 } 322 board.Total++ 323 if !isTerminalStage(a.Stage) { 324 board.Active++ 325 } 326 if a.Stage == stagePlaced { 327 board.Placed++ 328 } 329 marks = append(marks, stageOrd[a.FurthestStage]) 330 byStage[a.Stage] = append(byStage[a.Stage], a) 331 if a.DispositionCode != "" { 332 d := dispCount[a.DispositionCode] 333 d.Code, d.Stage, d.Count = a.DispositionCode, a.Stage, d.Count+1 334 dispCount[a.DispositionCode] = d 335 } 336 } 337 if e := rows.Err(); e != nil { 338 return e 339 } 340 341 // Every stage gets a column, including the empty ones, so the board does not change shape as 342 // candidates move and a recruiter can always drop someone into the next stage. 343 for _, st := range append(append([]string{}, ladder...), stageRejected, stageWithdrawn) { 344 apps := byStage[st] 345 if apps == nil { 346 apps = []Application{} 347 } 348 board.Columns = append(board.Columns, StageColumn{ 349 Stage: st, Label: stageLabels[st], Count: len(apps), Applications: apps, 350 }) 351 } 352 board.Funnel = funnelFrom(marks, board.Total) 353 board.Conversion = conversionFrom(board.Funnel) 354 for _, d := range dispCount { 355 d.Pct = pctOf(d.Count, board.Total) 356 board.Dispositions = append(board.Dispositions, d) 357 } 358 board.SeatsLeft = board.Openings - board.Placed 359 if board.SeatsLeft < 0 { 360 // Unreachable: the trigger refuses a placement past the last seat. Clamped rather than shown 361 // negative because a negative vacancy is never a real quantity. 362 board.SeatsLeft = 0 363 } 364 return nil 365 }) 366 if err != nil { 367 return PipelineBoard{}, err 368 } 369 return board, nil 370} 371 372// GetApplication returns one application with its full history. 373func (s *Service) GetApplication(ctx context.Context, id uuid.UUID) (ApplicationDetail, error) { 374 if s.db == nil { 375 return ApplicationDetail{}, ErrNoDatabase 376 } 377 out := ApplicationDetail{Events: []StageEvent{}} 378 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 379 now := time.Now().UTC() 380 a, e := scanApplication(tx.QueryRow(ctx, `SELECT`+appSelect+appFrom+` WHERE a.id = $1`, id), now) 381 if e != nil { 382 if errors.Is(e, pgx.ErrNoRows) { 383 return ErrNotFound 384 } 385 return translatePG(e) 386 } 387 out.Application = a 388 389 rows, e := tx.Query(ctx, ` 390 SELECT e.id::text, e.from_stage, e.to_stage, e.seconds_in_from::text, e.reason, 391 e.disposition_code, e.actor_user_id::text, NULLIF(e.actor_name, ''), e.occurred_at 392 FROM application_stage_events e 393 WHERE e.application_id = $1 394 ORDER BY e.occurred_at ASC, e.id ASC`, id) 395 if e != nil { 396 return translatePG(e) 397 } 398 defer rows.Close() 399 for rows.Next() { 400 var ev StageEvent 401 var from, secs, code, actor, actorName *string 402 var at time.Time 403 if e := rows.Scan(&ev.ID, &from, &ev.ToStage, &secs, &ev.Reason, &code, &actor, &actorName, &at); e != nil { 404 return translatePG(e) 405 } 406 ev.FromStage, ev.DispositionCode = str(from), str(code) 407 ev.ActorUserID, ev.ActorName = str(actor), str(actorName) 408 ev.FromStageLabel, ev.ToStageLabel = stageLabels[ev.FromStage], stageLabels[ev.ToStage] 409 ev.OccurredAt = ts(at) 410 if secs != nil { 411 if d, err := decimal.NewFromString(*secs); err == nil { 412 ev.DaysInFrom = days(d) 413 } 414 } 415 out.Events = append(out.Events, ev) 416 } 417 return rows.Err() 418 }) 419 if err != nil { 420 return ApplicationDetail{}, err 421 } 422 return out, nil 423} 424 425// ListApplications returns a candidate's applications across every job order, or a stage filtered list. 426func (s *Service) ListApplications(ctx context.Context, candidateID, stage string, limit, offset int) (ApplicationList, error) { 427 if s.db == nil { 428 return ApplicationList{}, ErrNoDatabase 429 } 430 if limit <= 0 || limit > 200 { 431 limit = 50 432 } 433 if offset < 0 { 434 offset = 0 435 } 436 where := []string{"true"} 437 var args []any 438 if candidateID != "" { 439 id, err := uuid.Parse(candidateID) 440 if err != nil { 441 return ApplicationList{}, &ValidationError{Field: "candidate_id", Message: "must be a candidate id"} 442 } 443 args = append(args, id) 444 where = append(where, "a.candidate_id = $"+itoa(len(args))) 445 } 446 switch stage { 447 case "": 448 case "active": 449 where = append(where, "a.stage NOT IN ('rejected','withdrawn')") 450 default: 451 if !isStage(stage) { 452 return ApplicationList{}, &ValidationError{Field: "stage", Message: "is not a permitted stage"} 453 } 454 args = append(args, stage) 455 where = append(where, "a.stage = $"+itoa(len(args))) 456 } 457 cond := strings.Join(where, " AND ") 458 459 out := ApplicationList{Applications: []Application{}, Limit: limit, Offset: offset} 460 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 461 if e := tx.QueryRow(ctx, `SELECT count(*)`+appFrom+` WHERE `+cond, args...).Scan(&out.Total); e != nil { 462 return translatePG(e) 463 } 464 rows, e := tx.Query(ctx, `SELECT`+appSelect+appFrom+` WHERE `+cond+` 465 ORDER BY a.stage_changed_at DESC, a.id ASC 466 LIMIT $`+itoa(len(args)+1)+` OFFSET $`+itoa(len(args)+2), append(args, limit, offset)...) 467 if e != nil { 468 return translatePG(e) 469 } 470 defer rows.Close() 471 now := time.Now().UTC() 472 for rows.Next() { 473 a, e := scanApplication(rows, now) 474 if e != nil { 475 return translatePG(e) 476 } 477 out.Applications = append(out.Applications, a) 478 } 479 return rows.Err() 480 }) 481 if err != nil { 482 return ApplicationList{}, err 483 } 484 return out, nil 485} 486 487func (s *Service) applicationRow(ctx context.Context, id uuid.UUID) (Application, error) { 488 var out Application 489 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 490 a, e := scanApplication(tx.QueryRow(ctx, `SELECT`+appSelect+appFrom+` WHERE a.id = $1`, id), time.Now().UTC()) 491 if e != nil { 492 if errors.Is(e, pgx.ErrNoRows) { 493 return ErrNotFound 494 } 495 return translatePG(e) 496 } 497 out = a 498 return nil 499 }) 500 return out, err 501} 502