Go Processor Blueprint

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-scheduler cluster (terraform/ecs.tf). EventBridge Scheduler triggers each task on its cron; failed invocations are routed to the vdb-scheduler-dlq SQS dead-letter queue (terraform/logs.tf) with 14-day retention. The scanner processor is also in this cluster but triggered on-demand via ecs:RunTask.

0. Naming convention (decided before any code is written)

Refer to AGENTS.md § “Processor Naming Convention” for the authoritative rules. Pick one combination from each axis and lock it in:

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})

1. Phase 0 — Source reconnaissance (DO THIS FIRST)

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:

1. Cadence: How often does it change? (drives cron_match and frequency_secs)
2. Volume: How many records total / per delta? (drives cpu_units/memory_mb/expected_duration_minutes)
3. Identifier: CVE-prefixed natively, source-prefixed, or hybrid? (drives §6 cveId policy)
4. Backfillability: Can the full historical archive be pulled in one job, or is the source incremental-only? (drives whether to add a cmd/{source}-{type}-backfill/ companion)

2. Phase 1 — Pick the processor archetype

Each of the existing types has a representative reference processor. Read its main.go end-to-end before writing yours.

If the source feeds cloud-resource advisories (vendor:product:resource_type triples) include CRIT staging following the aws-security-bulletins-json-processor pattern.

3. Design document (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

4. Source → DB field mapping

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).

5. Identifier policy

• What becomes CVEMetadata.cveId? (CVE-prefix vs source-prefix — see §6)
• What goes into CVEAlias? Direction. Bundle-suppression risk.
• If minting source IDs (e.g. {SOURCE}-YYYY-NNNN, GCVE form GCVE-110-{SOURCE}-YYYY-NNNN): how is the sequence loaded (db.LoadMaxGcveSequence pattern) and persisted?

6. CRIT / VEX

• Which (Provider, Service, ResourceType) triples does this source map to?
• Are extended dictionaries needed (internal/critutil/dictionaries/extended/{vendor}.json)?
• VEX status derivation rules.

7. S3 / source-file archive layout

• Bucket, key prefix, payload schema (raw or normalised)
• Quarantine reasons enumerated

8. Error handling & Slack

• Per-record vs batch failure semantics
• Retry profile (count, backoff, transient vs fatal)
• Slack events emitted (Started/Completed/Errored/NoWork) and the stats dict shape

9. Performance

• Concurrency (fan-out width), per-request timeouts, rate-limit pacing
• Soft deadline behaviour (EXPECTED_DURATION_MINUTES, 5–10 min grace)
• Resume strategy (LoadProcessedHashes, listing SHA tracker, per-item hash)

10. Backfill

• Is cmd/{source}-{type}-backfill/ warranted? If yes, what differs from the scheduled binary?
• Justfile recipe and expected runtime on prod data

---

## 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)
}

5. Central pipeline contract — DO NOT BYPASS

Every advisory MUST flow through processor.StoreCVESourceData (file: internal/processor/pipeline.go:595). It composes:

1. db.UpsertCVEMetadata (composite key (cveId, source))
2. db.InsertDescriptions, InsertReferences, InsertMetrics, InsertProblemTypes, UpsertAffected, InsertVersions
3. db.EnrichAffectedWithDependency (opt-in)
4. db.InsertAliases(ctx, tx, cveID, source, aliases, logger) — see § “Alias Writes” in AGENTS.md

Direct 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.

6. Identifier policy — CVEMetadata.cveId vs CVEAlias

Decision is per-source and must be documented in §5 of the design doc.

Rule of thumb (matches existing processors):

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.

7. CRIT and VEX records

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.

8. Niche tables (Exploit / Kev / Snort / Yara / CrowdSec / VulnCheck / AIDiscovery)

Pick one (or more) based on source character. Helpers all live in internal/db/:

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.

9. Schemas (schemas/{source}_*.schema.json)

Add a JSON Schema only when:

