Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/sam-sync/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/sam-sync/main.go |
| Original SHA-256 | dd6d4d19f4cb0582b7315729a325a6d7a409af9d1114b7f5c350526e89ca827a |
| Git blob SHA | 2b5ca3361f284798b4d7c571b0740543d2e1f34d |
| Reading-copy SHA-256 | dd6d4d19f4cb0582b7315729a325a6d7a409af9d1114b7f5c350526e89ca827a |
Recovered source transcript
Download complete source text ↓238 lines · 8.9 KB
// Command sam-sync pulls federal Contract Opportunities from SAM.gov and upserts them into the cached
// sam_opportunities catalog. It is a one-shot job: run it for a demo (`make sync-sam`) or daily via
// the systemd timer in infra/systemd. It pulls by NAICS only (the Galapagos profile), normalizes, and
// upserts on noticeId. On a 429 it logs and stops gracefully, it never hammers the API.
//
// Env (server-side only): SAM_API_KEY (required), SAM_API_BASE (optional, defaults prod),
// MM_MIGRATOR_DATABASE_URL (write DSN; falls back to MM_DATABASE_URL).
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"hash"
"io"
"log"
"os"
"strings"
"time"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/missionworkforce/missionmatch/internal/platform/database"
"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
"github.com/missionworkforce/missionmatch/internal/platform/samgov"
)
// pageCap bounds a single NAICS pull (limit 1000 * 5 = up to 5000 notices/NAICS) so one run can never
// exhaust the ~1000 req/day quota. A cap hit is logged (never silent).
const pageCap = 5
func main() {
ctx := context.Background()
key := strings.TrimSpace(os.Getenv("SAM_API_KEY"))
base := strings.TrimSpace(os.Getenv("SAM_API_BASE"))
dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL"))
if key == "" {
log.Fatal("sam-sync: SAM_API_KEY not set (server-side env only, never commit it)")
}
if dsn == "" {
log.Fatal("sam-sync: set MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL)")
}
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
log.Fatalf("sam-sync: db connect: %v", err)
}
defer pool.Close()
if err := pool.Ping(ctx); err != nil {
log.Fatalf("sam-sync: db ping: %v", err)
}
// sam_opportunities is a global catalog the application role may only read, so this writer needs the
// BYPASSRLS migrator DSN; assert it now for a legible failure instead of a mid run permission denied.
if err := database.AssertMigratorRole(ctx, pool); err != nil {
log.Fatalf("sam-sync: %v", err)
}
q := dbgen.New(pool)
client := samgov.New(base, key)
profile := samgov.Galapagos
now := time.Now()
from := now.AddDate(0, 0, -profile.LookbackDays)
log.Printf("sam-sync: profile=%q naics=%v window=%s..%s base=%s",
profile.Name, profile.NAICS, from.Format("2006-01-02"), now.Format("2006-01-02"), baseLabel(base))
grand := 0
throttled := false
// runHash accumulates the identity of everything this run ingested, in the deterministic order the
// profile iterates, so an unchanged upstream reruns to the same manifest hash.
runHash := sha256.New()
// seen retains the normalized notices this run upserted, keyed by notice id, so the amendment pass can
// compare each against the version a tracking tenant already holds without refetching from SAM.
seen := make(map[string]samgov.Normalized)
for _, naics := range profile.NAICS {
recs, total, capped, serr := client.SearchAll(ctx, samgov.SearchParams{
PostedFrom: from, PostedTo: now, NAICS: naics, Limit: 1000,
}, pageCap)
// Persist whatever we gathered, even on a partial/throttled result.
upserted := 0
for _, rec := range recs {
n := samgov.Normalize(rec)
if n.NoticeID == "" {
continue
}
if err := q.UpsertSamOpportunity(ctx, toParams(n)); err != nil {
log.Printf("sam-sync: upsert %s: %v", n.NoticeID, err)
continue
}
// The notice id plus its raw payload identifies both WHICH notices this run covered and whether
// any of their contents changed upstream.
_, _ = io.WriteString(runHash, n.NoticeID)
_, _ = runHash.Write([]byte(n.Raw))
seen[n.NoticeID] = n
upserted++
}
grand += upserted
if serr != nil {
if errors.Is(serr, samgov.ErrThrottled) {
var te *samgov.ThrottleError
errors.As(serr, &te)
log.Printf("sam-sync: NAICS %s: 429 throttled after %d upserts (retry after %s), stopping run gracefully",
naics, upserted, retryAfter(te))
throttled = true
break
}
log.Printf("sam-sync: NAICS %s: error after %d upserts: %v", naics, upserted, serr)
continue
}
if capped {
log.Printf("sam-sync: NAICS %s: upserted %d (page cap hit, %d available, not all retrieved)", naics, upserted, total)
} else {
log.Printf("sam-sync: NAICS %s: upserted %d of %d available", naics, upserted, total)
}
}
total := int64(-1)
if meta, err := q.SamSyncMeta(ctx); err == nil {
total = meta.Total
}
tail := ""
if throttled {
tail = " (run stopped early on 429)"
}
// Provenance: every external source records what it pulled, when, under what terms, and a hash of the
// run, so a synced row can always be traced back to the fetch that produced it. The manifest write must
// never fail the run: the opportunities are already committed, and losing the trace of a good sync is a
// smaller harm than a job that reports failure and gets rerun against a rate limited API.
manifestID, err := writeManifest(ctx, pool, base, grand, runHash)
if err != nil {
log.Printf("sam-sync: WARNING could not record ingest_manifest provenance: %v", err)
}
// Amendment detection, directive Section 4 steps 6, 7, and 11. A notice whose source payload changed
// since the version a tracking tenant holds produces a new immutable version and an alert. Tenants that
// track nothing in this run cost one indexed lookup and no writes.
nv, na, nf := applyAmendments(ctx, pool, seen, manifestID)
if nv > 0 {
log.Printf("sam-sync: amendments: wrote %d new opportunity versions and raised %d alerts", nv, na)
}
log.Printf("sam-sync: DONE, upserted %d this run; store now holds %d opportunities%s", grand, total, tail)
// FAIL LAST, finding D9. The catalog upserts are already committed and every other tenant's amendment has
// already been written, so exiting here costs nothing that succeeded. What it buys is that an unrecorded
// amendment reaches a human: this unit carries OnFailure, so a nonzero exit sends mail, and the previous
// behaviour of logging and returning success is exactly why a stuck version counter ran unnoticed from
// 27 July until someone happened to read the journal.
if nf > 0 {
log.Fatalf("sam-sync: %d amendment(s) FAILED to record. An amendment that does not land cannot flag an "+
"affected approved requirement, which is the guarantee this pipeline exists to provide.", nf)
}
}
// writeManifest records one ingest_manifest row per run. The content hash covers the notice ids and their
// source payload hashes in a stable order, so an identical rerun produces an identical hash and a changed
// upstream produces a different one, which is what makes the manifest useful for idempotence rather than
// just an audit line.
// It returns the manifest id so every version row this run writes cites the fetch that produced it.
func writeManifest(ctx context.Context, pool *pgxpool.Pool, base string, count int, h hash.Hash) (string, error) {
const samLicense = "US federal government public data, SAM.gov terms of use"
var id string
err := pool.QueryRow(ctx,
`INSERT INTO ingest_manifest (source, url, license, fetched_at, record_count, content_hash)
VALUES ($1, $2, $3, now(), $4, $5) RETURNING id::text`,
"SAM.gov", baseLabel(base), samLicense, count, hex.EncodeToString(h.Sum(nil))).Scan(&id)
return id, err
}
// toParams bridges a pure samgov.Normalized into the dbgen upsert params (Go-native -> pg types).
func toParams(n samgov.Normalized) dbgen.UpsertSamOpportunityParams {
var posted pgtype.Date
if n.PostedDate != nil {
posted = pgtype.Date{Time: *n.PostedDate, Valid: true}
}
return dbgen.UpsertSamOpportunityParams{
NoticeID: n.NoticeID,
Title: n.Title,
SolicitationNumber: n.SolicitationNumber,
Department: n.Department,
FullParentPath: n.FullParentPath,
SubAgency: n.SubAgency,
NaicsCode: n.NaicsCode,
PscCode: n.PSCCode,
SetAsideCode: n.SetAsideCode,
SetAside: n.SetAside,
NoticeType: n.NoticeType,
BaseType: n.BaseType,
PostedDate: posted,
ResponseDeadline: n.ResponseDeadline,
Active: n.Active,
PopCity: n.POPCity,
PopState: n.POPState,
PopZip: n.POPZip,
UiLink: n.UILink,
DescriptionUrl: n.DescriptionURL,
ResourceLinks: []byte(n.ResourceLinks),
Poc: []byte(n.POC),
Raw: []byte(n.Raw),
Source: n.Source,
}
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if s := strings.TrimSpace(v); s != "" {
return s
}
}
return ""
}
// baseLabel returns a safe label for logs (the base URL never carries the api_key).
func baseLabel(base string) string {
switch {
case base == "" || base == samgov.BaseProd:
return "prod"
case base == samgov.BaseAlpha:
return "alpha"
default:
return base
}
}
func retryAfter(te *samgov.ThrottleError) string {
if te == nil || te.RetryAfter == "" {
return "unknown"
}
return te.RetryAfter
}