Janady13/missionmatch / internal/aioe/updater.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/aioe/updater.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/aioe/updater.go
Original SHA-256	f8877e40fd15fa464e2bfcc062a8ab9d3dc37dd0dcabcc675d787f6b39808195
Git blob SHA	b9ae54685c598af89f280acd0af67f9c977f1b5e
Reading-copy SHA-256	f8877e40fd15fa464e2bfcc062a8ab9d3dc37dd0dcabcc675d787f6b39808195
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/updater.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/aioe/updater.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Recovered source transcript

Download complete source text ↓361 lines · 14.4 KB
package aioe

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/shopspring/decimal"
)

// Updater is the deterministic batch pass: it reads outcome_events in the trailing window per
// tenant, computes bounded weight updates, and writes a new versioned profile plus the append only
// adjustment audit rows. It runs nightly or on demand, NEVER per request. The pool it holds MUST
// authenticate as the missionmatch_aioe role: its grants are the isolation guarantee (the three
// aioe tables and nothing else), and the integration suite proves any other read is denied.
type Updater struct {
	db     *pgxpool.Pool
	params Params
	log    *slog.Logger
	// shadow, when true, makes a run write the new profile as a CANDIDATE (status 'candidate')
	// without flipping the active profile: the engine learns and proposes, but a human promotes.
	// This is the safe pre production and pilot posture (the MAG "never flip the champion" precedent).
	shadow bool
}

// NewUpdater builds the updater over the dedicated aioe pool. log may be nil.
func NewUpdater(db *pgxpool.Pool, params Params, log *slog.Logger) *Updater {
	if log == nil {
		log = slog.Default()
	}
	return &Updater{db: db, params: params, log: log}
}

// WithShadow sets shadow mode (write candidate profiles, never auto activate) and returns the
// updater for chaining. A human promotes a candidate through Promote.
func (u *Updater) WithShadow(shadow bool) *Updater {
	u.shadow = shadow
	return u
}

// Shadow reports whether the updater is in shadow mode.
func (u *Updater) Shadow() bool { return u.shadow }

// RunReport summarizes one pass for the run ledger and the status endpoint.
type RunReport struct {
	AsOf            time.Time `json:"asOf"`
	OrgsScanned     int       `json:"orgsScanned"`
	OutcomesSeen    int       `json:"outcomesSeen"`
	ProfilesUpdated int       `json:"profilesUpdated"`
}

// RunAll executes one deterministic pass over every tenant with outcomes, as of the given instant
// (truncated to the UTC day so a rerun inside the same day consumes the identical window).
// triggeredBy is "schedule" or "admin" for the run ledger.
func (u *Updater) RunAll(ctx context.Context, asOf time.Time, triggeredBy string) (RunReport, error) {
	asOf = asOf.UTC().Truncate(24 * time.Hour)
	report := RunReport{AsOf: asOf}
	windowStart := asOf.AddDate(0, 0, -u.params.WindowDays)

	rows, err := u.db.Query(ctx, `
		SELECT DISTINCT organization_id FROM outcome_events
		WHERE occurred_at > $1 AND occurred_at <= $2
		ORDER BY organization_id`, windowStart, asOf)
	if err != nil {
		return report, fmt.Errorf("aioe: list orgs: %w", err)
	}
	var orgs []uuid.UUID
	for rows.Next() {
		var id uuid.UUID
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return report, fmt.Errorf("aioe: scan org: %w", err)
		}
		orgs = append(orgs, id)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return report, fmt.Errorf("aioe: iterate orgs: %w", err)
	}

	for _, org := range orgs {
		used, updated, err := u.runOrg(ctx, org, windowStart, asOf)
		if err != nil {
			// One tenant's failure never blocks the sweep; it is logged and the pass continues.
			u.log.Error("aioe: tenant update failed", "org", org.String(), "err", err)
			continue
		}
		report.OrgsScanned++
		report.OutcomesSeen += used
		if updated {
			report.ProfilesUpdated++
		}
	}

	_, err = u.db.Exec(ctx, `
		INSERT INTO aioe_runs (as_of, window_days, orgs_scanned, outcomes_seen, profiles_updated, triggered_by)
		VALUES ($1, $2, $3, $4, $5, $6)`,
		asOf, u.params.WindowDays, report.OrgsScanned, report.OutcomesSeen, report.ProfilesUpdated, triggeredBy)
	if err != nil {
		return report, fmt.Errorf("aioe: record run: %w", err)
	}
	return report, nil
}

