Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/corpus-harvest/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/corpus-harvest/main.go |
| Original SHA-256 | 0941f0ed5f8be1cf5b71e42145568e25a988d802b2b35111bf759791a140ee23 |
| Git blob SHA | 4f700ec9c5b6ad5481bf6effca50c6e558ff8e49 |
| Reading-copy SHA-256 | 0941f0ed5f8be1cf5b71e42145568e25a988d802b2b35111bf759791a140ee23 |
Recovered source transcript
Download complete source text ↓637 lines · 24.8 KB
// Command corpus-harvest builds missionMATCH's own knowledge of federal opportunities from the documents
// attached to them. It has two roles and the same binary runs both.
//
// WHY TWO ROLES RATHER THAN ONE LOOP. The two halves of this job have opposite constraints.
//
// FETCH is not where the computing happens. It is a tap paced by politeness: one request per second to
// sam.gov, and every device in this federation egresses through a single public address, so running it
// on three machines would present as three requests per second from one IP and earn a block rather than
// a speedup. Exactly one fetcher runs. It is not a bottleneck worth parallelising because it is not
// doing any work, it is waiting.
//
// EMBED is where the computing happens, and it is fully liquid. Any number of workers on any number of
// machines claim chunks with FOR UPDATE SKIP LOCKED, which Postgres makes atomic without an orchestrator,
// a scheduler or a heartbeat. A node under load claims fewer rows; an idle node claims more. That is self
// balancing by construction rather than by policy, and a node that dies mid claim simply loses its
// transaction and the rows return to the queue.
//
// THE BINARY IS NEVER RETAINED. Measured against sam.gov, a notice carries 2.5 files and 2.3 MB on average,
// so the active federal landscape is roughly 69 GB of PDF. Keeping it would buy nothing: SAM is a permanent
// public store with stable urls, so the durable copy already exists. The fetcher downloads to memory,
// extracts, writes the prose and its citations, and drops the bytes. sha256 and source_url are recorded on
// every row so any original refetches and can be proved identical to what was learned from.
//
// EACH NODE EMBEDS AGAINST ITS OWN LOCAL MODEL. M4 and VALKYRIE are Apple Silicon and run the model on
// Metal; megamind-agi has no GPU and contributes at CPU speed. Pointing every worker at one endpoint would
// recreate the bottleneck the federation exists to remove, so the endpoint is per node and defaults to
// localhost.
//
// Env: MM_MIGRATOR_DATABASE_URL (writes the shared corpus), MM_EMBED_ENDPOINT (defaults to the local
// ollama), MM_EMBED_MODEL, MM_EMBED_DIMENSION.
//
// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build.
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/modules/opportunities"
"github.com/missionworkforce/missionmatch/internal/platform/docextract"
"github.com/missionworkforce/missionmatch/internal/platform/embeddings"
)
const (
// resourceListURL and fileURL are the PUBLIC sam.gov endpoints. They take no API key, which is the
// whole reason an attachment crawl is affordable: sam-sync spends the keyed quota on notices, and an
// attachment crawl is exactly the workload that would exhaust a daily quota and take the notice feed
// down with it.
resourceListURL = "https://sam.gov/api/prod/opps/v3/opportunities/%s/resources"
fileURL = "https://sam.gov/api/prod/opps/v3/opportunities/resources/files/%s/download"
// maxFileBytes refuses a file rather than pulling a 400 MB drawing set into memory to find it has no
// prose in it. Recorded as too_large so it is a decision on the record, not a silent skip.
maxFileBytes = 32 << 20
// embedBatch is how many chunks one worker claims at a time. Small on purpose: a big claim held by a
// slow node is work the fast nodes cannot take, which is the opposite of liquid.
embedBatch = 32
// embedRetries is how many consecutive failures a worker tolerates before it gives up and lets the
// other nodes carry the queue. Five is enough to ride out a link that blinks and few enough that a
// genuinely unreachable database still ends the run rather than looping forever.
embedRetries = 5
)
func main() {
role := flag.String("role", "embed", "fetch (one node only, paced) or embed (any number of nodes)")
limit := flag.Int("limit", 0, "stop after this many notices (fetch) or chunks (embed), 0 means until dry")
openOnly := flag.Bool("open-only", true, "fetch only notices whose response deadline has not passed")
rate := flag.Duration("rate", time.Second, "minimum interval between sam.gov requests")
dryRun := flag.Bool("dry-run", false, "fetch: list and size the work, download nothing")
flag.Parse()
dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL"))
if dsn == "" {
log.Fatal("corpus-harvest: MM_MIGRATOR_DATABASE_URL is required")
}
ctx, cancel := context.WithTimeout(context.Background(), 12*time.Hour)
defer cancel()
// THE POOL IS CAPPED AT TWO BECAUSE THAT IS WHAT THIS WORKER NEEDS, and for no other reason.
//
// pgxpool defaults to the greater of four and the CPU count, so an eight core box opens eight
// connections to claim one batch at a time inside one transaction. Two is honest sizing: one working
// connection, and one so a retry does not wait for it to be returned.
//
// It is recorded plainly that capping the pool was once believed to be the fix for the unreachable
// wired path. It was not. A single detached run succeeded with the cap at one and causation was read
// into one observation; the same run failed at two, and failed again with nothing else on the network.
// The real cause was the interface scoped route, and it is addressed by the bound dialer below.
cfg, err := pgxpool.ParseConfig(dsn)
if err != nil {
log.Fatalf("corpus-harvest: parse dsn: %v", err)
}
cfg.MaxConns = 2
cfg.MinConns = 0
// MM_BIND_INTERFACE names the wired interface this node should reach the federation on. Setting it
// makes every socket this command opens carry IP_BOUND_IF, which is what lets it use the interface
// scoped route macOS installs when a machine has both a wired default and a Tailscale tunnel. Leaving
// it unset is the ordinary case and changes nothing.
bindIface := strings.TrimSpace(os.Getenv("MM_BIND_INTERFACE"))
dialer, err := bindToInterface(bindIface)
if err != nil {
log.Fatalf("corpus-harvest: %v", err)
}
if dialer != nil {
cfg.ConnConfig.DialFunc = dialer
log.Printf("corpus-harvest: sockets bound to %s", bindIface)
}
pool, err := pgxpool.NewWithConfig(ctx, cfg)
if err != nil {
log.Fatalf("corpus-harvest: pool: %v", err)
}
defer pool.Close()
host, _ := os.Hostname()
switch *role {
case "fetch":
if err := runFetch(ctx, pool, host, *limit, *openOnly, *rate, *dryRun); err != nil {
log.Fatalf("corpus-harvest: fetch: %v", err)
}
case "embed":
if err := runEmbed(ctx, pool, host, *limit); err != nil {
log.Fatalf("corpus-harvest: embed: %v", err)
}
default:
log.Fatalf("corpus-harvest: unknown role %q, want fetch or embed", *role)
}
}
// ---------------------------------------------------------------------------------------------------
// FETCH: one node, paced, downloads nothing twice and keeps nothing.
// ---------------------------------------------------------------------------------------------------
// resource is one attachment as sam.gov actually describes it.
//
// THE FIELDS BEYOND THE IDENTIFIERS ARE GUARDS, NOT DECORATION. A first cut of this took the list at face
// value and produced four rows whose resource id was empty, because the payload nests: the
// opportunityAttachmentList is a list of GROUPS, each carrying its own attachments array. That failure was
// visible only by running it against sam.gov, and the empty download urls it wrote are the reason the
// parsing is now pinned to the real shape rather than an assumed one.
type resource struct {
ResourceID string `json:"resourceId"`
Name string `json:"name"`
Type string `json:"type"`
MimeType string `json:"mimeType"`
Size int64 `json:"size"`
FileExists string `json:"fileExists"`
DeletedFlag string `json:"deletedFlag"`
AccessLevel string `json:"accessLevel"`
AccessStatus string `json:"accessStatus"`
ExportControlled string `json:"exportControlled"`
}
// fetchable decides whether this attachment may be downloaded at all.
//
// EXPORT CONTROLLED MATERIAL IS NEVER FETCHED, and that is the load bearing line in this function. sam.gov
// marks some attachments as export controlled or non public, and a cleared workforce product pulling ITAR
// or EAR restricted documents into a corpus would be a serious problem regardless of how the crawl was
// intended. The default is REFUSE: anything that does not positively say it is public and unrestricted is
// left alone.
func (r resource) fetchable() (bool, string) {
if r.ResourceID == "" {
return false, "no resource id"
}
if r.ExportControlled != "" && r.ExportControlled != "0" {
return false, "export controlled, deliberately not fetched"
}
if r.AccessLevel != "" && !strings.EqualFold(r.AccessLevel, "public") {
return false, "not a public attachment"
}
if r.AccessStatus != "" && !strings.EqualFold(r.AccessStatus, "public") {
return false, "not publicly released"
}
if r.DeletedFlag == "1" {
return false, "withdrawn by the issuing office"
}
if r.FileExists == "0" {
return false, "the file is gone from sam.gov"
}
return true, ""
}
func runFetch(ctx context.Context, pool *pgxpool.Pool, host string, limit int, openOnly bool, rate time.Duration, dryRun bool) error {
// Notices that have never been looked at. A notice with any corpus_documents row has been visited, so
// a rerun continues rather than starting over, which matters on a crawl measured in days.
q := `
SELECT o.id::text, o.source_record_id, coalesce(o.title,'')
FROM opportunity_records o
WHERE o.source_record_id IS NOT NULL AND o.source_record_id <> ''
AND NOT EXISTS (SELECT 1 FROM corpus_documents d WHERE d.opportunity_record_id = o.id)`
if openOnly {
q += ` AND o.response_deadline >= now()`
}
q += ` ORDER BY o.response_deadline NULLS LAST`
if limit > 0 {
q += fmt.Sprintf(" LIMIT %d", limit)
}
rows, err := pool.Query(ctx, q)
if err != nil {
return err
}
type target struct{ id, notice, title string }
var targets []target
for rows.Next() {
var t target
if err := rows.Scan(&t.id, &t.notice, &t.title); err != nil {
rows.Close()
return err
}
targets = append(targets, t)
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
log.Printf("corpus-harvest: fetch on %s, %d notice(s) unvisited, open_only=%t rate=%s dry_run=%t",
host, len(targets), openOnly, rate, dryRun)
if len(targets) == 0 {
return nil
}
client := &http.Client{Timeout: 90 * time.Second}
var files, chunks, skipped int
var bytesIn int64
tick := time.NewTicker(rate)
defer tick.Stop()
for i, t := range targets {
<-tick.C
res, err := listResources(ctx, client, t.notice)
if err != nil {
log.Printf("corpus-harvest: notice %s resource list failed: %v", t.notice, err)
continue
}
if len(res) == 0 {
// A notice with no attachments is a FACT worth recording, not an absence to retry forever.
// One row with status skipped stops this notice costing a request on every future pass.
if !dryRun {
if err := recordNoAttachments(ctx, pool, t.id); err != nil {
log.Printf("corpus-harvest: notice %s: %v", t.notice, err)
}
}
continue
}
for _, r := range res {
// The guard runs BEFORE the request, so a restricted attachment costs sam.gov nothing and
// never touches this machine. The refusal is recorded so the corpus can account for what it
// deliberately did not read.
if ok, why := r.fetchable(); !ok {
if !dryRun {
if err := recordRefusal(ctx, pool, t.id, r, why); err != nil {
log.Printf("corpus-harvest: notice %s refusal: %v", t.notice, err)
}
}
skipped++
continue
}
if dryRun {
files++
bytesIn += r.Size
continue
}
<-tick.C
n, c, err := fetchOne(ctx, pool, client, t.id, t.notice, r)
if err != nil {
log.Printf("corpus-harvest: notice %s file %s: %v", t.notice, r.ResourceID, err)
continue
}
files++
bytesIn += n
chunks += c
}
if (i+1)%25 == 0 {
log.Printf("corpus-harvest: %d/%d notices, %d files, %d chunks, %.1f MB fetched and discarded",
i+1, len(targets), files, chunks, float64(bytesIn)/(1<<20))
}
}
log.Printf("corpus-harvest: fetch DONE, %d notices, %d files, %d chunks, %d refused, %.1f MB fetched and discarded",
len(targets), files, chunks, skipped, float64(bytesIn)/(1<<20))
return nil
}
func listResources(ctx context.Context, c *http.Client, noticeID string) ([]resource, error) {
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf(resourceListURL, noticeID), nil)
req.Header.Set("User-Agent", "missionMATCH corpus harvest")
resp, err := c.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("resource list status %d", resp.StatusCode)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
if err != nil {
return nil, err
}
// THE LIST IS NESTED TWO DEEP. opportunityAttachmentList is a list of GROUPS, one per opportunity, and
// each group carries its own attachments array. Reading the outer list as the attachments themselves
// decodes without error and yields rows with every field empty, which is exactly what the first live
// run wrote: four documents whose download url read files//download. The flat shape is still accepted
// because sam.gov has changed this envelope before, but the nested one is the shape it serves today.
var envelope struct {
Embedded struct {
Groups []struct {
OpportunityID string `json:"opportunityId"`
Attachments []resource `json:"attachments"`
} `json:"opportunityAttachmentList"`
} `json:"_embedded"`
Resources []resource `json:"resources"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, fmt.Errorf("resource list is not the expected json: %w", err)
}
if len(envelope.Resources) > 0 {
return envelope.Resources, nil
}
var out []resource
for _, g := range envelope.Embedded.Groups {
out = append(out, g.Attachments...)
}
return out, nil
}
// fetchOne downloads, extracts, writes and DISCARDS. The bytes exist only inside this function.
func fetchOne(ctx context.Context, pool *pgxpool.Pool, c *http.Client, recordID, noticeID string, r resource) (int64, int, error) {
status, errText := "ok", ""
var data []byte
var sum string
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf(fileURL, r.ResourceID), nil)
req.Header.Set("User-Agent", "missionMATCH corpus harvest")
resp, err := c.Do(req)
if err != nil {
status, errText = "fetch_failed", err.Error()
} else {
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
status, errText = "fetch_failed", fmt.Sprintf("status %d", resp.StatusCode)
} else {
data, err = io.ReadAll(io.LimitReader(resp.Body, maxFileBytes+1))
switch {
case err != nil:
status, errText = "fetch_failed", err.Error()
case int64(len(data)) > maxFileBytes:
status, errText, data = "too_large", "over the per file cap", nil
}
}
}
var chunks []opportunities.Chunk
if status == "ok" {
h := sha256.Sum256(data)
sum = hex.EncodeToString(h[:])
text, xerr := docextract.ExtractText(ctx, r.Name, data)
switch {
case xerr != nil:
status, errText = "parse_error", xerr.Error()
case strings.TrimSpace(text) == "":
status, errText = "no_prose", "nothing readable in this file"
default:
chunks = opportunities.ExtractDeterministic(text)
if len(chunks) == 0 {
status, errText = "no_prose", "no citable span found"
}
}
}
n := int64(len(data))
written, err := writeDocument(ctx, pool, recordID, noticeID, r, sum, status, errText, n, chunks)
// data goes out of scope here and is never written to disk. That is the compression.
return n, written, err
}
func writeDocument(ctx context.Context, pool *pgxpool.Pool, recordID, noticeID string, r resource,
sum, status, errText string, size int64, chunks []opportunities.Chunk) (int, error) {
tx, err := pool.Begin(ctx)
if err != nil {
return 0, err
}
defer tx.Rollback(ctx)
// sha256 and source_url are NOT NULL because a document whose original cannot be recovered is a claim
// without evidence. A failed fetch still records the url and an empty hash it can be retried from.
var docID string
err = tx.QueryRow(ctx, `
INSERT INTO corpus_documents
(opportunity_record_id, resource_id, file_name, media_type, size_bytes,
sha256, source_url, extract_status, extract_error, fetched_bytes, fetched_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now())
ON CONFLICT (opportunity_record_id, resource_id) DO UPDATE
SET extract_status = EXCLUDED.extract_status,
extract_error = EXCLUDED.extract_error,
sha256 = EXCLUDED.sha256,
fetched_bytes = EXCLUDED.fetched_bytes,
fetched_at = now(),
updated_at = now()
RETURNING id::text`,
recordID, r.ResourceID, r.Name, r.Type, nullableInt(r.Size),
sum, fmt.Sprintf(fileURL, r.ResourceID), status, truncate(errText, 500), size).Scan(&docID)
if err != nil {
return 0, err
}
written := 0
for _, c := range chunks {
// embedding is left NULL on purpose. It is the queue the embed workers drain, and leaving it null
// is what makes the expensive half of this job liquid across the federation.
if _, err := tx.Exec(ctx, `
INSERT INTO corpus_chunks
(document_id, page_number, section_label, text_content, content_hash,
extraction_method, confidence)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
docID, nullableInt(int64(c.PageNumber)), c.SectionLabel, c.Text, c.ContentHash,
c.Method, c.Confidence); err != nil {
return written, err
}
written++
}
return written, tx.Commit(ctx)
}
func recordNoAttachments(ctx context.Context, pool *pgxpool.Pool, recordID string) error {
_, err := pool.Exec(ctx, `
INSERT INTO corpus_documents
(opportunity_record_id, resource_id, file_name, sha256, source_url,
extract_status, extract_error, fetched_at)
VALUES ($1, 'none', '', '', 'none', 'skipped', 'this notice has no attachments', now())
ON CONFLICT (opportunity_record_id, resource_id) DO NOTHING`, recordID)
return err
}
// ---------------------------------------------------------------------------------------------------
// EMBED: any number of nodes, self balancing, no orchestrator.
// ---------------------------------------------------------------------------------------------------
func runEmbed(ctx context.Context, pool *pgxpool.Pool, host string, limit int) error {
// THE OPENAI COMPATIBLE PATH, not ollama's native one. Both are served on the same port and they take
// different shapes: /api/embeddings takes a single prompt and returns one embedding, while
// /v1/embeddings takes an input array and returns a data array, which is what the platform embedder
// already speaks. Pointing at the native path decoded cleanly and returned zero vectors for a batch of
// thirty two, which is a silent shape mismatch rather than an error, and both nodes failed identically
// on it within a second of each other.
endpoint := firstNonEmpty(os.Getenv("MM_EMBED_ENDPOINT"), "http://127.0.0.1:11434/v1/embeddings")
model := firstNonEmpty(os.Getenv("MM_EMBED_MODEL"), "qllama/bge-small-en-v1.5")
dim := 384
// The HTTP client gets the SAME binding. Today each node embeds against its own ollama, but the
// endpoint is configurable and pointing one node at another is exactly the liquid arrangement this
// exists for, so the transport must be able to reach across the switch too.
httpClient := &http.Client{Timeout: 60 * time.Second}
if dialer, derr := bindToInterface(strings.TrimSpace(os.Getenv("MM_BIND_INTERFACE"))); derr == nil && dialer != nil {
httpClient.Transport = &http.Transport{DialContext: dialer}
}
emb := embeddings.NewHTTP(endpoint, model, "", dim, httpClient)
log.Printf("corpus-harvest: embed on %s against %s model=%s dim=%d", host, endpoint, model, dim)
total := 0
// A WORKER THAT DIES ON ONE TRANSIENT CONNECT IS NOT A FEDERATION MEMBER, it is a single point of
// failure wearing a distributed hat. VALKYRIE reaches the database across a switch through a USB
// ethernet adapter, and a detached run occasionally got EHOSTUNREACH on its first connect while ten
// sequential runs in a held session succeeded every time. The adapter itself is clean, 51 million
// packets with zero errors, so the loss is somewhere above it and is not worth pretending will never
// happen. The queue is already safe against a worker vanishing mid batch, so the only thing missing was
// the patience to try again.
consecutiveFailures := 0
for {
n, err := embedOnce(ctx, pool, emb)
if err != nil {
consecutiveFailures++
if consecutiveFailures > embedRetries {
return fmt.Errorf("gave up after %d consecutive failures: %w", consecutiveFailures, err)
}
// Linear backoff, deliberately short. This is a transient network fault on a local switch, not
// a rate limited API, so seconds are the right unit and there is nothing to be polite to.
wait := time.Duration(consecutiveFailures) * 2 * time.Second
log.Printf("corpus-harvest: %s attempt %d failed (%v), retrying in %s",
host, consecutiveFailures, err, wait)
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(wait):
}
continue
}
consecutiveFailures = 0
if n == 0 {
log.Printf("corpus-harvest: embed DONE on %s, queue dry, %d chunk(s) embedded here", host, total)
return nil
}
total += n
if total%(embedBatch*8) == 0 {
log.Printf("corpus-harvest: %s has embedded %d chunk(s)", host, total)
}
if limit > 0 && total >= limit {
log.Printf("corpus-harvest: embed stopping at the requested limit, %d chunk(s) on %s", total, host)
return nil
}
}
}
// embedOnce claims a batch and embeds it inside ONE transaction.
//
// FOR UPDATE SKIP LOCKED is the whole distribution mechanism. Postgres hands each worker a disjoint set
// with no coordination between them, so a second and third node need no configuration to join and no
// permission to leave. Holding the transaction open across the embedding calls is deliberate: if this node
// dies mid batch the transaction dies with it and the rows are immediately claimable by somebody else,
// which is the failure behaviour a lease with a timeout only approximates.
func embedOnce(ctx context.Context, pool *pgxpool.Pool, emb embeddings.Embedder) (int, error) {
tx, err := pool.Begin(ctx)
if err != nil {
return 0, err
}
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx, `
SELECT id, text_content FROM corpus_chunks
WHERE embedding IS NULL
ORDER BY id
FOR UPDATE SKIP LOCKED
LIMIT $1`, embedBatch)
if err != nil {
return 0, err
}
var ids []int64
var texts []string
for rows.Next() {
var id int64
var t string
if err := rows.Scan(&id, &t); err != nil {
rows.Close()
return 0, err
}
ids = append(ids, id)
texts = append(texts, t)
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, err
}
if len(ids) == 0 {
return 0, nil
}
vecs, err := emb.EmbedBatch(ctx, texts)
if err != nil {
return 0, fmt.Errorf("embed batch of %d: %w", len(texts), err)
}
if len(vecs) != len(ids) {
return 0, fmt.Errorf("embedder returned %d vectors for %d chunks", len(vecs), len(ids))
}
for i, id := range ids {
if _, err := tx.Exec(ctx,
`UPDATE corpus_chunks SET embedding = $2::halfvec WHERE id = $1`, id, vectorLiteral(vecs[i])); err != nil {
return 0, err
}
}
return len(ids), tx.Commit(ctx)
}
// vectorLiteral renders a vector in the text form pgvector parses for both vector and halfvec, so the
// worker needs no pgvector Go type registration to write a halfvec column.
func vectorLiteral(v []float32) string {
var b strings.Builder
b.WriteByte('[')
for i, f := range v {
if i > 0 {
b.WriteByte(',')
}
fmt.Fprintf(&b, "%g", f)
}
b.WriteByte(']')
return b.String()
}
func nullableInt(n int64) any {
if n <= 0 {
return nil
}
return n
}
func truncate(s string, n int) string {
if len(s) <= n {
return s
}
return s[:n]
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if s := strings.TrimSpace(v); s != "" {
return s
}
}
return ""
}
// recordRefusal writes the fact that an attachment was deliberately NOT read, and why.
//
// A corpus that silently omits what it refused cannot be audited. An export controlled document that was
// left alone is a decision the product made, and it belongs on the record next to the ones it did read.
func recordRefusal(ctx context.Context, pool *pgxpool.Pool, recordID string, r resource, why string) error {
id := r.ResourceID
if id == "" {
id = "unknown-" + r.Name
}
_, err := pool.Exec(ctx, `
INSERT INTO corpus_documents
(opportunity_record_id, resource_id, file_name, media_type, sha256, source_url,
extract_status, extract_error, fetched_at)
VALUES ($1, $2, $3, $4, '', $5, 'skipped', $6, now())
ON CONFLICT (opportunity_record_id, resource_id) DO UPDATE
SET extract_status = 'skipped', extract_error = EXCLUDED.extract_error, updated_at = now()`,
recordID, id, r.Name, r.MimeType, fmt.Sprintf(fileURL, r.ResourceID), truncate(why, 500))
return err
}