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