Janady13/missionmatch / internal/modules/recruitmatch/service_queries.go

1package recruitmatch 2 3import ( 4 "context" 5 "fmt" 6 7 "github.com/jackc/pgx/v5" 8) 9 10// buildTemps materializes the per organization RLS scoped working sets once per request. ON COMMIT DROP 11// cleans them up when the transaction ends, and the rm_ prefix keeps them from ever colliding with the 12// commandcenter cc_ temps if both ever run in one transaction. 13// 14// rm_cand carries the funnel membership as three booleans computed in one pass. That is what makes the 15// funnel monotonic BY CONSTRUCTION rather than by convention: approved_on_slate implies on_slate, and 16// on_slate implies the screened predicate, so every stage is a strict subset of the one before it and no 17// arrangement of data can make a later stage larger. Computing the stages as four independent queries 18// would lose that guarantee, because a candidate can sit on a slate with no candidate_evaluations row 19// (slate generation scores its own members), which would let submitted exceed screened. 20func buildTemps(ctx context.Context, tx pgx.Tx, asOf string) error { 21 // Defense in depth: confirm the tenant GUC is set before any aggregate runs, so a wrong pool or role 22 // fails closed here instead of silently aggregating another tenant. On a platform sold to many 23 // clients this check is the difference between a bug and a cross tenant disclosure. 24 var guc string 25 if err := tx.QueryRow(ctx, `SELECT COALESCE(current_setting('app.current_org_id', true),'')`).Scan(&guc); err != nil { 26 return fmt.Errorf("tenant guc check: %w", err) 27 } 28 if guc == "" { 29 return fmt.Errorf("recruitmatch: tenant GUC not set, refusing to aggregate") 30 } 31 32 // Live bench only. anonymized_at is excluded alongside deleted_at because an anonymized record is a 33 // GDPR erasure tombstone, not a person who can be staffed. 34 if _, err := tx.Exec(ctx, ` 35 CREATE TEMP TABLE rm_cand ON COMMIT DROP AS 36 SELECT c.id, 37 c.full_name, 38 c.clearance_level, 39 c.available, 40 EXISTS (SELECT 1 FROM candidate_evaluations e WHERE e.candidate_id = c.id) AS has_eval, 41 EXISTS (SELECT 1 FROM slate_members sm WHERE sm.candidate_id = c.id) AS on_slate, 42 EXISTS (SELECT 1 FROM slate_members sm 43 WHERE sm.candidate_id = c.id AND sm.decision = 'approved') AS approved_on_slate 44 FROM candidates c 45 WHERE c.deleted_at IS NULL AND c.anonymized_at IS NULL`); err != nil { 46 return fmt.Errorf("rm_cand: %w", err) 47 } 48 if _, err := tx.Exec(ctx, `CREATE INDEX ON rm_cand (clearance_level)`); err != nil { 49 return fmt.Errorf("rm_cand index: %w", err) 50 } 51 52 // Open requirements. A requirement with no opportunity is open by definition, matching the 53 // ListOpenRequirements contract the rest of the app already uses. 54 if _, err := tx.Exec(ctx, ` 55 CREATE TEMP TABLE rm_active_req ON COMMIT DROP AS 56 SELECT r.id, r.title, COALESCE(r.customer_agency,'') AS customer_agency, r.required_level 57 FROM requirements r 58 LEFT JOIN opportunities o ON o.id = r.opportunity_id 59 WHERE r.deleted_at IS NULL AND (o.id IS NULL OR o.status = 'open')`); err != nil { 60 return fmt.Errorf("rm_active_req: %w", err) 61 } 62 63 // Submissions, restricted to the live bench so a deleted candidate cannot inflate a requirement's 64 // submission count. Joined to slates because slate_members carries no requirement of its own. 65 if _, err := tx.Exec(ctx, ` 66 CREATE TEMP TABLE rm_sub ON COMMIT DROP AS 67 SELECT s.requirement_id, sm.candidate_id, sm.decision, sm.decided_by 68 FROM slate_members sm 69 JOIN slates s ON s.id = sm.slate_id 70 JOIN rm_cand c ON c.id = sm.candidate_id`); err != nil { 71 return fmt.Errorf("rm_sub: %w", err) 72 } 73 if _, err := tx.Exec(ctx, `CREATE INDEX ON rm_sub (requirement_id)`); err != nil { 74 return fmt.Errorf("rm_sub requirement index: %w", err) 75 } 76 if _, err := tx.Exec(ctx, `CREATE INDEX ON rm_sub (decided_by)`); err != nil { 77 return fmt.Errorf("rm_sub decider index: %w", err) 78 } 79 80 _ = asOf // asOf is reserved for the dated panels; the current set is point in time by construction. 81 return nil 82} 83 84// headline computes the whole funnel and the stat tiles in ONE pass over rm_cand. Because every stage is 85// a FILTER over the same rows, the returned counts are guaranteed non increasing. 86func headline(ctx context.Context, tx pgx.Tx) (Headline, error) { 87 var h Headline 88 err := tx.QueryRow(ctx, ` 89 SELECT count(*), 90 count(*) FILTER (WHERE has_eval OR on_slate), 91 count(*) FILTER (WHERE on_slate), 92 count(*) FILTER (WHERE approved_on_slate), 93 count(*) FILTER (WHERE available) 94 FROM rm_cand`). 95 Scan(&h.Identified, &h.Screened, &h.Submitted, &h.Approved, &h.OnBench) 96 if err != nil { 97 return h, fmt.Errorf("headline: %w", err) 98 } 99 if err := tx.QueryRow(ctx, `SELECT count(*) FROM rm_active_req`).Scan(&h.ActiveJobs); err != nil { 100 return h, fmt.Errorf("active jobs: %w", err) 101 } 102 return h, nil 103} 104 105// decisionCounts totals slate membership decisions. These count MEMBERSHIPS, not people, which is why 106// they are reported apart from the funnel. 107func decisionCounts(ctx context.Context, tx pgx.Tx) (approved, rejected, pending int64, err error) { 108 e := tx.QueryRow(ctx, ` 109 SELECT count(*) FILTER (WHERE decision = 'approved'), 110 count(*) FILTER (WHERE decision = 'rejected'), 111 count(*) FILTER (WHERE decision = 'pending') 112 FROM rm_sub`).Scan(&approved, &rejected, &pending) 113 if e != nil { 114 return 0, 0, 0, fmt.Errorf("decision counts: %w", e) 115 } 116 return approved, rejected, pending, nil 117} 118 119// bandMix is the stored band split of persisted evaluations. recruitMATCH reports the band the frozen 120// engine already assigned and never recomputes one. 121func bandMix(ctx context.Context, tx pgx.Tx) (BandMix, error) { 122 var b BandMix 123 err := tx.QueryRow(ctx, ` 124 SELECT count(*) FILTER (WHERE band = 'green'), 125 count(*) FILTER (WHERE band = 'yellow'), 126 count(*) FILTER (WHERE band = 'red'), 127 count(*) 128 FROM candidate_evaluations`).Scan(&b.Green, &b.Yellow, &b.Red, &b.Total) 129 if err != nil { 130 return b, fmt.Errorf("band mix: %w", err) 131 } 132 return b, nil 133} 134 135// clearanceRows returns the live bench split by clearance level. It drives from clearance_levels so every 136// level appears even at zero, which keeps the chart shape stable as an organization fills up, and orders 137// by the reference ladder so the axis reads least to most cleared regardless of what the org holds. 138func clearanceRows(ctx context.Context, tx pgx.Tx) ([]ClearanceRow, error) { 139 rows, err := tx.Query(ctx, ` 140 SELECT cl.code, cl.label, count(c.id) 141 FROM clearance_levels cl 142 LEFT JOIN rm_cand c ON c.clearance_level = cl.code 143 GROUP BY cl.code, cl.label, cl.rank 144 ORDER BY cl.rank, cl.code`) 145 if err != nil { 146 return nil, fmt.Errorf("clearance rows: %w", err) 147 } 148 defer rows.Close() 149 150 out := []ClearanceRow{} 151 var total int64 152 for rows.Next() { 153 var r ClearanceRow 154 if err := rows.Scan(&r.Code, &r.Label, &r.Count); err != nil { 155 return nil, fmt.Errorf("clearance scan: %w", err) 156 } 157 total += r.Count 158 out = append(out, r) 159 } 160 if err := rows.Err(); err != nil { 161 return nil, fmt.Errorf("clearance iterate: %w", err) 162 } 163 // Percentages are computed after the full total is known, so they are shares of the bench rather than 164 // of a running subtotal. Rounded shares are not guaranteed to sum to exactly 100.00; that is inherent 165 // to rounding and is not corrected here, because inventing a residual to force the sum would misstate 166 // a category count. 167 for i := range out { 168 out[i].Pct = pct(out[i].Count, total).StringFixed(2) 169 } 170 return out, nil 171} 172 173// topJobs ranks open requirements by DISTINCT candidates submitted. One set based query with no per row 174// lookup. The ordering carries a full tiebreak chain ending in the id so the result is byte identical 175// across runs for identical data. 176func topJobs(ctx context.Context, tx pgx.Tx, limit int) ([]TopJob, error) { 177 rows, err := tx.Query(ctx, ` 178 SELECT r.id::text, r.title, r.customer_agency, r.required_level, 179 count(DISTINCT s.candidate_id) AS submitted 180 FROM rm_active_req r 181 LEFT JOIN rm_sub s ON s.requirement_id = r.id 182 GROUP BY r.id, r.title, r.customer_agency, r.required_level 183 ORDER BY count(DISTINCT s.candidate_id) DESC, r.title ASC, r.id ASC 184 LIMIT $1`, limit) 185 if err != nil { 186 return nil, fmt.Errorf("top jobs: %w", err) 187 } 188 defer rows.Close() 189 190 out := []TopJob{} 191 for rows.Next() { 192 var j TopJob 193 if err := rows.Scan(&j.RequirementID, &j.Title, &j.CustomerAgency, &j.RequiredLevel, &j.Submitted); err != nil { 194 return nil, fmt.Errorf("top jobs scan: %w", err) 195 } 196 out = append(out, j) 197 } 198 if err := rows.Err(); err != nil { 199 return nil, fmt.Errorf("top jobs iterate: %w", err) 200 } 201 return out, nil 202} 203 204// recruiters attributes screening and slate decisions to active users in this organization. users is RLS 205// scoped, so this cannot see another tenant's people. It works unchanged for a one recruiter shop and a 206// twenty recruiter shop. 207func recruiters(ctx context.Context, tx pgx.Tx) ([]RecruiterRow, error) { 208 rows, err := tx.Query(ctx, ` 209 SELECT u.id::text, u.full_name, 210 (SELECT count(*) FROM candidate_evaluations e WHERE e.evaluated_by = u.id), 211 (SELECT count(*) FROM rm_sub s WHERE s.decided_by = u.id), 212 (SELECT count(*) FROM rm_sub s WHERE s.decided_by = u.id AND s.decision = 'approved') 213 FROM users u 214 WHERE u.status = 'active' 215 ORDER BY u.full_name ASC, u.id ASC`) 216 if err != nil { 217 return nil, fmt.Errorf("recruiters: %w", err) 218 } 219 defer rows.Close() 220 221 out := []RecruiterRow{} 222 for rows.Next() { 223 var r RecruiterRow 224 if err := rows.Scan(&r.UserID, &r.FullName, &r.Screened, &r.Decided, &r.Approved); err != nil { 225 return nil, fmt.Errorf("recruiters scan: %w", err) 226 } 227 out = append(out, r) 228 } 229 if err := rows.Err(); err != nil { 230 return nil, fmt.Errorf("recruiters iterate: %w", err) 231 } 232 return out, nil 233} 234