Janady13/missionmatch / internal/modules/staffing/portfolio.go

1package staffing 2 3import ( 4 "context" 5 "sort" 6 7 "github.com/google/uuid" 8 "github.com/jackc/pgx/v5" 9 "github.com/shopspring/decimal" 10 11 "github.com/missionworkforce/missionmatch/internal/modules/matching" 12 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 13 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 14) 15 16// GeneratePortfolio allocates the available bench across the open requirements to maximize total value 17// (POST /recommendations/portfolio). The available bench excludes candidates already committed elsewhere; 18// each requirement's open slots are its quantity minus its current commitments. The optimizer arbitrates 19// only among engine qualified, available candidates, and each assignment carries the value matrix 20// counterfactual. Two portfolio runs for the same org serialize on a Postgres advisory lock, so a fresh 21// run never interleaves its supersede and insert with another. An empty eligible bench or no open 22// requirement returns an empty allocation (200), not an error; an over large problem returns 23// ErrProblemTooLarge and a solve over budget returns ErrSolveBudget. 24func (s *Service) GeneratePortfolio(ctx context.Context, req PortfolioRequest) (RunResponse, error) { 25 if s.db == nil { 26 return RunResponse{}, ErrNoDatabase 27 } 28 orgID, err := s.orgID(ctx) 29 if err != nil { 30 return RunResponse{}, err 31 } 32 floor, adj, verr := parseOverrides(req.StaffingFloor, req.AdjacencyThreshold) 33 if verr != nil { 34 return RunResponse{}, verr 35 } 36 cfg := s.cfg.WithOverrides(floor, adj) 37 asOf := s.clock() 38 39 var ( 40 pin PortfolioInput 41 resByPair = map[[2]int]engine.MatchResult{} 42 candIDs []string // candidate index -> uuid string (available bench, stable order) 43 candProv []string // candidate index -> clearance provenance 44 candComp []decimal.Decimal // candidate index -> profile completeness 45 reqIDs []string // requirement index -> uuid string 46 valueSource string 47 verr2 *ValidationError 48 ) 49 // Phase A (read): score the bench against each in scope open requirement and build the assignment graph. 50 err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 51 q := dbgen.New(tx) 52 score, e := s.scorerOnTx(ctx, tx) 53 if e != nil { 54 return e 55 } 56 openRows, e := q.ListOpenRequirementsWithValue(ctx, dbgen.ListOpenRequirementsWithValueParams{Limit: openRequirementCap, Offset: 0}) 57 if e != nil { 58 return e 59 } 60 priorities, vs := buildPriorities(cfg, openRows) 61 valueSource = vs 62 63 var subset map[string]bool 64 if len(req.RequirementIDs) > 0 { 65 known := map[string]bool{} 66 for _, r := range openRows { 67 known[r.ID.String()] = true 68 } 69 subset = map[string]bool{} 70 for _, idStr := range req.RequirementIDs { 71 if _, perr := uuid.Parse(idStr); perr != nil || !known[idStr] { 72 verr2 = validationErr("requirement_ids", "must all be open requirement ids in this tenant") 73 return nil // surfaced as 422 after the tx 74 } 75 subset[idStr] = true 76 } 77 } 78 79 commitRows, e := q.ListCommitments(ctx) 80 if e != nil { 81 return e 82 } 83 commit := buildCommitments(commitRows) 84 cands, e := q.ListCandidates(ctx, dbgen.ListCandidatesParams{Limit: benchScanLimit, Offset: 0}) 85 if e != nil { 86 return e 87 } 88 // Available bench = candidates not already committed anywhere. 89 var avail []dbgen.Candidate 90 for _, c := range cands { 91 if _, committed := commit.byCand[c.ID.String()]; committed { 92 continue 93 } 94 avail = append(avail, c) 95 } 96 candIDs = make([]string, len(avail)) 97 candProv = make([]string, len(avail)) 98 candComp = make([]decimal.Decimal, len(avail)) 99 ecs := make([]engine.Candidate, len(avail)) 100 for i, c := range avail { 101 ec, e := matching.LoadEngineCandidate(ctx, q, c, asOf) 102 if e != nil { 103 return e 104 } 105 candIDs[i] = c.ID.String() 106 candProv[i] = c.ClearanceProvenance 107 candComp[i] = completeness(ec) 108 ecs[i] = ec 109 } 110 pin.NumCandidates = len(avail) 111 112 for _, rr := range openRows { 113 if subset != nil && !subset[rr.ID.String()] { 114 continue 115 } 116 openSlots := int(rr.Quantity) - commit.countByReq[rr.ID.String()] 117 if openSlots <= 0 { 118 continue // already fully committed: no open slots to allocate 119 } 120 er, e := matching.LoadEngineRequirement(ctx, q, rr.ID) 121 if e != nil { 122 return e 123 } 124 reqIdx := len(reqIDs) 125 reqIDs = append(reqIDs, rr.ID.String()) 126 pin.Slots = append(pin.Slots, openSlots) 127 priority := priorities[rr.ID.String()] 128 for candIdx, ec := range ecs { 129 res := score(ctx, ec, er) 130 if !ec.Available || res.Total.LessThan(cfg.StaffingFloor) { 131 continue // the optimizer only sees engine qualified, available candidates 132 } 133 v := ScaledValue(res.Total, priority, cfg.ValueScalePrecision) 134 if v <= 0 { 135 continue 136 } 137 pin.Edges = append(pin.Edges, AssignEdge{CandidateIdx: candIdx, RequirementIdx: reqIdx, Value: v}) 138 resByPair[[2]int{candIdx, reqIdx}] = res 139 } 140 } 141 return nil 142 }) 143 if err != nil { 144 return RunResponse{}, err 145 } 146 if verr2 != nil { 147 return RunResponse{}, verr2 148 } 149 150 // Phase A.5 (no tx): solve, bounded by the size cap and the solve budget. 151 solveCtx, cancel := context.WithTimeout(ctx, cfg.SolveBudget) 152 defer cancel() 153 sol, err := SolvePortfolio(solveCtx, pin, cfg.MaxProblemEdges, cfg.MaxProblemNodes) 154 if err != nil { 155 return RunResponse{}, err // ErrProblemTooLarge or ErrSolveBudget, surfaced as 413 or 503 156 } 157 cf := Counterfactuals(pin, sol) 158 159 // Build the recommendations: an ASSIGN per optimal assignment, a SOURCE per requirement left short. 160 recs := s.buildPortfolioRecs(cfg, sol, pin, cf, resByPair, candIDs, candProv, candComp, reqIDs) 161 162 // The frozen result for each assigned pair, keyed for the eval upsert in persistRun. 163 assignedRes := map[string]engine.MatchResult{} 164 for _, a := range sol.Assignments { 165 assignedRes[pairKey(candIDs[a.CandidateIdx], reqIDs[a.RequirementIdx])] = resByPair[[2]int{a.CandidateIdx, a.RequirementIdx}] 166 } 167 168 // Phase B (no tx): narrate. 169 enriched := s.enrich(ctx, recs) 170 171 // Phase C (write): serialize per org on an advisory lock, then persist atomically. 172 var run dbgen.RecommendationRun 173 var rows []dbgen.StaffingRecommendation 174 err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 175 // Serialize portfolio runs for this org so a fresh run never interleaves its supersede and insert 176 // with another (the read committed snapshot would otherwise leave two live pending sets). 177 if e := acquireAdvisoryLock(ctx, tx, advisoryKey(ViewPortfolio, orgID, "")); e != nil { 178 return e 179 } 180 var e error 181 run, rows, e = s.persistRun(ctx, tx, orgID, ViewPortfolio, nil, runParams(cfg, valueSource), "staffing.portfolio", enriched, assignedRes, 182 func(ctx context.Context, q *dbgen.Queries) error { return q.SupersedePendingPortfolio(ctx) }) 183 return e 184 }) 185 if err != nil { 186 return RunResponse{}, err 187 } 188 return RunResponse{Run: runView(run), Recommendations: recommendationViews(rows)}, nil 189} 190 191// buildPortfolioRecs turns the optimal allocation into ranked recommendations: ASSIGN rows (highest value 192// first) with the value matrix counterfactual, then a SOURCE row per requirement whose open slots the bench 193// could not fill. 194func (s *Service) buildPortfolioRecs(cfg Config, sol PortfolioSolution, pin PortfolioInput, cf map[int]AssignCounterfactual, resByPair map[[2]int]engine.MatchResult, candIDs, candProv []string, candComp []decimal.Decimal, reqIDs []string) []Recommendation { 195 assigns := append([]Assignment(nil), sol.Assignments...) 196 sort.SliceStable(assigns, func(i, j int) bool { 197 if assigns[i].Value != assigns[j].Value { 198 return assigns[i].Value > assigns[j].Value 199 } 200 if assigns[i].CandidateIdx != assigns[j].CandidateIdx { 201 return assigns[i].CandidateIdx < assigns[j].CandidateIdx 202 } 203 return assigns[i].RequirementIdx < assigns[j].RequirementIdx 204 }) 205 206 filled := make([]int, len(pin.Slots)) 207 var recs []Recommendation 208 rank := 0 209 for _, a := range assigns { 210 filled[a.RequirementIdx]++ 211 rank++ 212 res := resByPair[[2]int{a.CandidateIdx, a.RequirementIdx}] 213 ev := Evidence{ 214 EngineTotal: res.Total.StringFixed(2), 215 Band: res.Band.Code(), 216 Counterfactual: portfolioCounterfactual(cf[a.CandidateIdx], res, candIDs, reqIDs), 217 } 218 conf := Confidence(cfg, ScoredCandidate{Total: res.Total, Provenance: candProv[a.CandidateIdx], Completeness: candComp[a.CandidateIdx]}) 219 value := a.Value 220 recs = append(recs, Recommendation{ 221 RequirementID: reqIDs[a.RequirementIdx], 222 CandidateID: candIDs[a.CandidateIdx], 223 Action: ActionAssign, 224 Rank: rank, 225 Confidence: conf, 226 ValueScore: &value, 227 Evidence: ev, 228 }) 229 } 230 for j := range pin.Slots { 231 if remaining := pin.Slots[j] - filled[j]; remaining > 0 { 232 rank++ 233 rem := remaining 234 recs = append(recs, Recommendation{ 235 RequirementID: reqIDs[j], 236 Action: ActionSource, 237 Rank: rank, 238 Confidence: decimal.NewFromInt(1), 239 Evidence: Evidence{OpenSlots: &rem}, 240 }) 241 } 242 } 243 return recs 244} 245 246func portfolioCounterfactual(cf AssignCounterfactual, res engine.MatchResult, candIDs, reqIDs []string) *Counterfactual { 247 c := &Counterfactual{AssignedTotal: res.Total.StringFixed(2)} 248 if cf.CandidateNextReqIdx >= 0 && cf.CandidateNextReqIdx < len(reqIDs) { 249 c.CandidateNextRequirement = reqIDs[cf.CandidateNextReqIdx] 250 d := cf.CandidateValueDelta 251 c.CandidateValueDelta = &d 252 } 253 if cf.RequirementNextCandIdx >= 0 && cf.RequirementNextCandIdx < len(candIDs) { 254 c.RequirementNextCandidate = candIDs[cf.RequirementNextCandIdx] 255 d := cf.RequirementValueDelta 256 c.RequirementValueDelta = &d 257 } 258 return c 259} 260