// runOrg updates one tenant: load its active profile (or the global default), consume its window
// of outcomes, compute the bounded update, and persist a new profile version plus the audit rows
// when anything moved. Returns the outcomes consumed and whether a new profile was produced.
func (u *Updater) runOrg(ctx context.Context, org uuid.UUID, windowStart, asOf time.Time) (int, bool, error) {
	// The outcome window is read BEFORE the transaction opens, deliberately. It is an append only
	// history that no concurrent writer here can revert, so it is not part of the race below, and
	// reading it outside keeps the profile lock held for as short a time as possible.
	events, err := u.loadEvents(ctx, org, windowStart, asOf)
	if err != nil {
		return 0, false, err
	}

	tx, err := u.db.Begin(ctx)
	if err != nil {
		return 0, false, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	// The active profile read and the profile write MUST be one atomic unit, and the read MUST take a
	// row lock. Reading the active profile outside the transaction and then writing a profile derived
	// from that read is a textbook lost update: a Promote committing in between activates a candidate,
	// and this pass then retires it and installs weights computed from the profile Promote just
	// superseded, silently reverting the human in the loop decision. Worse, the aioe_weight_adjustments
	// trail would record old_value from the stale profile, so the append only compliance record of an
	// automated employment decision tool would state a starting point that was never in effect at the
	// moment of the change. FOR UPDATE makes Promote and this pass serialize on the same row: whichever
	// takes the lock first completes, and the other sees the committed result.
	old := DefaultWeightsMap()
	var activeID *uuid.UUID
	var activeVersion int
	{
		var raw map[string]string
		var id uuid.UUID
		err := tx.QueryRow(ctx, `
			SELECT id, profile_version, dimension_weights FROM aioe_weight_profiles
			WHERE organization_id = $1 AND status = 'active'
			FOR UPDATE`, org).Scan(&id, &activeVersion, &raw)
		switch {
		case err == pgx.ErrNoRows:
			// No profile yet: the update starts from the global default. There is no row to lock, so two
			// concurrent first passes could both try to insert an active profile; the one active per org
			// unique index refuses the second, which fails the run loudly instead of producing two.
		case err != nil:
			return 0, false, fmt.Errorf("load active profile: %w", err)
		default:
			activeID = &id
			parsed, perr := parseWeightMap(raw)
			if perr != nil {
				return 0, false, fmt.Errorf("active profile malformed: %w", perr)
			}
			old = parsed
		}
	}

	next, changes, used := ComputeUpdate(old, events, u.params)
	if len(changes) == 0 {
		// Nothing moved. The deferred rollback releases the row lock; no profile is written.
		return used, false, nil
	}

	var newVersion int
	if err := tx.QueryRow(ctx, `
		SELECT COALESCE(MAX(profile_version), 0) + 1 FROM aioe_weight_profiles
		WHERE organization_id = $1`, org).Scan(&newVersion); err != nil {
		return used, false, fmt.Errorf("next version: %w", err)
	}
	// Status and precedence differ by mode. In LIVE mode a run retires the current active profile and
	// the new one becomes active immediately. In SHADOW mode the new profile is a candidate proposed
	// FROM the active weights; the active profile is left untouched and any prior candidate is retired,
	// so at most one pending candidate stands for a human to promote.
	status := "active"
	if u.shadow {
		status = "candidate"
		if _, err := tx.Exec(ctx, `
			UPDATE aioe_weight_profiles SET status = 'retired'
			WHERE organization_id = $1 AND status = 'candidate'`, org); err != nil {
			return used, false, fmt.Errorf("retire prior candidate: %w", err)
		}
	} else if activeID != nil {
		if _, err := tx.Exec(ctx, `
			UPDATE aioe_weight_profiles SET status = 'retired'
			WHERE organization_id = $1 AND id = $2`, org, *activeID); err != nil {
			return used, false, fmt.Errorf("retire profile v%d: %w", activeVersion, err)
		}
	}
	payload, err := json.Marshal(weightMapToJSON(next))
	if err != nil {
		return used, false, err
	}
	var profileID uuid.UUID
	if err := tx.QueryRow(ctx, `
		INSERT INTO aioe_weight_profiles (organization_id, profile_version, dimension_weights, status)
		VALUES ($1, $2, $3, $4) RETURNING id`, org, newVersion, payload, status).Scan(&profileID); err != nil {
		return used, false, fmt.Errorf("insert profile: %w", err)
	}
	for _, ch := range changes {
		if _, err := tx.Exec(ctx, `
			INSERT INTO aioe_weight_adjustments (
				organization_id, profile_id, produced_version, dimension,
				old_value, new_value, window_start, window_end, outcomes_used, rationale)
			VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`,
			org, profileID, newVersion, ch.Dimension, ch.Old, ch.New,
			windowStart, asOf, used, ch.Rationale); err != nil {
			return used, false, fmt.Errorf("insert adjustment %s: %w", ch.Dimension, err)
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return used, false, err
	}
	u.log.Info("aioe: profile updated", "org", org.String(), "version", newVersion, "changes", len(changes), "outcomes", used)
	return used, true, nil
}

// PromoteReport names the outcome of a promotion.
type PromoteReport struct {
	PromotedVersion int    `json:"promotedVersion"`
	RetiredVersion  int    `json:"retiredVersion"` // the previously active version, 0 if there was none
	Message         string `json:"message"`
}

// Promote activates the newest candidate profile for one org and retires the currently active
// profile (and any other candidates), the human in the loop step that ends shadow mode for a tenant.
// It runs on the aioe pool (which holds the UPDATE grant and the cross tenant policy) scoped to the
// caller's org, so an admin promotes only their own tenant. Idempotent shape: with no pending
// candidate it reports so and changes nothing.
func (u *Updater) Promote(ctx context.Context, org uuid.UUID) (PromoteReport, error) {
	tx, err := u.db.Begin(ctx)
	if err != nil {
		return PromoteReport{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	// The ACTIVE row is locked FIRST, before the candidate is even looked up, for two reasons. It is the
	// same row runOrg locks, so the two writers serialize on it instead of racing: a nightly pass can no
	// longer compute from a profile this promotion is about to supersede, and this promotion can no longer
	// overwrite a profile the pass just installed. Taking it first also fixes ONE lock order for both
	// paths (active, then candidates), so a shadow mode pass retiring candidates cannot deadlock against a
	// promotion holding a candidate and waiting on the active row.
	var retiredVersion int
	err = tx.QueryRow(ctx, `
		SELECT profile_version FROM aioe_weight_profiles
		WHERE organization_id = $1 AND status = 'active'
		FOR UPDATE`, org).Scan(&retiredVersion)
	if err != nil && err != pgx.ErrNoRows {
		return PromoteReport{}, fmt.Errorf("find active: %w", err)
	}

	var candID uuid.UUID
	var candVersion int
	err = tx.QueryRow(ctx, `
		SELECT id, profile_version FROM aioe_weight_profiles
		WHERE organization_id = $1 AND status = 'candidate'
		ORDER BY profile_version DESC LIMIT 1
		FOR UPDATE`, org).Scan(&candID, &candVersion)
	if err == pgx.ErrNoRows {
		return PromoteReport{Message: "no pending candidate profile to promote"}, nil
	}
	if err != nil {
		return PromoteReport{}, fmt.Errorf("find candidate: %w", err)
	}

	// Retire the current active and every candidate other than the one being promoted, then activate
	// it. The one active per org unique index is satisfied because the old active is retired first.
	if _, err := tx.Exec(ctx, `
		UPDATE aioe_weight_profiles SET status = 'retired'
		WHERE organization_id = $1 AND (status = 'active' OR (status = 'candidate' AND id <> $2))`, org, candID); err != nil {
		return PromoteReport{}, fmt.Errorf("retire prior: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		UPDATE aioe_weight_profiles SET status = 'active'
		WHERE organization_id = $1 AND id = $2`, org, candID); err != nil {
		return PromoteReport{}, fmt.Errorf("activate candidate: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return PromoteReport{}, err
	}
	u.log.Info("aioe: candidate promoted", "org", org.String(), "version", candVersion, "retired", retiredVersion)
	return PromoteReport{PromotedVersion: candVersion, RetiredVersion: retiredVersion,
		Message: fmt.Sprintf("promoted candidate profile v%d to active", candVersion)}, nil
}

// loadEvents reads the tenant's outcome window: the signal and the denormalized per dimension
// scores from metadata. Events without dimension scores still count toward the window total (they
// move MinEvents) but contribute no per dimension evidence.
func (u *Updater) loadEvents(ctx context.Context, org uuid.UUID, windowStart, asOf time.Time) ([]Event, error) {
	rows, err := u.db.Query(ctx, `
		SELECT signal_value, metadata FROM outcome_events
		WHERE organization_id = $1 AND occurred_at > $2 AND occurred_at <= $3
		ORDER BY occurred_at, id`, org, windowStart, asOf)
	if err != nil {
		return nil, fmt.Errorf("load events: %w", err)
	}
	defer rows.Close()
	var events []Event
	for rows.Next() {
		var signal decimal.Decimal
		var meta []byte
		if err := rows.Scan(&signal, &meta); err != nil {
			return nil, fmt.Errorf("scan event: %w", err)
		}
		events = append(events, Event{Signal: signal, DimScores: dimScoresFromMetadata(meta)})
	}
	return events, rows.Err()
}

// dimScoresFromMetadata extracts metadata.dimension_scores (decimal strings keyed by canonical
// dimension). Anything malformed simply yields no scores for that event, never an error: the
// updater is a consumer of best effort capture, not a validator of history.
func dimScoresFromMetadata(meta []byte) map[string]decimal.Decimal {
	var wrapper struct {
		DimensionScores map[string]string `json:"dimension_scores"`
	}
	if err := json.Unmarshal(meta, &wrapper); err != nil || len(wrapper.DimensionScores) == 0 {
		return nil
	}
	out := make(map[string]decimal.Decimal, len(wrapper.DimensionScores))
	for _, dim := range Dimensions {
		raw, ok := wrapper.DimensionScores[dim]
		if !ok {
			continue
		}
		if d, err := decimal.NewFromString(raw); err == nil {
			out[dim] = d
		}
	}
	return out
}

func parseWeightMap(raw map[string]string) (map[string]decimal.Decimal, error) {
	out := make(map[string]decimal.Decimal, len(Dimensions))
	for _, dim := range Dimensions {
		s, ok := raw[dim]
		if !ok {
			return nil, fmt.Errorf("missing dimension %s", dim)
		}
		d, err := decimal.NewFromString(s)
		if err != nil {
			return nil, fmt.Errorf("dimension %s: %w", dim, err)
		}
		out[dim] = d
	}
	return out, nil
}

func weightMapToJSON(m map[string]decimal.Decimal) map[string]string {
	out := make(map[string]string, len(Dimensions))
	for _, dim := range Dimensions {
		out[dim] = m[dim].String()
	}
	return out
}