// Command opp-sync runs the Opportunity Intelligence ingestion. It pulls the configured federal feeds // (SAM active solicitations, and federal contract awards from the SAM Contract Awards API with a // USASpending fallback) through the OpportunitySource adapters and upserts them into the unified // opportunity_records catalog, retaining each raw payload to object storage and writing one audit event // per source. It runs as the migrator role and is safe to rerun: every upsert is idempotent. // // Env: MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL), SAM_API_KEY, SAM_API_BASE (optional), // SAM_AWARDS_BASE (optional), USASPENDING_BASE (optional), MM_OPP_RAW_DIR (object storage root, default // ./data/opp-raw). package main import ( "context" "log" "os" "strings" "time" "github.com/jackc/pgx/v5/pgxpool" oppintel "github.com/missionworkforce/missionmatch/internal/modules/opportunity_intelligence" "github.com/missionworkforce/missionmatch/internal/modules/opportunity_intelligence/adapters" "github.com/missionworkforce/missionmatch/internal/platform/audit" "github.com/missionworkforce/missionmatch/internal/platform/marketintel" "github.com/missionworkforce/missionmatch/internal/platform/samawards" "github.com/missionworkforce/missionmatch/internal/platform/samgov" "github.com/missionworkforce/missionmatch/internal/platform/storage" "github.com/missionworkforce/missionmatch/internal/platform/tenant" ) func main() { dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL")) if dsn == "" { log.Fatal("opp-sync: set MM_MIGRATOR_DATABASE_URL (the migrator DSN)") } samKey := strings.TrimSpace(os.Getenv("SAM_API_KEY")) if samKey == "" { log.Fatal("opp-sync: SAM_API_KEY is [credential removed]") } rawDir := firstNonEmpty(os.Getenv("MM_OPP_RAW_DIR"), "./data/opp-raw") ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() pool, err := pgxpool.New(ctx, dsn) if err != nil { log.Fatalf("opp-sync: connect: %v", err) } defer pool.Close() store, err := storage.NewLocal(rawDir) if err != nil { log.Fatalf("opp-sync: storage: %v", err) } ing := oppintel.NewIngester(store, oppintel.NewDBWriter(pool), audit.New(tenant.New(pool))) // The federal award feed is the SAM Contract Awards API (the FPDS replacement) as the primary, with // the free USASpending pull as the resilient fallback if SAM is throttled or unavailable. samAwards := adapters.NewSAMContractAwardsSource(samawards.New(os.Getenv("SAM_AWARDS_BASE"), samKey), galapagosAwardsParams(), 5) usaspending := adapters.NewUSASpendingSource(marketintel.NewUSASpending(os.Getenv("USASPENDING_BASE")), galapagosAwardQuery(), 5) sources := []oppintel.OpportunitySource{ adapters.NewSAMSource(samgov.New(os.Getenv("SAM_API_BASE"), samKey), samgov.Galapagos, 5), adapters.NewFallbackAwardsSource(samAwards, usaspending), } var grand int for _, src := range sources { res, ierr := ing.Ingest(ctx, src) if ierr != nil { log.Fatalf("opp-sync: ingest %s: %v", src.Name(), ierr) } log.Printf("opp-sync: %s fetched %d upserted %d by_kind %v", res.Source, res.Fetched, res.Upserted, res.ByKind) grand += res.Upserted } log.Printf("opp-sync: done, %d records upserted into opportunity_records", grand) } // galapagosAwardsParams is the SAM Contract Awards pull (the Galapagos NAICS family under the Department // of Defense, the last three years of signed actions). The Contract Awards API takes all NAICS in one // query, and place of performance is filtered in the read layer, the same way the opportunities feed // filters state against the cache. func galapagosAwardsParams() samawards.SearchParams { now := time.Now().UTC() return samawards.SearchParams{ NAICS: samgov.Galapagos.NAICS, Department: "DEFENSE", SignedFrom: now.AddDate(-3, 0, 0), SignedTo: now, Limit: 100, } } // galapagosAwardQuery is the USASpending fallback pull (DoD, the Galapagos NAICS family, HI and NM, the // last three years of contract actions). It mirrors the standalone awards-sync market pass. func galapagosAwardQuery() marketintel.AwardQuery { now := time.Now().UTC() start := now.AddDate(-3, 0, 0) return marketintel.AwardQuery{ NAICS: samgov.Galapagos.NAICS, States: samgov.Galapagos.States, AwardTypeCodes: []string{"A", "B", "C", "D"}, AwardingToptier: "Department of Defense", TimePeriods: []marketintel.TimePeriod{{Start: start.Format("2006-01-02"), End: now.Format("2006-01-02")}}, Limit: 100, } } func firstNonEmpty(vals ...string) string { for _, v := range vals { if s := strings.TrimSpace(v); s != "" { return s } } return "" }