1. The source emits a JSON/YAML format not already covered, AND
2. Either Go code will validate parsed payloads at runtime (//go:embed + jsonschema), or
3. The schema is needed as fixture/contract documentation for parse_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/.

10. Database migrations (../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:

1. Add the model edits to the appropriate file under ../saas/prisma/models/ (e.g. niche tables → exploit.prisma, threat-intel.prisma, ai-discovery.prisma; CRIT → crit.prisma).
2. Generate a new migration directory at ../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");

3. Apply locally via 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).

11. Infrastructure wiring — every place to touch

Every file edit listed here is required. Skipping any one of them breaks deploy or scheduling.

11.1 Containerfile.go-processors

Append 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:

• Git-baked data (≤12 MB repo): add a 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.
• Runtime git pull (large repo): omit the data stage; the binary calls processor.PullOrClone at runtime.
• Needs Chromium (Cloudflare bypass): use FROM alpine:3.21 final stage and RUN apk add --no-cache chromium. Reserved for cert-il-fetch-processor-class sources.
• Needs runtime git (small repo, want freshness on each ECS run): use FROM alpine:3.21 + RUN apk add --no-cache git ca-certificates (see ocaml-git-processor, hsec-git-processor).

11.2 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"

11.3 terraform/go-schedules.tf

module "{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.

11.4 .claude/hooks/post-push-ecr.sh

Add {source}-{type}-processor to TARGETS (alphabetical order).

11.5 scripts/task-dashboard/cmd/ecr-build/targets.go

Add to simpleTargets (no data stage) or dataTargets (git-baked).

11.6 justfile

Backfill 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).

11.7 .github/workflows/go-ecr-deploy.yml

The 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.

11.8 scripts/go-processors/entrypoint.sh

No edit required — already constructs DATABASE_URL/DATABASE_URL_READ from ECS secrets for every processor.

11.9 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.

12. Slack notifications (mandatory)

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.

13. Performance & resilience checklist

• HTTP client uses httpclient.New (HTTP/2 default) or httpclient.NewHTTP1 (only when source rejects HTTP/2 — common for older WAFs).
• Per-request context.WithTimeout (90 s default; tune for the source).
• Concurrent fetches behind a chan struct{} semaphore (4 workers is the default in existing processors).
• Per-record retry: 3 attempts, exponential backoff, classify unexpected EOF / connection reset as transient.
• Resume set loaded once at startup via db.LoadProcessedHashes.
• Tracker freshness check skipped under --force.
• Soft deadline checked before dispatching new work, not in the middle of a transaction.
• Pool sized via db.NewPool defaults (read=5, write=5). Do not create ad-hoc pgx.Pool instances.
• S3 archive happens after transaction commit (post-commit hook). Quarantine on parse/store failure.

14. Tests

• Place golden-file fixtures under 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.
• Run from repo root:

  cd scripts/go-processors && go test ./internal/{source}/... ./cmd/{source}-{type}-processor/...
  

15. Build & deploy verification

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

16. End-to-end run/verify (REQUIRED before declaring done)

This is the gate. The processor is not finished until these steps pass against real data and the resulting rows are inspected.

16.1 Local run against production DB (read-only verification)

# 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.

16.2 Database inspection

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:

16.3 ECS smoke 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.

16.4 Re-run idempotency

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.

17. Definition of done

Tick every box before merging. Anything unticked is a regression vector.

• Phase 0 recon documented in design doc § 2 (with curl evidence)
• cmd/{source}-{type}-processor/main.go compiles and go vet is clean
• Reuses an existing internal/ package wherever possible; new packages have a parse_test.go golden test
• processor.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)
• Migration added to ../saas/prisma/migrations/ if schema changed; SQL is idempotent and uses CREATE INDEX CONCURRENTLY on large tables
• CRIT staging + drain wired (only if cloud-resource source)
• Slack Started/Completed/Errored/NoWork events confirmed in channel
• § 16 run/verify executed end-to-end against .env.production (logs, DB rows, Slack, idempotency check)
• One Fargate fire observed in CloudWatch with exit code 0