Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/sam-sync/amendments.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/sam-sync/amendments.go |
| Original SHA-256 | df41044e51612903d93b3e0a6c6a9ce07b4634de56e371c56a5f016b19a235c6 |
| Git blob SHA | 6e4b78025ac40940cd17c0aea5fa9e96899cb862 |
| Reading-copy SHA-256 | df41044e51612903d93b3e0a6c6a9ce07b4634de56e371c56a5f016b19a235c6 |
Recovered source transcript
Download complete source text ↓200 lines · 9.0 KB
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"log"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/missionworkforce/missionmatch/internal/modules/opportunities"
"github.com/missionworkforce/missionmatch/internal/platform/samgov"
)
// The amendment path, directive Section 4 steps 6, 7, and 11.
//
// A refresh compares the source payload hash of each notice against the hash stored on the CURRENT version
// of every tenant opportunity that tracks it. When they differ, the source published something new, so a new
// immutable opportunity_versions row is written and the tenant is alerted. When they match, nothing is
// written at all, which is what makes a rerun of the sync a no op rather than a version storm.
//
// Two invariants this code exists to hold:
//
// 1. It NEVER touches tenant authored data. It inserts versions and alerts and updates only the is_current
// flag of a prior version. No column a capture owner typed on tracked_opportunities is read for writing
// or written at all, so a refresh cannot overwrite the tenant's own capture intelligence. Directive
// Section 4 item 12.
// 2. An approved requirement is never silently rewritten. The amendment produces a VERSION and an ALERT, a
// named human then accepts the operational change. Nothing downstream moves on its own.
//
// It runs as the migrator, which is BYPASSRLS, because one sync run serves every tenant tracking a notice.
// organization_id is therefore written explicitly from the row that was read, never from a session GUC.
// trackedRef is one tenant opportunity that tracks a SAM notice, with the state of its current version.
type trackedRef struct {
orgID string
oppID string
noticeID string
curHash string
curVersion int
hasVersion bool
curDeadline *time.Time
}
// applyAmendments writes a new immutable version, and an alert, for every tracked opportunity whose source
// payload changed in this run. It returns how many versions and alerts it wrote. A failure on one tenant is
// logged and skipped rather than failing the run: the catalog upserts already committed, and one tenant's
// version write must not cost every other tenant their amendment notice.
func applyAmendments(ctx context.Context, pool *pgxpool.Pool, seen map[string]samgov.Normalized, manifestID string) (versions int, alerts int, failures int) {
if len(seen) == 0 {
return 0, 0, 0
}
ids := make([]string, 0, len(seen))
for id := range seen {
ids = append(ids, id)
}
refs, err := loadTrackedRefs(ctx, pool, ids)
if err != nil {
log.Printf("sam-sync: amendment scan: %v", err)
return 0, 0, 1
}
if len(refs) == 0 {
return 0, 0, 0 // nobody tracks any notice in this run, the common case early on
}
for _, ref := range refs {
n, ok := seen[ref.noticeID]
if !ok {
continue
}
hash := payloadHash(n)
if ref.hasVersion && ref.curHash == hash {
continue // unchanged upstream, write nothing
}
wroteAlert, err := writeVersionAndAlert(ctx, pool, ref, n, hash, manifestID)
if err != nil {
// FINDING D9, THE SECOND HALF. This used to log and skip and tell the caller nothing, so the run
// reported success while an amendment went unrecorded. That is how a permanently stuck version
// counter survived from 27 July unnoticed: the only trace was one line in a journal nobody reads,
// and the exit code said everything was fine.
//
// A FAILED AMENDMENT IS A COMPLIANCE EVENT, not a transient hiccup. The product's promise is that
// an amendment FLAGS an affected approved requirement. If the version never lands, that promise is
// broken silently for that opportunity. So failures are COUNTED and returned, and the caller exits
// nonzero, which puts the run on the OnFailure alert path that already exists.
log.Printf("sam-sync: AMENDMENT FAILED for opportunity %s: %v", ref.oppID, err)
failures++
continue
}
versions++
if wroteAlert {
alerts++
}
}
return versions, alerts, failures
}
// loadTrackedRefs finds every tenant opportunity tracking one of the notices in this run, joined to its
// current version so the hash comparison is one round trip rather than one per opportunity.
func loadTrackedRefs(ctx context.Context, pool *pgxpool.Pool, noticeIDs []string) ([]trackedRef, error) {
rows, err := pool.Query(ctx, `
SELECT t.organization_id::text, t.id::text, t.sam_notice_id,
COALESCE(v.source_payload_hash, ''), COALESCE(v.version_number, 0), (v.id IS NOT NULL),
t.response_deadline
FROM tracked_opportunities t
LEFT JOIN opportunity_versions v
ON v.organization_id = t.organization_id AND v.opportunity_id = t.id AND v.is_current
WHERE t.sam_notice_id = ANY($1)`, noticeIDs)
if err != nil {
return nil, err
}
defer rows.Close()
var out []trackedRef
for rows.Next() {
var r trackedRef
if err := rows.Scan(&r.orgID, &r.oppID, &r.noticeID, &r.curHash, &r.curVersion, &r.hasVersion, &r.curDeadline); err != nil {
return nil, err
}
out = append(out, r)
}
return out, rows.Err()
}
// writeVersionAndAlert inserts the successor version and flips the predecessor in ONE transaction, which is
// what the is_current contract requires: the trigger on opportunity_versions permits exactly that flag to
// change and refuses every other update and any delete, so the lineage cannot be rewritten afterwards.
func writeVersionAndAlert(ctx context.Context, pool *pgxpool.Pool, ref trackedRef, n samgov.Normalized, hash, manifestID string) (bool, error) {
var alerted bool
var nextVersion int
err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
// FINDING D9. The version number is computed HERE, inside the transaction, from the highest number ever
// written, rather than from ref.curVersion which came from a join on is_current. When no row was
// flagged current that join matched nothing, the number read 0, and the insert asked for version 1
// again on an opportunity that already had one. The unique constraint refused it, the error was logged
// and skipped, and the next run computed 0 again. Permanently stuck, silently.
var vErr error
if nextVersion, vErr = opportunities.NextVersionNumber(ctx, tx, ref.orgID, ref.oppID); vErr != nil {
return vErr
}
// Retire unconditionally. The old code gated this on the same stale belief about what was current.
if rErr := opportunities.RetireCurrentVersions(ctx, tx, ref.orgID, ref.oppID); rErr != nil {
return rErr
}
var manifest any
if manifestID != "" {
manifest = manifestID
}
if _, err := tx.Exec(ctx, `
INSERT INTO opportunity_versions (
organization_id, opportunity_id, version_number, amendment_number, source_version_label,
source_payload_hash, normalized_payload, fetched_at, ingest_manifest_id, is_current)
VALUES ($1, $2, $3, $4, $5, $6, $7, now(), $8, true)`,
ref.orgID, ref.oppID, nextVersion, "", n.NoticeType,
hash, []byte(n.Raw), manifest); err != nil {
return fmt.Errorf("insert version: %w", err)
}
// The alert names only this tenant's own tracked opportunity. The dedupe key carries the version, so a
// rerun of the same amendment refreshes one row and a genuinely new amendment raises a new one.
title := "Amendment on " + firstNonEmpty(n.Title, n.SolicitationNumber, ref.noticeID)
detail := fmt.Sprintf("The source published a change. Version %d recorded, review the scope before it affects approved requirements.", nextVersion)
severity := "watch"
alertType := "opportunity_amendment"
// A moved response deadline is the change a capture team must see first, so it is called out as its
// own kind at urgent severity rather than folded into the generic amendment line.
if n.ResponseDeadline != nil && ref.curDeadline != nil && !n.ResponseDeadline.Equal(*ref.curDeadline) {
alertType = "opportunity_deadline"
severity = "urgent"
title = "Response deadline moved on " + firstNonEmpty(n.Title, n.SolicitationNumber, ref.noticeID)
detail = "The source deadline is now " + n.ResponseDeadline.Format(time.RFC3339) + "."
}
dedupe := fmt.Sprintf("%s:%s:v%d", alertType, ref.oppID, nextVersion)
if _, err := tx.Exec(ctx, `
INSERT INTO alerts (organization_id, alert_type, severity, subject_type, subject_id, title, detail, dedupe_key)
VALUES ($1, $2, $3, 'opportunity', $4, $5, $6, $7)
ON CONFLICT (organization_id, dedupe_key)
DO UPDATE SET status = 'unread', updated_at = now()`,
ref.orgID, alertType, severity, ref.oppID, title, detail, dedupe); err != nil {
return fmt.Errorf("raise alert: %w", err)
}
alerted = true
return nil
})
return alerted, err
}
// payloadHash fingerprints the SOURCE payload of a notice. It hashes the raw upstream JSON, not the
// normalized struct, so a change SAM published is detected even in a field this platform does not yet map,
// and a rerun over unchanged upstream data produces an identical hash and therefore writes nothing.
func payloadHash(n samgov.Normalized) string {
h := sha256.New()
_, _ = h.Write([]byte(n.NoticeID))
_, _ = h.Write([]byte(n.Raw))
return hex.EncodeToString(h.Sum(nil))
}