Documentation
End-to-end recipe for adding a new `{source}-{type}-processor` to the Vulnetix VDB pipeline. Every section maps to concrete files in this repo and to schemas/migrations in `../saas/`. **The process...
End-to-end recipe for adding a new {source}-{type}-processor to the Vulnetix VDB
pipeline. Every section maps to concrete files in this repo and to schemas/migrations
in ../saas/. The processor is not “done” until the run/verify section at the end
has been executed against .env and the resulting rows inspected.
Where processors run: All scheduled processors execute as ECS Fargate tasks in the
vdb-schedulercluster (terraform/ecs.tf). EventBridge Scheduler triggers each task on its cron; failed invocations are routed to thevdb-scheduler-dlqSQS dead-letter queue (terraform/logs.tf) with 14-day retention. The scanner processor is also in this cluster but triggered on-demand viaecs:RunTask.
Refer to AGENTS.md § “Processor Naming Convention” for the authoritative rules.
Pick one combination from each axis and lock it in:
| Axis | Values |
|---|---|
source | short slug, lowercase, hyphens (rhsa, cisa-kev, nist-nvd, enrich-nuclei, …) |
type | git json rss fetch cvrf csaf csv gsheet snort yara |
role | processor (cloud) and/or backfill (local-only) |
Locked names show up in 8 places — keep them character-identical:
cmd/{source}-{type}-processor/ scripts/go-processors/
Containerfile.go-processors → AS {source}-{type}-processor
ECR tag → go-{source}-{type}-processor-latest
ECS family / EventBridge → go-{source}-{type}-processor
task-manager.toml key → [tasks.{source}-{type}-processor]
terraform module → module "{source}_{type}_processor"
post-push-ecr.sh TARGETS → {source}-{type}-processor
justfile recipe → go-{source}-{type}-backfill (or go-enrich-{source}-{type})
Before touching any file in the repo, verify the source is reachable, parseable, and has the shape you assume. Most processor rewrites trace back to skipping this.
Run from a scratch directory:
# 1. Reachability + headers (anti-bot? Cloudflare? auth?)
curl -sSI -A 'Mozilla/5.0 …' "$URL" | head -40
# 2. Sample payload (≥1 advisory, ≥1 page-2 if paginated)
curl -sS -A 'Mozilla/5.0 …' "$URL" -o /tmp/sample.{json,xml,html}
# 3. Identifier inspection — what IS the natural key?
jq '.[0] | keys' /tmp/sample.json
# or
xmllint --xpath '//*[local-name()="vulnerability"][1]' /tmp/sample.xml | head
# 4. CVE alias presence — does this source emit CVE-YYYY-NNNN?
grep -oE 'CVE-[0-9]{4}-[0-9]{3,7}' /tmp/sample.* | sort -u | wc -l
# 5. Pagination / freshness signal
# Is there a Last-Modified header? An incremental "since" parameter? A "page" cursor?
# A SHA-able listing endpoint? An RSS pubDate? A baked git HEAD?
Document the answers in the design doc (§3) under Source contract. The four questions that determine the rest of the architecture:
cron_match and frequency_secs)cpu_units/memory_mb/expected_duration_minutes)cmd/{source}-{type}-backfill/ companion)Each of the existing types has a representative reference processor. Read its
main.go end-to-end before writing yours.
| Type | Representative | Use when |
|---|---|---|
git | cmd/pypi-git-processor/main.go | Source is a git repo of YAML/JSON advisories. Bake small repos (≤12 MB) in the Containerfile data stage; clone large repos at runtime via processor.PullOrClone. |
json | cmd/aws-security-bulletins-json-processor/main.go | REST/JSON API or downloadable JSON dump. Add CRIT staging here when records map to cloud resources. |
rss | cmd/acsc-rss-processor/main.go | RSS/Atom feed. Use when no JSON/git equivalent exists. |
fetch | cmd/pwno-fetch-processor/main.go | HTML scraping or non-API HTTP fetch. Often needs httpclient.NewHTTP1 + browser headers. Use cert-il-fetch (Alpine + Chromium / go-rod) only when Cloudflare challenge bypass is required. |
csaf / cvrf | cmd/redhat-csaf-processor/, cmd/cisco-cvrf-processor/ | OASIS CSAF 2.0 JSON or CVRF XML provider trees. |
csv / gsheet | cmd/epss-csv-backfill/, cmd/gpz-0day-itw-gsheet-processor/ | CSV-typed feed, including public Google Sheets exports. |
snort / yara | cmd/community-snort-processor/, cmd/github-yara-fetch-processor/ | Detection-rule corpora. Populates SnortRule/YaraRule and *RuleCVE junctions, not CVEMetadata directly. |
| KEV-only | cmd/cisa-kev-json-processor/ | Source emits only (cveId, exploit metadata) rows. Writes the Kev table; does not create CVEMetadata. |
| AI-discovery | cmd/pwno-fetch-processor/, cmd/gpz-0day-itw-gsheet-processor/ | Source attributes 0-days to AI-driven discovery. Writes CVEAiDiscovery + AIDiscoveryTag. |
If the source feeds cloud-resource advisories (vendor:product:resource_type triples)
include CRIT staging following the aws-security-bulletins-json-processor pattern.
scripts/go-processors/{source}-{type}-processor.design.md)Required before writing main.go. Use this template (mirrors
pypi-processor.design.md, sap-patch-day-fetch-processor.design.md):
# {source}-{type}-processor — Design
## 1. Overview
- Purpose, source URL, owner, licence
- Schedule cadence (`cron(...)`) and rationale
- ECS resources: cpu_units, memory_mb, expected_duration_minutes
- Reads: external API/feed, BulkDataDumpTracker, LoadProcessedHashes
- Writes: CVEMetadata / CVEAlias / Exploit / Kev / SnortRule / YaraRule / CrowdSecSighting / CritRecord / VEX / AIDiscovery / S3 archive
## 2. Source contract (from Phase 0 recon)
- Cadence, volume, identifier shape, backfill feasibility
- Anti-bot / auth / rate-limit profile
- Pagination/freshness mechanism (Last-Modified, ETag, listing SHA, cursor, etc.)
## 3. Architecture diagram
```mermaid
graph LR
Source --> HTTP[httpclient/git PullOrClone] --> Parse --> Map[osv.MapAdvisory or source-specific]
Map --> Tx[(pgx Tx)] --> Pipeline[processor.StoreCVESourceData]
Pipeline --> CVEMetadata
Pipeline --> CVEAlias[(db.InsertAliases)]
Pipeline --> CVEMetric & CVEDescription & CVEReference & CVEAffected
Tx -.commit.-> S3[s3client.ArchiveRecord]
Pipeline --> CRIT[critutil.NewEnvelope -> StageCandidate] --> Drain[critpublisher.DrainKeys] --> CritRecord & VEX
Exhaustive table: every field on the source record → target column on
CVEMetadata / Description / Reference / Metric / ProblemType / Affected / AffectedVersion / Alias.
Note any synthesised fields (e.g. derived CVSS via cvss.DeriveV4FromDescription).
CVEMetadata.cveId? (CVE-prefix vs source-prefix — see §6)CVEAlias? Direction. Bundle-suppression risk.{SOURCE}-YYYY-NNNN, GCVE form GCVE-110-{SOURCE}-YYYY-NNNN):
how is the sequence loaded (db.LoadMaxGcveSequence pattern) and persisted?internal/critutil/dictionaries/extended/{vendor}.json)?fan-out width), per-request timeouts, rate-limit pacingcmd/{source}-{type}-backfill/ warranted? If yes, what differs from the scheduled binary?
---
## 4. Code structure
scripts/go-processors/ ├── cmd/ │ └── {source}-{type}-processor/ │ └── main.go # CLI flags, env, pool init, top-level orchestration ├── internal/ │ └── {source}/ # Source-specific parser/mapper (only if no existing pkg fits) │ ├── client.go # HTTP / git / file fetch │ ├── parse.go # XML/JSON/RSS → typed structs │ ├── map.go # typed structs → osv.CVESourceData │ └── parse_test.go # golden-file tests against schemas/{source}_*.schema.json └── {source}-{type}-processor.design.md
**Reuse first** — before adding a new `internal/{source}/` package, check whether
an existing one fits (`internal/osv` for OSV YAML/JSON, `internal/cisco`, `internal/redhat`,
`internal/critutil`, `internal/sourceident` for ID extraction).
### `main.go` skeleton (json/fetch archetype)
```go
package main
import (
"context"
"flag"
"log/slog"
"os"
"time"
"github.com/vulnetix/vdb-manager/go-processors/internal/db"
"github.com/vulnetix/vdb-manager/go-processors/internal/httpclient"
"github.com/vulnetix/vdb-manager/go-processors/internal/notify"
"github.com/vulnetix/vdb-manager/go-processors/internal/osv"
"github.com/vulnetix/vdb-manager/go-processors/internal/processor"
"github.com/vulnetix/vdb-manager/go-processors/internal/s3client"
)
const (
processorName = "{source}-{type}-processor"
sourceSlug = "{source}"
frequencySecs = 86_400 // tracker cadence
)
func main() {
force := flag.Bool("force", false, "Bypass tracker freshness check")
limit := flag.Int("limit", 0, "Limit records (0 = all)")
flag.Parse()
logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
notif := notify.New(logger)
notif.Started(processorName)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
notif.SetOvertimeCancel(ctx, cancel) // honours EXPECTED_DURATION_MINUTES
pool, err := db.NewPool(ctx, db.EnvDatabaseURL(), db.EnvDatabaseURLRead())
if err != nil { notif.Errored(processorName, nil, err); os.Exit(1) }
defer pool.Close()
// 0. Zombie / lock check (only if multiple writers possible)
// 1. Tracker freshness
if !*force {
fresh, err := db.TrackerFresh(ctx, pool.Read, sourceSlug, frequencySecs)
if err != nil { notif.Errored(processorName, nil, err); os.Exit(1) }
if fresh { notif.NoWork(processorName); return }
}
httpC := httpclient.New() // or NewHTTP1 for anti-bot sources
uploader := s3client.NewUploaderFromEnv(logger)
// 2. Fetch + parse
items, listingSHA, err := fetchAndParse(ctx, httpC)
if err != nil { notif.Errored(processorName, nil, err); os.Exit(1) }
// 3. Resume set
seen, err := db.LoadProcessedHashes(ctx, pool.Read, sourceSlug)
if err != nil { notif.Errored(processorName, nil, err); os.Exit(1) }
stats := map[string]any{"fetched": len(items)}
softDeadline := notif.SoftDeadline()
// 4. Per-item processing
for i, it := range items {
if *limit > 0 && i >= *limit { break }
if !softDeadline.IsZero() && time.Now().After(softDeadline) {
logger.Warn("soft deadline reached, stopping cleanly")
break
}
if _, dup := seen[it.SourceFileHash]; dup && !*force { continue }
data := mapToSourceData(it) // *osv.CVESourceData with Aliases populated
if err := processor.RunOne(ctx, pool, data, processor.Hooks{
OnArchive: uploader.PipelineArchive(sourceSlug),
OnQuarantine: uploader.PipelineQuarantine(processorName),
}); err != nil {
notif.RecordError(it.ID + ": " + err.Error())
continue
}
stats["stored"] = stats["stored"].(int) + 1
}
// 5. Tracker bump (only if no fatal errors)
if !notif.HasErrors() {
_ = db.UpsertTracker(ctx, pool.Write, sourceSlug, listingSHA, frequencySecs)
}
if notif.HasErrors() {
notif.Errored(processorName, stats, nil)
os.Exit(1)
}
notif.Completed(processorName, stats)
}
Every advisory MUST flow through processor.StoreCVESourceData (file:
internal/processor/pipeline.go:595). It composes:
db.UpsertCVEMetadata (composite key (cveId, source))db.InsertDescriptions, InsertReferences, InsertMetrics,
InsertProblemTypes, UpsertAffected, InsertVersionsdb.EnrichAffectedWithDependency (opt-in)db.InsertAliases(ctx, tx, cveID, source, aliases, logger) — see § “Alias Writes” in AGENTS.mdDirect UpsertCVEMetadata callers MUST also call db.InsertAliases (even with
aliases == nil) so the same-cveId cross-source backfill runs. Never issue
raw INSERT INTO "CVEAlias" SQL — bundle-suppression and canonicalisation are
implemented inside InsertAliases only.
CVEMetadata.cveId vs CVEAliasDecision is per-source and must be documented in §5 of the design doc.
Rule of thumb (matches existing processors):
| Source emits… | CVEMetadata.cveId | CVEAlias rows |
|---|---|---|
| Native CVE-prefixed IDs only (cve.org, NVD) | CVE-YYYY-NNNN | Cross-source peers (nist-nvd ⇄ cve.org) auto-linked |
| Source-prefixed + CVE alias(es) (RHSA, GHSA, MSRC, EUVD) | CVE-YYYY-NNNN (preferred — osv.MapAdvisory extracts it from aliases[]) | The source-prefixed ID (RHSA-…, GHSA-…) |
Source-prefixed only, no CVE alias (PWNO, CIRCL, BDU, CNVD, ACSC-minted ACSC-…) | The source/minted ID | Any future CVE references discovered later |
| Source-prefixed and want to KEEP the source ID as primary | Use osv.MapAdvisoryKeepID(...) | All CVE-prefixed aliases |
Why prefer the CVE-prefix as primary when it exists: it maximises hit-rate for
downstream lookups (Kev, EPSS, Exploit, customer SBOM scans) which all key on
CVE-YYYY-NNNN. The source-prefixed ID survives as a CVEAlias row, so reverse
lookups from RHSA-/GHSA-/MSRC- still resolve via the alias graph.
Bundle suppression: when an RHSA/MSRC/etc. bundles N CVE aliases, only the first
becomes cveId; the rest are dropped from the alias edges to prevent fan-out
(implemented in internal/db/cvealias.go:50). Document the chosen ordering.
Minted IDs: if the source needs a synthetic prefix (e.g. ACSC’s
ACSC-YYYY-NNNN, no VVD- prefix), follow the GCVE pattern: db.LoadMaxGcveSequence at
startup, increment in-memory, persist via the GCVE issuance row + GcveAlias
junctions. The GCVE identifier uses Vulnetix’s GCVE Numbering Authority (GNA) id
110 — i.e. GCVE-110-ACSC-YYYY-NNNN. Mirror the CVE alias edges via
db.InsertAliases so reverse lookups work in both directions.
CRIT (cloud advisory) records belong to processors whose source attributes vulnerabilities to (Provider, Service, ResourceType) triples — AWS bulletins, GCP bulletins, Cloudflare advisories, Oracle CPU, ServiceNow KB, SAP Patch Day, IBM Security Bulletins, Broadcom/VMware, Salesforce.
Pattern (from aws-security-bulletins-json-processor):
import (
"github.com/vulnetix/vdb-manager/go-processors/internal/critutil"
"github.com/vulnetix/vdb-manager/go-processors/internal/critpublisher"
)
// Once per process:
dictRegistry, _ := critutil.LoadAllExtendedDictionaries()
// or critutil.LoadDictionary("internal/critutil/dictionaries/extended/{vendor}.json")
// Per (vulnId × ServiceMatch):
candidate := mapBulletinToCRIT(item, match) // build critutil.Candidate
env, err := critutil.NewEnvelope(processorName, candidate, provenance, validation)
key, err := critutil.StageCandidate(ctx, uploader, env) // S3 → crit-candidates/pending/...
_ = db.RegisterS3QueueObject(ctx, pool.Write, "pending", key)
// End of run:
if !envBool("CRIT_DISABLE_INPROCESS_DRAIN") {
_ = critpublisher.DrainKeys(ctx, pool, uploader, stagedKeys, logger)
}
DrainKeys validates each envelope against crit-record-v0.2.0.schema.json,
inserts CritRecord + sibling VEX statement (openvex table), or moves the
envelope to the rejected/spec-violation prefix. Processors do not emit VEX
directly — that responsibility belongs to the publisher.
If the source needs new (provider, service, resource_type) tokens, extend
internal/critutil/dictionaries/extended/{vendor}.json and reference it from
the design doc.
Pick one (or more) based on source character. Helpers all live in internal/db/:
| Table family | Helper | Used by |
|---|---|---|
Exploit + ExploitCVE | db.UpsertExploit, db.InsertExploitCVE | 0day-today, ExploitDB, Metasploit, snort/yara processors (legacy compat) |
Kev | db.UpsertKev | cisa-kev, eukev, vulncheck-kev (writes Kev only — does NOT create CVEMetadata) |
VulnetixKev | db.UpsertVulnetixKev(reason) | vulnetix-kev (synthesised, reason ∈ {crowdsec_sighting, snort_rule, nuclei_template, nse_script, big_sleep, …}) |
SnortRule + SnortRuleCVE | db.UpsertSnortRule, db.InsertSnortRuleCVE | community-snort, emergingthreats-snort, community-suricata-snort |
YaraRule + YaraRuleCVE | db.UpsertYaraRule, db.InsertYaraRuleCVE | github-yara-fetch |
CrowdSecSighting + CrowdSecLog | db.UpsertCrowdSecSighting | crowdsec-json-processor |
VulnCheckKEV + VulnCheckKEVCVE | db.UpsertVulnCheckKEV | vulncheck-kev-json |
CVEAiDiscovery + AIDiscoveryTag | db.UpsertAIDiscovery | pwno-fetch, gpz-0day-itw-gsheet, big-sleep |
For rule-corpus processors (snort/yara) the workflow is:
parse rule → extract CVE refs → upsert *Rule row → write *RuleCVE junctions →
optionally write a legacy Exploit row for backward compat (sourceSlug = source name).
No CVEMetadata write; rules attach to whatever CVEMetadata the central pipeline
already created.
schemas/{source}_*.schema.json)Add a JSON Schema only when:
//go:embed + jsonschema), orparse_test.go golden tests.Place at schemas/{source}_advisory.schema.json following neighbouring files
(e.g. apk_secdb_advisory.schema.json, csaf_2.0.schema.json). For CRIT
producers, the canonical schema lives in
scripts/go-processors/internal/critutil/crit-record-v0.2.0.schema.json — do
not duplicate it under schemas/.
../saas/prisma/migrations/)The vdb-manager repo defines schema in ../saas/prisma/models/*.prisma but
all migrations are owned by the saas project. If your processor needs new
columns/tables/indexes:
../saas/prisma/models/
(e.g. niche tables → exploit.prisma, threat-intel.prisma,
ai-discovery.prisma; CRIT → crit.prisma).../saas/prisma/migrations/YYYYMMDDNNNNNN_add_{thing}/migration.sql with
idempotent SQL:-- Migration: Add {thing}
-- Idempotent: safe to run multiple times
CREATE TABLE IF NOT EXISTS "{Table}" ( ... );
DO $$ BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = '{Table}_unique_key') THEN
ALTER TABLE "{Table}" ADD CONSTRAINT "{Table}_unique_key" UNIQUE ("col1","col2");
END IF;
END $$;
CREATE INDEX CONCURRENTLY IF NOT EXISTS "{Table}_idx" ON "{Table}" ("col");
npx prisma migrate deploy from ../saas/; apply to
production via set -a; source .env.production; set +a; psql "$DATABASE_URL" -f migration.sql
from the ../saas/ directory. Use CREATE INDEX CONCURRENTLY on large tables
to avoid locking out ECS connections.Do NOT commit a Prisma migration without first running the SQL against a local
copy of production schema (pg_dump --schema-only is sufficient).
Every file edit listed here is required. Skipping any one of them breaks deploy or scheduling.
Containerfile.go-processorsAppend a final-target stage. Pattern:
# ---- {Source} {Type} Processor (final target) ----
FROM scratch AS {source}-{type}-processor
COPY --from=cert-builder /etc/ssl/ /etc/ssl/
COPY --from=builder --chmod=0755 /out/{source}-{type}-processor /app/{source}-{type}-processor
ENTRYPOINT ["/app/{source}-{type}-processor"]
Variants:
FROM alpine:3.21 AS {source}-git-data
stage above and COPY --from={source}-git-data /data/{repo} /data/{repo} into
the final stage. Always write .baked-sha so the scratch container can
resolve HEAD without a git binary.processor.PullOrClone at runtime.FROM alpine:3.21 final stage and
RUN apk add --no-cache chromium. Reserved for cert-il-fetch-processor-class
sources.FROM alpine:3.21 + RUN apk add --no-cache git ca-certificates (see
ocaml-git-processor, hsec-git-processor).scripts/task-manager.toml[tasks.{source}-{type}-processor]
name = "{Source} {Type} Processor"
description = "One-line purpose"
category = "go-standalone" # or "go-git"
container = true
runner_cmd = "go {source}-{type}-processor"
aws_schedule = "go-{source}-{type}-processor"
cron_match = "0 6 * * *" # daily 06:00 UTC unless source dictates otherwise
cpu_units = 256
memory_mb = 512
expected_duration_minutes = 30
design_doc = "scripts/go-processors/{source}-{type}-processor.design.md"
[[tasks.{source}-{type}-processor.args]]
name = "force"
type = "flag"
required = false
default = "false"
terraform/go-schedules.tfmodule "{source}_{type}_processor" {
source = "./modules/ecs-go-task"
task_name = "{source}-{type}-processor"
schedule_expression = "cron(0 6 * * ? *)"
command = ["/app/{source}-{type}-processor"]
cpu = 256
memory = 512
schedule_enabled = true
expected_duration_minutes = 30
cluster_arn = aws_ecs_cluster.vdb.arn
container_image = "${aws_ecr_repository.go_processors.repository_url}:go-{source}-{type}-processor-${var.go_container_image_tag}"
execution_role_arn = aws_iam_role.ecs_execution.arn
task_role_arn = aws_iam_role.ecs_task.arn
scheduler_role_arn = aws_iam_role.scheduler.arn
subnet_ids = var.subnet_ids
security_group_ids = [aws_security_group.ecs_tasks.id]
secrets = local.go_task_secrets # use _github variant if GH_TOKEN needed
environment = local.go_task_environment
dlq_arn = aws_sqs_queue.scheduler_dlq.arn
aws_region = var.aws_region
log_retention_days = var.log_retention_days
}
task_name MUST match the value scheduleToTaskName(aws_schedule) derives —
i.e. the toml aws_schedule with go- stripped. Stale task_name values land
the row in the dashboard’s “Unknown” frequency bucket.
Every processor log group (/ecs/vdb-scheduler/*) keeps only 3 days of
CloudWatch retention (var.log_retention_days). A CloudWatch subscription
filter streams all events through a single Kinesis Firehose into the private
vdb-manager-logs S3 bucket, where a lifecycle rule transitions objects to
Glacier Deep Archive at day 30 and expires them at day 395 — durable ~1-year
history at Glacier prices, without paying for long CloudWatch retention. The
Firehose wiring lives in terraform/logs-archive.tf; the per-group filter is
part of the ecs-go-task module.
.claude/hooks/post-push-ecr.shAdd {source}-{type}-processor to TARGETS (alphabetical order).
scripts/task-dashboard/cmd/ecr-build/targets.goAdd to simpleTargets (no data stage) or dataTargets (git-baked).
justfileBackfill recipe:
go-{source}-{type}-backfill TARGET="local" BATCH_SIZE="200" FORCE="false" LIMIT="0":
#!/usr/bin/env bash
set -euo pipefail
if [ "{{TARGET}}" = "prod" ]; then ENV_FILE=".env.production"; else ENV_FILE=".env"; fi
[ -f "$ENV_FILE" ] || { echo "ERROR: $ENV_FILE not found"; exit 1; }
set -a; source "$ENV_FILE"; set +a
unset EXPECTED_DURATION_MINUTES # backfill MUST run to completion — no soft deadline
cd scripts/go-processors && go run ./cmd/{source}-{type}-processor \
--batch-size={{BATCH_SIZE}} --force={{FORCE}} --limit={{LIMIT}}
For git-baked sources also bind --repo from REPO/REPO_ABS=$(realpath …)
before cd scripts/go-processors (the binary’s cwd changes).
For enrichment processors use go-enrich-{source}-{type} (no -backfill suffix).
.github/workflows/go-ecr-deploy.ymlThe matrix is sourced from task-manager.toml; once steps 11.2 and 11.4 are
complete the manual workflow_dispatch trigger picks up the new target without
file edits to the workflow itself. Confirm by listing inputs:
gh workflow view go-ecr-deploy.yml.
scripts/go-processors/entrypoint.shNo edit required — already constructs DATABASE_URL/DATABASE_URL_READ from
ECS secrets for every processor.
scripts/sql/Only add a SQL helper here for one-off operational fixes (e.g. backfilling a
source attribution column for legacy rows). Pattern:
normalize-{thing}.sql. Not used at processor runtime.
Every processor wraps its top-level orchestration with the notify package
(internal/notify/notify.go):
notif := notify.New(logger)
notif.Started(processorName)
defer func() {
if r := recover(); r != nil {
notif.Errored(processorName, stats, fmt.Errorf("panic: %v", r))
os.Exit(2)
}
}()
// per-record:
notif.RecordError(itemID + ": " + err.Error())
// final:
if notif.HasErrors() { notif.Errored(processorName, stats, nil); os.Exit(1) }
notif.Completed(processorName, stats)
notif.SetOvertimeCancel(ctx, cancel) cancels the root context 10 min before
EXPECTED_DURATION_MINUTES elapses, letting in-flight transactions commit. SNS
topic + bot are wired through ECS env vars by Terraform — no per-processor
config.
httpclient.New (HTTP/2 default) or httpclient.NewHTTP1
(only when source rejects HTTP/2 — common for older WAFs).context.WithTimeout (90 s default; tune for the source).chan struct{} semaphore (4 workers is the
default in existing processors).unexpected EOF / connection reset as transient.db.LoadProcessedHashes.--force.db.NewPool defaults (read=5, write=5). Do not create
ad-hoc pgx.Pool instances.internal/{source}/testdata/.parse_test.go: parse a recorded payload, assert mapped osv.CVESourceData
matches an expected JSON snapshot (use cmp.Diff).map_test.go: cveId selection across the cases enumerated in design § 5.alias_test.go (when bundle suppression is in play): assert
db.InsertAliases keeps the right primary CVE under multi-CVE bundles.cd scripts/go-processors && go test ./internal/{source}/... ./cmd/{source}-{type}-processor/...
Before opening a PR:
# 1. Compile every binary (catches type errors that go test misses)
cd scripts/go-processors
go build ./...
go vet ./...
# 2. Local container build (smoke — uses podman, ARM64 cross-compile)
cd ../..
podman build -f Containerfile.go-processors --target {source}-{type}-processor \
-t local/{source}-{type}-processor:dev .
# 3. Run via the dashboard TUI — exercises the targets.go entry
just dashboard # navigate to the new processor, "Build" then "Run"
Manual ECR push for the first production build:
# Triggered three ways — pick whichever the situation calls for
gh workflow run go-ecr-deploy.yml -f target={source}-{type}-processor # GHA
just ecr-push TARGET={source}-{type}-processor # justfile
# Or simply: git push origin main → .claude/hooks/post-push-ecr.sh fires
This is the gate. The processor is not finished until these steps pass against real data and the resulting rows are inspected.
# From vdb-manager root
just go-{source}-{type}-backfill TARGET=prod LIMIT=10 FORCE=true
Expected log shape:
{"time":"…","level":"INFO","msg":"started","processor":"{source}-{type}-processor"}
{"time":"…","level":"INFO","msg":"fetched","count":<n>}
{"time":"…","level":"INFO","msg":"stored","cveId":"<id>","source":"{source}"}
{"time":"…","level":"INFO","msg":"completed","stats":{"fetched":10,"stored":N,"skipped":M}}
Confirm Slack channel received Started and Completed events.
set -a; source .env.production; set +a
psql "$DATABASE_URL" <<'SQL'
-- 1. CVEMetadata rows landed and have the expected source slug
SELECT "cveId","source","datePublished","title"
FROM "CVEMetadata" WHERE "source" = '{source}'
ORDER BY "datePublished" DESC NULLS LAST LIMIT 10;
-- 2. Aliases were written and direction is canonical
SELECT "primaryCveId","primarySource","aliasCveId","aliasSource","discoveredFrom"
FROM "CVEAlias"
WHERE "primarySource" = '{source}' OR "aliasSource" = '{source}'
ORDER BY "discoveredAt" DESC LIMIT 20;
-- 3. References + Metrics + Affected populated
SELECT count(*) FROM "CVEReference" r
JOIN "CVEMetadata" m ON r."cveId"=m."cveId" AND r."source"=m."source"
WHERE m."source" = '{source}';
-- 4. BulkDataDumpTracker bumped
SELECT * FROM "BulkDataDumpTracker" WHERE "source" = '{source}';
-- 5. Niche tables (only the ones your processor writes)
-- SELECT count(*) FROM "Kev" WHERE "source" = '{source}';
-- SELECT count(*) FROM "SnortRule" WHERE "source" = '{source}';
-- SELECT count(*) FROM "CritRecord" WHERE "producer" LIKE '{source}%';
SQL
Failure modes to watch for and what they mean:
| Symptom | Likely cause |
|---|---|
discoveredAt near 1700000000 (Unix seconds, not millis) | Used time.Now().Unix() — must be UnixMilli() |
| Alias rows missing the same-cveId cross-source backfill | Bypassed db.InsertAliases (raw SQL or only called on non-empty lists) |
cveId is the source-prefixed ID despite a CVE alias being present | osv.MapAdvisory not used or KeepID accidentally selected |
| Tracker not bumped | Final UpsertTracker conditioned incorrectly, or processor exited via os.Exit(1) after errors (correct — only bump on clean exit) |
Connection error to 127.0.0.1:5432 | Pool init bug (see CLAUDE.md “Prisma Connection Issues” — applies to pgx too: pass DSN string, not a pre-built Pool) |
CRIT_DISABLE_INPROCESS_DRAIN rows stuck in S3QueueObject with processingStatus='pending' | critpublisher.DrainKeys not invoked at end of run |
After the first ECR push, manually invoke the EventBridge target once to confirm Fargate task scheduling, IAM, secrets, and CloudWatch logging:
aws scheduler get-schedule --name go-{source}-{type}-processor --group-name vdb
aws events test-event-pattern ... # if pattern-based
# Easiest: tweak the cron to `cron(*/5 * * * ? *)` in terraform, apply, watch one
# fire in CloudWatch Logs `/ecs/vdb-scheduler/go-{source}-{type}-processor`,
# then revert.
Confirm the ECS task exits 0 with a Completed Slack message.
just go-{source}-{type}-backfill TARGET=prod LIMIT=10 FORCE=false
Should report NoWork (tracker fresh) or zero new rows. Re-running with
FORCE=true must not produce duplicate CVEMetadata, duplicate CVEAlias,
or duplicate Exploit/Kev/SnortRule rows. If it does, an upsert key is wrong.
Tick every box before merging. Anything unticked is a regression vector.
cmd/{source}-{type}-processor/main.go compiles and go vet is cleaninternal/ package wherever possible; new packages
have a parse_test.go golden testprocessor.StoreCVESourceData is the only path to CVEMetadata writes
(or db.UpsertCVEMetadata + explicit db.InsertAliases if direct)Containerfile.go-processors, task-manager.toml, terraform/go-schedules.tf,
.claude/hooks/post-push-ecr.sh, scripts/task-dashboard/cmd/ecr-build/targets.go,
justfile all updated with the locked name (character-identical)../saas/prisma/migrations/ if schema changed; SQL is
idempotent and uses CREATE INDEX CONCURRENTLY on large tablesStarted/Completed/Errored/NoWork events confirmed in channel.env.production (logs, DB
rows, Slack, idempotency check)