package aioe import ( "context" "encoding/json" "fmt" "github.com/jackc/pgx/v5" ) // BackfillReport counts what one org's backfill produced. type BackfillReport struct { SlateDecisions int `json:"slateDecisions"` VendorOutcomes int `json:"vendorOutcomes"` BidOutcomes int `json:"bidOutcomes"` SkippedExisting int `json:"skippedExisting"` } // Total is the number of new outcome events written for the org. func (r BackfillReport) Total() int { return r.SlateDecisions + r.VendorOutcomes + r.BidOutcomes } // BackfillVersion marks rows this adapter wrote, so a re-run is idempotent: every insert is guarded // by NOT EXISTS a prior event carrying the same backfill source and key. The table stays append // only (no deletes); a re-run simply skips what it already produced. const BackfillVersion = "v1" // BackfillOrg converts an org's real decision history into outcome_events on the caller's tenant tx // (RLS scoped, app role, INSERT only). Three sources, richest first: // // - slate_members with a human decision (approved or rejected): the strongest signal, and it // carries the per dimension scores inline, so the AIOE evidence is exact. approved is +1, // rejected is -1, module slate, event slate_result. // - vendor_submissions lifecycle: placed +1, interviewed +0.5, rejected -1, module screen, event // placement_result. Dimension scores come from the stored evaluation on the (candidate, // requirement) pair when one exists, else the event carries the signal alone. // - bid_outcomes win or loss on a slate: reflected onto each APPROVED member of that slate as an // offer_result (+1 win, -1 loss), dimension scores from the member row. // // It NEVER reads a demographic or protected attribute (none of these tables carry one), and it // records occurred_at from the real historical timestamp so the updater's window math is honest. func BackfillOrg(ctx context.Context, tx pgx.Tx, actorNote string) (BackfillReport, error) { var rep BackfillReport // 1. Slate member decisions (richest: dimension scores inline). slateRows, err := tx.Query(ctx, ` SELECT sm.slate_id, sm.candidate_id, s.requirement_id, sm.decision, sm.decided_at::text, sm.decided_by::text, sm.clearance_score::text, sm.skills_score::text, sm.certs_score::text, sm.customer_score::text, sm.experience_score::text, sm.location_score::text, sm.total_score::text FROM slate_members sm JOIN slates s ON s.organization_id = sm.organization_id AND s.id = sm.slate_id WHERE sm.decision IN ('approved','rejected')`) if err != nil { return rep, fmt.Errorf("query slate decisions: %w", err) } type slateDec struct { slateID, candidateID, requirementID string decision string decidedAt, decidedBy *string cl, sk, ce, cu, ex, lo, total string } var decisions []slateDec for slateRows.Next() { var d slateDec if err := slateRows.Scan(&d.slateID, &d.candidateID, &d.requirementID, &d.decision, &d.decidedAt, &d.decidedBy, &d.cl, &d.sk, &d.ce, &d.cu, &d.ex, &d.lo, &d.total); err != nil { slateRows.Close() return rep, fmt.Errorf("scan slate decision: %w", err) } decisions = append(decisions, d) } slateRows.Close() if err := slateRows.Err(); err != nil { return rep, err } for _, d := range decisions { signal := "1" if d.decision == "rejected" { signal = "-1" } meta := backfillMeta("slate_member", d.slateID+":"+d.candidateID, map[string]string{ "clearance": d.cl, "skills": d.sk, "certifications": d.ce, "customer_experience": d.cu, "experience": d.ex, "location": d.lo, }, d.total, d.requirementID) ins, err := insertBackfillEvent(ctx, tx, "slate", "candidate", d.candidateID, "slate_result", signal, "backfill_slate_decision", d.decidedBy, d.decidedAt, meta) if err != nil { return rep, err } if ins { rep.SlateDecisions++ } else { rep.SkippedExisting++ } } // 2. Vendor submission lifecycle, dimension scores from the stored evaluation when present. vendorRows, err := tx.Query(ctx, ` SELECT vs.candidate_id, vs.requirement_id, vs.status, vs.decided_at::text, ce.clearance_score::text, ce.skills_score::text, ce.certs_score::text, ce.customer_score::text, ce.experience_score::text, ce.location_score::text, ce.total_score::text FROM vendor_submissions vs LEFT JOIN candidate_evaluations ce ON ce.organization_id = vs.organization_id AND ce.candidate_id = vs.candidate_id AND ce.requirement_id = vs.requirement_id WHERE vs.status IN ('interviewed','placed','rejected')`) if err != nil { return rep, fmt.Errorf("query vendor submissions: %w", err) } type vsub struct { candidateID, requirementID, status string decidedAt *string cl, sk, ce, cu, ex, lo, total *string } var subs []vsub for vendorRows.Next() { var v vsub if err := vendorRows.Scan(&v.candidateID, &v.requirementID, &v.status, &v.decidedAt, &v.cl, &v.sk, &v.ce, &v.cu, &v.ex, &v.lo, &v.total); err != nil { vendorRows.Close() return rep, fmt.Errorf("scan vendor submission: %w", err) } subs = append(subs, v) } vendorRows.Close() if err := vendorRows.Err(); err != nil { return rep, err } for _, v := range subs { signal := map[string]string{"placed": "1", "interviewed": "0.5", "rejected": "-1"}[v.status] var dims map[string]string var total string if v.cl != nil { dims = map[string]string{ "clearance": deref(v.cl), "skills": deref(v.sk), "certifications": deref(v.ce), "customer_experience": deref(v.cu), "experience": deref(v.ex), "location": deref(v.lo), } total = deref(v.total) } meta := backfillMeta("vendor_submission", v.candidateID+":"+v.requirementID+":"+v.status, dims, total, v.requirementID) ins, err := insertBackfillEvent(ctx, tx, "screen", "candidate", v.candidateID, "placement_result", signal, "backfill_vendor_"+v.status, nil, v.decidedAt, meta) if err != nil { return rep, err } if ins { rep.VendorOutcomes++ } else { rep.SkippedExisting++ } } // 3. Bid outcomes reflected onto approved slate members (offer level signal). bidRows, err := tx.Query(ctx, ` SELECT sm.candidate_id, s.requirement_id, bo.result, bo.award_date::text, bo.slate_id, sm.clearance_score::text, sm.skills_score::text, sm.certs_score::text, sm.customer_score::text, sm.experience_score::text, sm.location_score::text, sm.total_score::text FROM bid_outcomes bo JOIN slates s ON s.organization_id = bo.organization_id AND s.id = bo.slate_id JOIN slate_members sm ON sm.organization_id = bo.organization_id AND sm.slate_id = bo.slate_id WHERE bo.result IN ('win','loss') AND bo.slate_id IS NOT NULL AND sm.decision = 'approved'`) if err != nil { return rep, fmt.Errorf("query bid outcomes: %w", err) } type bid struct { candidateID, requirementID, result, slateID string awardDate *string cl, sk, ce, cu, ex, lo, total string } var bids []bid for bidRows.Next() { var b bid if err := bidRows.Scan(&b.candidateID, &b.requirementID, &b.result, &b.awardDate, &b.slateID, &b.cl, &b.sk, &b.ce, &b.cu, &b.ex, &b.lo, &b.total); err != nil { bidRows.Close() return rep, fmt.Errorf("scan bid outcome: %w", err) } bids = append(bids, b) } bidRows.Close() if err := bidRows.Err(); err != nil { return rep, err } for _, b := range bids { signal := "1" if b.result == "loss" { signal = "-1" } meta := backfillMeta("bid_outcome", b.slateID+":"+b.candidateID, map[string]string{ "clearance": b.cl, "skills": b.sk, "certifications": b.ce, "customer_experience": b.cu, "experience": b.ex, "location": b.lo, }, b.total, b.requirementID) ins, err := insertBackfillEvent(ctx, tx, "slate", "candidate", b.candidateID, "offer_result", signal, "backfill_bid_"+b.result, nil, b.awardDate, meta) if err != nil { return rep, err } if ins { rep.BidOutcomes++ } else { rep.SkippedExisting++ } } return rep, nil } // backfillMeta builds the event metadata: the dedup source and key, plus the denormalized dimension // scores and total the updater consumes (omitted when the source had no scores). func backfillMeta(source, key string, dims map[string]string, total, requirementID string) map[string]any { m := map[string]any{"backfill": BackfillVersion, "source": source, "source_key": key} if len(dims) > 0 { m["dimension_scores"] = dims } if total != "" { m["total_score"] = total } if requirementID != "" { m["requirement_id"] = requirementID } return m } // insertBackfillEvent inserts one outcome event guarded by NOT EXISTS a prior event with the same // backfill source and source_key (idempotent re-runs). occurredAt is the real historical timestamp // as a string (date or timestamptz text) or nil, in which case the event dates to now(). Returns // true when a row was inserted, false when an existing backfilled event made it a no op. func insertBackfillEvent(ctx context.Context, tx pgx.Tx, module, subjectType, subjectID, eventType, signal, reason string, actorID, occurredAt *string, meta map[string]any) (bool, error) { source, _ := meta["source"].(string) key, _ := meta["source_key"].(string) var exists bool if err := tx.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM outcome_events WHERE metadata->>'backfill' = $1 AND metadata->>'source' = $2 AND metadata->>'source_key' = $3 )`, BackfillVersion, source, key).Scan(&exists); err != nil { return false, fmt.Errorf("dedup check: %w", err) } if exists { return false, nil } metaJSON, err := json.Marshal(meta) if err != nil { return false, err } occ := "now()" args := []any{module, subjectType, subjectID, eventType, signal, reason, metaJSON} if occurredAt != nil && *occurredAt != "" { occ = "$8::timestamptz" args = append(args, *occurredAt) } // actor is optional; a backfilled decision keeps its human actor when the source had one. actorSQL := "NULL" if actorID != nil && *actorID != "" { actorSQL = fmt.Sprintf("$%d::uuid", len(args)+1) args = append(args, *actorID) } sql := fmt.Sprintf(` INSERT INTO outcome_events ( organization_id, module, subject_type, subject_id, event_type, signal_value, reason_code, metadata, occurred_at, actor_id) VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1, $2, $3, $4, $5::numeric, $6, $7::jsonb, %s, %s)`, occ, actorSQL) if _, err := tx.Exec(ctx, sql, args...); err != nil { return false, fmt.Errorf("insert backfill event: %w", err) } return true, nil } func deref(s *string) string { if s == nil { return "" } return *s }