Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/sam-attachment-sync/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/sam-attachment-sync/main.go |
| Original SHA-256 | e6043327402f4ccc0c335fdf89dd4887ae2d6481a399e27ac712265042021a98 |
| Git blob SHA | 31a96a6ae9c133d08f6b4ad239d62bb69cf625ee |
| Reading-copy SHA-256 | e6043327402f4ccc0c335fdf89dd4887ae2d6481a399e27ac712265042021a98 |
Recovered source transcript
Download complete source text ↓568 lines · 20.4 KB
// Command sam-attachment-sync pulls the public attachment package for tracked SAM.gov opportunities.
//
// This is the deferred Phase 1 item. It was recorded as blocked on documents existing; the document trial
// rehearsal showed that was only half true, and this is the other half.
//
// IT SPENDS NO API KEY BUDGET. The two endpoints it uses are the PUBLIC resource endpoints and take no key:
//
// list https://sam.gov/api/prod/opps/v3/opportunities/{noticeId}/resources
// file https://sam.gov/api/prod/opps/v3/opportunities/resources/files/{resourceId}/download
//
// The keyed api.sam.gov endpoints are deliberately NOT used here. sam-sync already spends the key budget on
// notices, and an attachment crawl is exactly the workload that would exhaust a daily quota and take the
// notice feed down with it. Recorded because the temptation to reuse the existing keyed client is obvious.
//
// PROVENANCE IS BORN public_cache. A SAM attachment is a published federal document. Migration 0081 stamps
// it, and the INSERT below omits the column so that is visible at the call site rather than assumed.
//
// IT IS POLITE AND BOUNDED. Roughly one request per second, retries with backoff and a ceiling, a per file
// size cap, a per run daily cap, and a sha256 dedupe against what is already attached. Anything it cannot
// have it SKIPS AND RECORDS rather than failing the run: one unavailable attachment must not stop the other
// forty.
//
// TIER ONE, which is what this implements: every tracked opportunity pulls its full package on first
// tracking and on any amendment. Tier two, the rolling corpus harvest by NAICS, is booked separately in
// docs/IMPROVEMENTS.md and is NOT in this binary.
//
// Usage:
//
// sam-attachment-sync -dsn <dsn> [-limit 25] [-max-file-mb 32] [-daily-cap 500] [-dry-run]
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"flag"
"fmt"
"io"
"log"
"net/http"
"os"
"strings"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/missionworkforce/missionmatch/internal/platform/storage"
)
const (
resourcesURL = "https://sam.gov/api/prod/opps/v3/opportunities/%s/resources"
downloadURL = "https://sam.gov/api/prod/opps/v3/opportunities/resources/files/%s/download"
userAgent = "missionMATCH/1.0 (federal opportunity intake; contact via sam.gov registration)"
)
// Skip reasons are a closed vocabulary, so a run's skips can be counted by kind rather than read as prose.
const (
skipTooLarge = "too_large"
skipDuplicate = "duplicate_sha256"
skipUnavailable = "unavailable"
skipDailyCap = "daily_cap_reached"
skipBadType = "media_type_not_allowed"
skipExportControlled = "export_controlled"
skipNotPublic = "not_public_access"
skipNotAFile = "not_a_file"
)
// allowedMedia mirrors the intake allowlist. Kept as its own map on purpose: this job pulls from a source
// nobody here controls, and silently widening what the app accepts because a federal site served something
// unusual is how an allowlist stops meaning anything.
var allowedMedia = map[string]string{
"application/pdf": ".pdf",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
"application/msword": ".doc",
"application/vnd.ms-excel": ".xls",
"text/plain": ".txt",
"text/html": ".html",
"application/zip": ".zip",
"application/rtf": ".rtf",
}
func main() {
dsn := flag.String("dsn", "", "migrator DSN, defaults to MM_MIGRATOR_DATABASE_URL")
storeRoot := flag.String("storage-root", os.Getenv("MM_STORAGE_ROOT"), "object storage root")
limit := flag.Int("limit", 25, "how many tracked opportunities to consider in this run")
maxFileMB := flag.Int("max-file-mb", 32, "per file size cap in MiB")
dailyCap := flag.Int("daily-cap", 500, "maximum files this run will download")
rate := flag.Duration("rate", time.Second, "minimum interval between requests to sam.gov")
dryRun := flag.Bool("dry-run", false, "list what would be pulled, download nothing")
flag.Parse()
if *dsn == "" {
*dsn = os.Getenv("MM_MIGRATOR_DATABASE_URL")
}
if *dsn == "" {
log.Fatal("sam-attachment-sync: -dsn or MM_MIGRATOR_DATABASE_URL is required")
}
if *storeRoot == "" && !*dryRun {
log.Fatal("sam-attachment-sync: -storage-root or MM_STORAGE_ROOT is required unless -dry-run")
}
var store storage.Storer
if !*dryRun {
st, err := storage.NewLocal(*storeRoot)
if err != nil {
log.Fatalf("sam-attachment-sync: storage: %v", err)
}
store = st
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, *dsn)
if err != nil {
log.Fatalf("sam-attachment-sync: pool: %v", err)
}
defer pool.Close()
targets, err := selectTargets(ctx, pool, *limit)
if err != nil {
log.Fatalf("sam-attachment-sync: select: %v", err)
}
if len(targets) == 0 {
log.Print("sam-attachment-sync: no tracked opportunity needs its package pulled")
return
}
log.Printf("sam-attachment-sync: %d opportunity(ies), dry_run=%t, rate=%s, cap=%d files, max %d MiB each",
len(targets), *dryRun, *rate, *dailyCap, *maxFileMB)
c := &crawler{
http: &http.Client{Timeout: 90 * time.Second},
rate: *rate,
maxBytes: int64(*maxFileMB) << 20,
remaining: *dailyCap,
store: store,
dry: *dryRun,
skips: map[string]int{},
}
var stored, considered int
for _, t := range targets {
n, seen, err := c.pullPackage(ctx, pool, t)
if err != nil {
// One opportunity failing is not the run failing. Recorded and the crawl continues, because the
// alternative leaves every later opportunity unpulled to report a single problem.
log.Printf(" %s: SKIPPED, %v", t.NoticeID, err)
c.skips[skipUnavailable]++
continue
}
stored += n
considered += seen
log.Printf(" %s: %d/%d file(s) stored", t.NoticeID, n, seen)
}
log.Printf("sam-attachment-sync: DONE stored=%d considered=%d", stored, considered)
for reason, n := range c.skips {
log.Printf(" skipped %-22s %d", reason, n)
}
if c.remaining <= 0 {
log.Printf(" NOTE: the daily cap of %d was reached, so this run is INCOMPLETE by design. "+
"Remaining files will be pulled on the next run.", *dailyCap)
}
if !*dryRun {
restricted := c.skips[skipExportControlled] + c.skips[skipNotPublic]
if restricted > 0 {
log.Printf(" REFUSED %d restricted attachment(s). This is policy, not a failure, and it is "+
"recorded in ingest_manifest.", restricted)
}
if err := writeManifest(ctx, pool, stored, considered, restricted); err != nil {
log.Printf("sam-attachment-sync: manifest write failed: %v", err)
}
}
}
type target struct{ OrgID, OppID, NoticeID string }
// selectTargets picks tracked SAM opportunities whose package has not been pulled, plus any whose notice has
// been amended since the last pull. The amendment case is the tier one requirement: a package is not a one
// time fetch, it is refreshed whenever the solicitation moves.
func selectTargets(ctx context.Context, pool *pgxpool.Pool, limit int) ([]target, error) {
rows, err := pool.Query(ctx, `
SELECT t.organization_id::text, t.id::text, t.sam_notice_id
FROM tracked_opportunities t
JOIN sam_opportunities s ON s.notice_id = t.sam_notice_id
WHERE t.source_type = 'sam_gov' AND t.sam_notice_id <> ''
AND (
NOT EXISTS (SELECT 1 FROM opportunity_documents d
WHERE d.organization_id = t.organization_id AND d.opportunity_id = t.id
AND d.source_type = 'sam_attachment')
OR s.synced_at > COALESCE((SELECT max(d2.created_at) FROM opportunity_documents d2
WHERE d2.organization_id = t.organization_id
AND d2.opportunity_id = t.id
AND d2.source_type = 'sam_attachment'), 'epoch')
)
ORDER BY t.created_at DESC
LIMIT $1`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var out []target
for rows.Next() {
var t target
if e := rows.Scan(&t.OrgID, &t.OppID, &t.NoticeID); e != nil {
return nil, e
}
out = append(out, t)
}
return out, rows.Err()
}
type crawler struct {
http *http.Client
rate time.Duration
last time.Time
maxBytes int64
remaining int
store storage.Storer
dry bool
skips map[string]int
}
// wait enforces the request interval. Politeness to a public federal endpoint is not optional: a crawler
// that hammers sam.gov gets the whole platform blocked, and there is no key to rotate to.
func (c *crawler) wait() {
if d := c.rate - time.Since(c.last); d > 0 {
time.Sleep(d)
}
c.last = time.Now()
}
// get retries with backoff and a ceiling. It retries a 429 and a 5xx, and it does NOT retry a 404, because
// a missing attachment is an answer rather than a failure.
func (c *crawler) get(ctx context.Context, url string) (*http.Response, error) {
var lastErr error
for attempt := 0; attempt < 4; attempt++ {
if attempt > 0 {
time.Sleep(time.Duration(1<<attempt) * time.Second)
}
c.wait()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", userAgent)
req.Header.Set("Accept", "*/*")
resp, err := c.http.Do(req)
if err != nil {
lastErr = err
continue
}
if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
resp.Body.Close()
return nil, fmt.Errorf("not available (%d)", resp.StatusCode)
}
if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 {
resp.Body.Close()
lastErr = fmt.Errorf("status %d", resp.StatusCode)
continue
}
if resp.StatusCode != http.StatusOK {
resp.Body.Close()
return nil, fmt.Errorf("status %d", resp.StatusCode)
}
return resp, nil
}
return nil, fmt.Errorf("gave up after 4 attempts: %w", lastErr)
}
// resource is one attachment as the public endpoint ACTUALLY describes it, verified against two live
// notices rather than assumed from documentation.
//
// TWO THINGS WERE WRONG IN THE FIRST VERSION AND ONLY THE LIVE CALL FOUND THEM:
//
// the list is nested, _embedded.opportunityAttachmentList[].attachments[], not a flat array
// mimeType carries an EXTENSION (".pdf"), not a media type, so matching it against a MIME allowlist
// silently refused every real attachment
//
// It also carries access control fields that the first version ignored entirely. For a product whose users
// hold clearances, pulling an EXPORT CONTROLLED attachment into a tenant corpus is not a bug to fix later.
type resource struct {
ID string `json:"resourceId"`
AttachmentID string `json:"attachmentId"`
Name string `json:"name"`
Type string `json:"type"`
MimeType string `json:"mimeType"`
Size int64 `json:"size"`
AccessLevel string `json:"accessLevel"`
AccessStatus string `json:"accessStatus"`
ExportControlled string `json:"exportControlled"`
ExplicitAccess string `json:"explicitAccess"`
FileExists string `json:"fileExists"`
DeletedFlag string `json:"deletedFlag"`
}
// mayPull decides whether an attachment is one this platform is permitted to hold.
//
// EXPORT CONTROL POLICY, recorded 28 July 2026 as a standing rule and not a configuration:
//
// an attachment flagged export controlled, or carrying a restricted access level, is REFUSED BY DEFAULT
// the refusal is recorded in ingest_manifest as a skipped restricted count, so a run can always say how
// many it declined rather than leaving the number invisible
// the corpus harvest NEVER pulls one under any configuration. There is deliberately no flag, no
// environment variable, and no argument that changes this, because a switch that exists gets flipped
// tracked opportunity pulls honour the SAME default, so the two paths cannot diverge
// any future exception is a PER TENANT LEGAL DECISION made by the owner, not a config toggle
//
// It FAILS CLOSED: an unexpected value in any of these fields is treated as not permitted, because the cost
// of wrongly skipping a public attachment is one missing file and the cost of wrongly pulling a controlled
// one is a compliance incident on a cleared workforce product.
func (r resource) mayPull() (bool, string) {
// A resource can be a LINK to somewhere else rather than a file. One live notice returned four of them,
// all with an empty name and size 0, and they were reported as media_type_not_allowed, which reads as
// "the allowlist is too narrow" and would tempt the next person to widen it for no reason. A misleading
// skip reason is worse than a blunt one because it prompts the wrong repair.
if r.Type != "" && !strings.EqualFold(r.Type, "file") {
return false, skipNotAFile
}
if r.ExportControlled != "" && r.ExportControlled != "0" {
return false, skipExportControlled
}
if r.ExplicitAccess != "" && r.ExplicitAccess != "0" {
return false, skipNotPublic
}
if r.AccessLevel != "" && !strings.EqualFold(r.AccessLevel, "public") {
return false, skipNotPublic
}
if r.AccessStatus != "" && !strings.EqualFold(r.AccessStatus, "public") {
return false, skipNotPublic
}
if r.DeletedFlag != "" && r.DeletedFlag != "0" {
return false, skipUnavailable
}
if r.FileExists != "" && r.FileExists != "1" {
return false, skipUnavailable
}
return true, ""
}
func (c *crawler) listResources(ctx context.Context, noticeID string) ([]resource, error) {
resp, err := c.get(ctx, fmt.Sprintf(resourcesURL, noticeID))
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
if err != nil {
return nil, err
}
// Two shapes have been seen in the wild, a bare array and an envelope. Both are handled rather than
// assuming the one that happened to be served the day this was written.
var direct []resource
if json.Unmarshal(body, &direct) == nil && len(direct) > 0 {
return direct, nil
}
var env struct {
Embedded struct {
List []struct {
Attachments []resource `json:"attachments"`
} `json:"opportunityAttachmentList"`
} `json:"_embedded"`
Resources []resource `json:"resources"`
}
if err := json.Unmarshal(body, &env); err != nil {
return nil, fmt.Errorf("resource list was not json this job understands: %w", err)
}
if len(env.Resources) > 0 {
return env.Resources, nil
}
var out []resource
for _, group := range env.Embedded.List {
out = append(out, group.Attachments...)
}
return out, nil
}
func (c *crawler) pullPackage(ctx context.Context, pool *pgxpool.Pool, t target) (int, int, error) {
res, err := c.listResources(ctx, t.NoticeID)
if err != nil {
return 0, 0, err
}
stored := 0
for _, r := range res {
if strings.TrimSpace(r.ID) == "" {
c.skips[skipUnavailable]++
continue
}
if c.remaining <= 0 {
c.skips[skipDailyCap]++
continue
}
if ok, reason := r.mayPull(); !ok {
c.skips[reason]++
log.Printf(" %s: skipped, %s", r.Name, reason)
continue
}
if r.Size > c.maxBytes {
c.skips[skipTooLarge]++
continue
}
mt := normalizeMedia(r.MimeType, r.Name)
if _, ok := allowedMedia[mt]; !ok {
c.skips[skipBadType]++
continue
}
if c.dry {
log.Printf(" would pull %s (%s, %d bytes)", r.Name, mt, r.Size)
stored++
c.remaining--
continue
}
n, err := c.fetchAndStore(ctx, pool, t, r, mt)
if err != nil {
log.Printf(" %s: skipped, %v", r.Name, err)
c.skips[skipUnavailable]++
continue
}
stored += n
}
return stored, len(res), nil
}
func (c *crawler) fetchAndStore(ctx context.Context, pool *pgxpool.Pool, t target, r resource, mt string) (int, error) {
resp, err := c.get(ctx, fmt.Sprintf(downloadURL, r.ID))
if err != nil {
return 0, err
}
defer resp.Body.Close()
// One byte of headroom so a file at exactly the cap is distinguishable from one over it.
body, err := io.ReadAll(io.LimitReader(resp.Body, c.maxBytes+1))
if err != nil {
return 0, err
}
c.remaining--
if int64(len(body)) > c.maxBytes {
c.skips[skipTooLarge]++
return 0, nil
}
if len(body) == 0 {
c.skips[skipUnavailable]++
return 0, nil
}
sum := sha256.Sum256(body)
hexSum := hex.EncodeToString(sum[:])
var exists bool
if err := pool.QueryRow(ctx,
`SELECT EXISTS (SELECT 1 FROM opportunity_documents
WHERE organization_id=$1 AND opportunity_id=$2 AND sha256=$3)`,
t.OrgID, t.OppID, hexSum).Scan(&exists); err != nil {
return 0, err
}
if exists {
c.skips[skipDuplicate]++
return 0, nil
}
key := fmt.Sprintf("org/%s/opportunity/%s/sam/%s%s", t.OrgID, t.OppID, hexSum, allowedMedia[mt])
if err := c.store.Put(ctx, key, body); err != nil {
return 0, err
}
tx, err := pool.Begin(ctx)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback(ctx) }()
if _, err := tx.Exec(ctx, `SELECT set_config('app.current_org_id', $1, true)`, t.OrgID); err != nil {
return 0, err
}
// PROVENANCE OMITTED DELIBERATELY: migration 0081 stamps public_cache on a sam_attachment row.
if _, err := tx.Exec(ctx, `
INSERT INTO opportunity_documents
(organization_id, opportunity_id, document_type, file_name, media_type, storage_key,
size_bytes, sha256, source_url, source_type, extraction_status)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,'sam_attachment','pending')`,
t.OrgID, t.OppID, documentTypeFor(r.Name, r.Type), baseName(r.Name), mt, key,
int64(len(body)), hexSum, fmt.Sprintf(downloadURL, r.ID)); err != nil {
return 0, err
}
if err := tx.Commit(ctx); err != nil {
return 0, err
}
return 1, nil
}
// documentTypeFor makes a conservative guess from the file name and stops at "attachment" when it does not
// know. Guessing "pws" wrongly puts a pricing sheet in front of a reviewer as a statement of work.
func documentTypeFor(name, declared string) string {
n := strings.ToLower(name + " " + declared)
switch {
case strings.Contains(n, "amendment"), strings.Contains(n, "amd"):
return "amendment"
case strings.Contains(n, "pws"), strings.Contains(n, "performance work"):
return "pws"
case strings.Contains(n, "sow"), strings.Contains(n, "statement of work"):
return "sow"
case strings.Contains(n, "soo"):
return "soo"
case strings.Contains(n, "solicitation"), strings.Contains(n, "rfp"), strings.Contains(n, "rfq"):
return "solicitation"
case strings.Contains(n, "wage"), strings.Contains(n, "labor categor"):
return "labor_categories"
case strings.Contains(n, "pricing"), strings.Contains(n, "price"):
return "pricing_template"
case strings.Contains(n, "evaluation"):
return "evaluation_criteria"
case strings.Contains(n, "q&a"), strings.Contains(n, "questions"):
return "q_and_a"
default:
return "attachment"
}
}
// normalizeMedia handles BOTH shapes this endpoint returns. sam.gov puts an EXTENSION in mimeType (".pdf"),
// which is not a media type at all, and the first version compared it against a MIME allowlist and refused
// every real attachment while looking like it was working.
func normalizeMedia(declared, name string) string {
d := strings.ToLower(strings.TrimSpace(strings.SplitN(declared, ";", 2)[0]))
if _, ok := allowedMedia[d]; ok {
return d
}
if strings.HasPrefix(d, ".") {
for mt, ext := range allowedMedia {
if ext == d {
return mt
}
}
}
lower := strings.ToLower(name)
for mt, ext := range allowedMedia {
if strings.HasSuffix(lower, ext) {
return mt
}
}
return d
}
func baseName(p string) string {
if i := strings.LastIndexAny(p, `/\`); i >= 0 {
return p[i+1:]
}
return p
}
// writeManifest records the pull in the provenance ledger. EVERY external pull writes one, so the corpus can
// always answer where a document came from and when.
//
// THE RESTRICTED COUNT IS PART OF THE RECORD. A run that declined twelve export controlled attachments and
// a run that saw none look identical without it, and the difference is exactly what an assessor would ask
// about. It goes in the source label so it survives in a table that has no column for it, which is uglier
// than a column and considerably better than losing the number.
func writeManifest(ctx context.Context, pool *pgxpool.Pool, stored, considered, restricted int) error {
sum := sha256.Sum256([]byte(fmt.Sprintf("sam-attachment-sync:%d:%d:%d", stored, considered, restricted)))
source := "SAM.gov attachments"
if restricted > 0 {
source = fmt.Sprintf("SAM.gov attachments (skipped restricted: %d)", restricted)
}
_, err := pool.Exec(ctx, `
INSERT INTO ingest_manifest (source, url, license, fetched_at, record_count, content_hash)
VALUES ($4, $1, 'US Government work, public domain', now(), $2, $3)`,
"https://sam.gov/api/prod/opps/v3/opportunities/{id}/resources", stored,
hex.EncodeToString(sum[:]), source)
return err
}