Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/modules/aioe/service.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | internal/modules/aioe/service.go |
| Original SHA-256 | b627d21d4bc38e29d897ee03866a7c6a440e86a11d3f21f126f6dc0efa428eff |
| Git blob SHA | 6984dd2103422cb5d43e2a03adbc18b6b39332ac |
| Reading-copy SHA-256 | b627d21d4bc38e29d897ee03866a7c6a440e86a11d3f21f126f6dc0efa428eff |
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/modules/aioe/service.go | feature/opportunity-to-outcome | 720d47fae476b0d38477ae4401697d372b1ea2bc | Open original |
| Janady13/missionmatch / internal/modules/aioe/service.go | feature/aioe-upgrade | 2178e7298d0e5d9af83afa12402fd587f0ba6ed8 | Open original |
Recovered source transcript
Download complete source text ↓252 lines · 8.8 KB
package aioemodule
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/missionworkforce/missionmatch/internal/aioe"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
var (
ErrBadRequest = errors.New("aioe: bad request")
ErrNoDatabase = errors.New("aioe: no database configured")
ErrNotConfigured = errors.New("aioe: updater not configured")
)
// maxMetadataBytes bounds caller supplied metadata so an event stays a signal, not a payload store.
const maxMetadataBytes = 16 * 1024
// Service records outcome events on the tenant transaction (RLS scoped, app role) and serves the
// admin status and history reads. The updater is optional: when the dedicated aioe pool is not
// configured the on demand run endpoint answers 503 and the nightly schedule is off, while outcome
// capture keeps working (events accumulate for the next configured run).
type Service struct {
db *tenant.DB
updater *aioe.Updater
clock func() time.Time
}
func NewService(db *tenant.DB, updater *aioe.Updater) *Service {
return &Service{db: db, updater: updater, clock: time.Now}
}
// RecordOutcome validates and inserts one outcome event. When the subject is a candidate and a
// requirement is named, the served per dimension scores and the weight profile version are copied
// from the stored evaluation into the event metadata: the capture side of the "updater consumes
// outcome_events only" contract.
func (s *Service) RecordOutcome(ctx context.Context, req PostOutcomeRequest) (PostOutcomeResponse, error) {
if s.db == nil {
return PostOutcomeResponse{}, ErrNoDatabase
}
if !validModules[req.Module] || !validEventTypes[req.EventType] {
return PostOutcomeResponse{}, fmt.Errorf("%w: unknown module or event type", ErrBadRequest)
}
if req.SubjectID == uuid.Nil {
return PostOutcomeResponse{}, fmt.Errorf("%w: subjectId is required", ErrBadRequest)
}
if req.SubjectType == "" {
req.SubjectType = "candidate"
}
signal, ok := parseSignal(req.SignalValue)
if !ok {
return PostOutcomeResponse{}, fmt.Errorf("%w: signalValue must be a number in [-1, 1]", ErrBadRequest)
}
if len(req.Metadata) > maxMetadataBytes {
return PostOutcomeResponse{}, fmt.Errorf("%w: metadata too large", ErrBadRequest)
}
meta := map[string]json.RawMessage{}
if len(req.Metadata) > 0 {
if err := json.Unmarshal(req.Metadata, &meta); err != nil {
return PostOutcomeResponse{}, fmt.Errorf("%w: metadata must be a json object", ErrBadRequest)
}
}
occurred := s.clock().UTC()
if req.OccurredAt != nil {
occurred = req.OccurredAt.UTC()
}
var actor *uuid.UUID
if p, ok := tenant.UserFromContext(ctx); ok {
if id, err := uuid.Parse(p.UserID); err == nil {
actor = &id
}
}
var out PostOutcomeResponse
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
// Denormalize the served dimension scores for the updater. Best effort: an outcome for a
// pair that was never scored still records, it just carries no per dimension evidence.
if req.RequirementID != nil && req.SubjectType == "candidate" {
enrichOutcomeMetadata(ctx, tx, req.SubjectID, *req.RequirementID, meta)
}
metaJSON, err := json.Marshal(meta)
if err != nil {
return err
}
return tx.QueryRow(ctx, `
INSERT INTO outcome_events (
organization_id, module, subject_type, subject_id, event_type,
signal_value, reason_code, actor_id, occurred_at, metadata)
VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid,
$1, $2, $3, $4, $5, $6, $7, $8, $9)
RETURNING id, occurred_at`,
req.Module, req.SubjectType, req.SubjectID, req.EventType,
signal, req.ReasonCode, actor, occurred, metaJSON,
).Scan(&out.ID, &out.OccurredAt)
})
return out, err
}
// enrichOutcomeMetadata copies the stored evaluation's per dimension raw scores and weight profile
// version into the event metadata, silently skipping when no evaluation exists. Only whitelisted
// keys are written; the caller's own metadata keys are preserved unless they collide.
func enrichOutcomeMetadata(ctx context.Context, tx pgx.Tx, candidateID, requirementID uuid.UUID, meta map[string]json.RawMessage) {
var cl, sk, ce, cu, ex, lo, total, profile string
err := tx.QueryRow(ctx, `
SELECT clearance_score::text, skills_score::text, certs_score::text, customer_score::text,
experience_score::text, location_score::text, total_score::text, weight_profile_version
FROM candidate_evaluations
WHERE candidate_id = $1 AND requirement_id = $2`, candidateID, requirementID).
Scan(&cl, &sk, &ce, &cu, &ex, &lo, &total, &profile)
if err != nil {
return
}
scores, err := json.Marshal(map[string]string{
"clearance": cl, "skills": sk, "certifications": ce,
"customer_experience": cu, "experience": ex, "location": lo,
})
if err != nil {
return
}
meta["dimension_scores"] = scores
if b, err := json.Marshal(total); err == nil {
meta["total_score"] = b
}
if b, err := json.Marshal(profile); err == nil {
meta["weight_profile_version"] = b
}
if b, err := json.Marshal(requirementID.String()); err == nil {
meta["requirement_id"] = b
}
}
// Status assembles the admin proof of life panel.
func (s *Service) Status(ctx context.Context) (StatusResponse, error) {
if s.db == nil {
return StatusResponse{}, ErrNoDatabase
}
out := StatusResponse{ActiveProfileVersion: "global.default", UpdaterConfigured: s.updater != nil}
if s.updater != nil {
out.ShadowMode = s.updater.Shadow()
}
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
if err := tx.QueryRow(ctx, `SELECT count(*) FROM outcome_events`).Scan(&out.OutcomesRecorded); err != nil {
return err
}
var version int
err := tx.QueryRow(ctx, `
SELECT profile_version FROM aioe_weight_profiles WHERE status = 'active' LIMIT 1`).Scan(&version)
if err == nil {
out.ActiveProfileVersion = aioe.ProfileLabel(version)
} else if err != pgx.ErrNoRows {
return err
}
// A pending candidate awaiting a human promotion (shadow mode proof of life).
var candVersion int
cerr := tx.QueryRow(ctx, `
SELECT profile_version FROM aioe_weight_profiles WHERE status = 'candidate'
ORDER BY profile_version DESC LIMIT 1`).Scan(&candVersion)
if cerr == nil {
out.PendingCandidateVersion = aioe.ProfileLabel(candVersion)
} else if cerr != pgx.ErrNoRows {
return cerr
}
if err := tx.QueryRow(ctx, `
SELECT count(*) FROM aioe_weight_adjustments
WHERE created_at >= date_trunc('month', now())`).Scan(&out.AdjustmentsThisMonth); err != nil {
return err
}
var ranAt time.Time
var seen, moved int
err = tx.QueryRow(ctx, `
SELECT ran_at, outcomes_seen, profiles_updated FROM aioe_runs
ORDER BY ran_at DESC LIMIT 1`).Scan(&ranAt, &seen, &moved)
if err == nil {
out.LastRunAt = &ranAt
out.LastRunOutcomesSeen = &seen
out.LastRunProfilesMoved = &moved
} else if err != pgx.ErrNoRows {
return err
}
return nil
})
return out, err
}
// History returns the org's adjustment audit trail, newest first.
func (s *Service) History(ctx context.Context, limit int) ([]AdjustmentView, error) {
if s.db == nil {
return nil, ErrNoDatabase
}
if limit <= 0 || limit > 500 {
limit = 100
}
out := []AdjustmentView{}
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
rows, err := tx.Query(ctx, `
SELECT dimension, old_value::text, new_value::text, produced_version,
window_start, window_end, outcomes_used, rationale, created_at
FROM aioe_weight_adjustments
ORDER BY created_at DESC, dimension
LIMIT $1`, limit)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var v AdjustmentView
if err := rows.Scan(&v.Dimension, &v.OldValue, &v.NewValue, &v.ProducedVersion,
&v.WindowStart, &v.WindowEnd, &v.OutcomesUsed, &v.Rationale, &v.CreatedAt); err != nil {
return err
}
out = append(out, v)
}
return rows.Err()
})
return out, err
}
// RunNow triggers one on demand engine pass (admin action). The pass is the same deterministic
// batch the nightly schedule runs; only the trigger differs.
func (s *Service) RunNow(ctx context.Context) (aioe.RunReport, error) {
if s.updater == nil {
return aioe.RunReport{}, ErrNotConfigured
}
return s.updater.RunAll(ctx, s.clock(), "admin")
}
// Promote activates the caller org's pending candidate weight profile (admin action, the human in
// the loop step that ends shadow mode for the tenant). It resolves the org from the validated
// session and promotes only that tenant.
func (s *Service) Promote(ctx context.Context) (aioe.PromoteReport, error) {
if s.updater == nil {
return aioe.PromoteReport{}, ErrNotConfigured
}
p, ok := tenant.OrgFromContext(ctx)
if !ok {
return aioe.PromoteReport{}, tenant.ErrNoTenant
}
org, err := uuid.Parse(p)
if err != nil {
return aioe.PromoteReport{}, fmt.Errorf("%w: invalid org", ErrBadRequest)
}
return s.updater.Promote(ctx, org)
}