// 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_FORECAST_CSV (optional, a URL or a file // path to a GSA or agency forecast export), MM_OPP_RAW_DIR (object storage root, default ./data/opp-raw). package main import ( "context" "log" "net/http" "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/forecast" "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), } // The forecast feed (the GSA Forecast of Contracting Opportunities, or any agency forecast CSV) is // enabled only when an export is configured, because the GSA export is a manual download from a gated // dashboard rather than a clean API. MM_FORECAST_CSV is a URL or a local file path. if fc := strings.TrimSpace(os.Getenv("MM_FORECAST_CSV")); fc != "" { var fetch forecast.Fetcher if strings.HasPrefix(fc, "http://") || strings.HasPrefix(fc, "https://") { fetch = forecast.HTTPFetcher(&http.Client{Timeout: 60 * time.Second}, fc) } else { fetch = forecast.FileFetcher(fc) } sources = append(sources, adapters.NewForecastSource(forecast.NewSource("GSA Forecast", fetch, forecast.GSAFCOMapping()))) } 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 "" }