Janady13/missionmatch / internal/aioe/backfill.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/aioe/backfill.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/aioe/backfill.go
Original SHA-256	4dd96952f6547be0e540360b8d8d613cd8b5cb3eba0df0fcecd757c0fb479f17
Git blob SHA	19d72b0f9660e1b5f3978e7610c71b5e9210588d
Reading-copy SHA-256	4dd96952f6547be0e540360b8d8d613cd8b5cb3eba0df0fcecd757c0fb479f17
Original source ↗
Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/aioe/backfill.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/aioe/backfill.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Recovered source transcript

Download complete source text ↓279 lines · 10.4 KB
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
}