Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/aioe-backfill/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/aioe-backfill/main.go |
| Original SHA-256 | ada5f20120f22df7be8742a16ef8fe3b4e418006b342b9824bdb87ebcf02e4f2 |
| Git blob SHA | 672441a71c754bd41cad9064d8c087002a699ae6 |
| Reading-copy SHA-256 | ada5f20120f22df7be8742a16ef8fe3b4e418006b342b9824bdb87ebcf02e4f2 |
Recovered source transcript
Download complete source text ↓150 lines · 4.8 KB
// Command aioe-backfill converts an organization's real decision history into outcome_events so the
// Adaptive Intelligence Optimization Engine has genuine, attributed evidence on day one instead of a
// cold empty ledger. It reads three existing sources, richest first: slate_members human decisions
// (approve or reject, with the per dimension scores inline), vendor_submissions lifecycle (placed,
// interviewed, rejected), and bid_outcomes win or loss reflected onto approved slate members.
//
// It follows the alert-sweep pattern: list organizations through the migrator role (which sees every
// org), then do the per org work through the NOBYPASSRLS app role under RLS, so one tenant's backfill
// can never read or write another's. It is INSERT only and idempotent: every event carries a
// deterministic backfill source and key, and a re-run skips what it already produced. Output is per
// source COUNTS only, never PII. Pass -org to limit the run to a single organization.
//
// Env (server side only): MM_DATABASE_URL (the app DSN, NOBYPASSRLS, required),
// MM_MIGRATOR_DATABASE_URL (the migrator DSN for listing orgs; falls back to MM_DATABASE_URL).
//
// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run with GODEBUG=fips140=on.
package main
import (
"context"
"crypto/fips140"
"crypto/pbkdf2"
"crypto/sha256"
"flag"
"log"
"os"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/missionworkforce/missionmatch/internal/aioe"
"github.com/missionworkforce/missionmatch/internal/platform/config"
"github.com/missionworkforce/missionmatch/internal/platform/database"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
func main() {
orgFlag := flag.String("org", "", "limit the backfill to a single organization id (default: all orgs)")
flag.Parse()
if err := fipsSelfTest(); err != nil {
log.Fatalf("aioe-backfill: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err)
}
log.Printf("aioe-backfill: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version())
cfg, err := config.Load()
if err != nil {
log.Fatalf("aioe-backfill: config: %v", err)
}
if cfg.AppDatabaseURL == "" {
log.Fatal("aioe-backfill: MM_DATABASE_URL (the app DSN) is required")
}
migDSN := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), cfg.AppDatabaseURL)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
defer cancel()
appPool, err := database.NewPool(ctx, cfg.AppDatabaseURL)
if err != nil {
log.Fatalf("aioe-backfill: app pool: %v", err)
}
defer appPool.Close()
if err := database.AssertRole(ctx, appPool, database.AppRoleName()); err != nil {
log.Fatalf("aioe-backfill: app pool role check: %v", err)
}
appDB := tenant.New(appPool)
orgIDs, err := listOrgs(ctx, migDSN, *orgFlag)
if err != nil {
log.Fatalf("aioe-backfill: list orgs: %v", err)
}
if len(orgIDs) == 0 {
log.Print("aioe-backfill: no organizations to backfill")
return
}
var done int
var totSlate, totVendor, totBid, totSkip int
for _, orgID := range orgIDs {
orgCtx := tenant.WithOrg(ctx, orgID.String())
var rep aioe.BackfillReport
err := appDB.RunInTenantTx(orgCtx, func(ctx context.Context, tx pgx.Tx) error {
var e error
rep, e = aioe.BackfillOrg(ctx, tx, "aioe-backfill")
return e
})
if err != nil {
log.Printf("aioe-backfill: org %s failed: %v", orgID, err)
continue
}
done++
totSlate += rep.SlateDecisions
totVendor += rep.VendorOutcomes
totBid += rep.BidOutcomes
totSkip += rep.SkippedExisting
if rep.Total() > 0 || rep.SkippedExisting > 0 {
log.Printf("aioe-backfill: org %s slate_decisions=%d vendor=%d bid=%d skipped_existing=%d",
orgID, rep.SlateDecisions, rep.VendorOutcomes, rep.BidOutcomes, rep.SkippedExisting)
}
}
log.Printf("aioe-backfill: done, %d of %d org(s), slate_decisions=%d vendor=%d bid=%d skipped_existing=%d",
done, len(orgIDs), totSlate, totVendor, totBid, totSkip)
}
func listOrgs(ctx context.Context, migDSN, only string) ([]uuid.UUID, error) {
if only != "" {
id, err := uuid.Parse(strings.TrimSpace(only))
if err != nil {
return nil, err
}
return []uuid.UUID{id}, nil
}
pool, err := pgxpool.New(ctx, migDSN)
if err != nil {
return nil, err
}
defer pool.Close()
rows, err := pool.Query(ctx, "SELECT id FROM organizations ORDER BY id")
if err != nil {
return nil, err
}
defer rows.Close()
var out []uuid.UUID
for rows.Next() {
var id uuid.UUID
if err := rows.Scan(&id); err != nil {
return nil, err
}
out = append(out, id)
}
return out, rows.Err()
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if s := strings.TrimSpace(v); s != "" {
return s
}
}
return ""
}
func fipsSelfTest() error {
_, err := pbkdf2.Key(sha256.New, "aioe-backfill-self-test", []byte("aioe-backfill-salt"), 600000, 32)
return err
}