Merge pull request 'fix(ingest): skip known KEV/FIRMS rows; CI skips unchanged images' (#31) from feat/lean-ingest-ci-skip-unchanged into master

Reviewed-on: #31

.forgejo/workflows/build.yml

@@ -1,6 +1,10 @@
-# Build all OSINT images, publish to the Forgejo container registry, then
+# Build changed OSINT images, publish to the Forgejo container registry, then
 # redeploy on the Pi runner (docker.sock mounted).
 #
+# Unchanged images are skipped. Dockerfile.pg / osint-dashboard-pg is NOT
+# rebuilt or pulled on a normal merge — Postgres stays up. Rebuild it only
+# when Dockerfile.pg changes, or via workflow_dispatch rebuild_pg.
+#
 # Public pull host: forgejo.siriusdevops.com (NOT ghcr.io)
 # CI push host: 127.0.0.1:3000 — Cloudflare 413s layers ≳100MB on the public
 # hostname, even from the Pi (hairpins out through the tunnel).
@@ -19,6 +23,15 @@ on:
   push:
     branches: [main, master]
   workflow_dispatch:
+    inputs:
+      rebuild_pg:
+        description: Rebuild Timescale+PostGIS (Dockerfile.pg)
+        type: boolean
+        default: false
+      rebuild_all:
+        description: Rebuild every app image (ignore path filter)
+        type: boolean
+        default: false
 
 env:
   PUBLIC_REGISTRY: ${{ vars.FORGEJO_REGISTRY || 'forgejo.siriusdevops.com' }}
@@ -36,6 +49,68 @@ jobs:
     steps:
       - name: Checkout
         uses: https://code.forgejo.org/actions/checkout@v4
+        with:
+          fetch-depth: 50
+
+      - name: Plan image builds
+        id: plan
+        run: |
+          set -euo pipefail
+          APP=0
+          SCRAPER=0
+          SUM=0
+          PG=0
+          COMPOSE=0
+
+          mark() {
+            case "$1" in
+              Dockerfile.pg)
+                PG=1 ;;
+              Dockerfile|app/*|alembic/*|alembic.ini)
+                APP=1 ;;
+              news/scraper/*)
+                SCRAPER=1 ;;
+              news/summerizer/*)
+                SUM=1 ;;
+              docker-compose.yml|scripts/compose-reup.sh)
+                COMPOSE=1 ;;
+            esac
+          }
+
+          if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
+            APP=1; SCRAPER=1; SUM=1
+            if [ "${{ github.event.inputs.rebuild_all }}" = "true" ]; then
+              APP=1; SCRAPER=1; SUM=1; PG=1
+            fi
+            if [ "${{ github.event.inputs.rebuild_pg }}" = "true" ]; then
+              PG=1
+            fi
+          else
+            BEFORE="${{ github.event.before }}"
+            SHA="${GITHUB_SHA}"
+            ZEROS="0000000000000000000000000000000000000000"
+            if [ -z "$BEFORE" ] || [ "$BEFORE" = "$ZEROS" ]; then
+              echo "No previous SHA — build app images, skip pg"
+              APP=1; SCRAPER=1; SUM=1
+            elif ! git cat-file -e "${BEFORE}^{commit}" 2>/dev/null; then
+              echo "Previous SHA $BEFORE not in history — build app images, skip pg"
+              APP=1; SCRAPER=1; SUM=1
+            else
+              while IFS= read -r f; do
+                [ -z "$f" ] && continue
+                mark "$f"
+              done < <(git diff --name-only "$BEFORE" "$SHA")
+            fi
+          fi
+
+          {
+            echo "app=$APP"
+            echo "scraper=$SCRAPER"
+            echo "summarizer=$SUM"
+            echo "pg=$PG"
+            echo "compose=$COMPOSE"
+          } >> "$GITHUB_OUTPUT"
+          echo "plan app=$APP scraper=$SCRAPER summarizer=$SUM pg=$PG compose=$COMPOSE"
 
       - name: Image refs
         id: img
@@ -59,6 +134,7 @@ jobs:
           echo "SHA tag: $SHA"
 
       - name: Login to Forgejo registry
+        if: steps.plan.outputs.app == '1' || steps.plan.outputs.scraper == '1' || steps.plan.outputs.summarizer == '1' || steps.plan.outputs.pg == '1'
         run: |
           set -euo pipefail
           # GITHUB_TOKEN login "succeeds" but blob uploads 401 (Forgejo packages
@@ -71,6 +147,7 @@ jobs:
             -u sirius --password-stdin
 
       - name: Build application image (api / ingester / cameras)
+        if: steps.plan.outputs.app == '1'
         run: |
           set -ex
           APP="${{ steps.img.outputs.app }}"
@@ -81,6 +158,7 @@ jobs:
           docker push "${APP}:${SHA}"
 
       - name: Build news-scraper image
+        if: steps.plan.outputs.scraper == '1'
         run: |
           set -ex
           IMG="${{ steps.img.outputs.scraper }}"
@@ -91,6 +169,7 @@ jobs:
           docker push "${IMG}:${SHA}"
 
       - name: Build news-summarizer image
+        if: steps.plan.outputs.summarizer == '1'
         run: |
           set -ex
           IMG="${{ steps.img.outputs.summarizer }}"
@@ -101,22 +180,17 @@ jobs:
           docker push "${IMG}:${SHA}"
 
       - name: Build / refresh Timescale+PostGIS image
+        if: steps.plan.outputs.pg == '1'
         run: |
           set -ex
           PG="${{ steps.img.outputs.pg }}"
           SHA="${{ steps.img.outputs.sha }}"
-          # Prefer rebuild so registry always has a current pg image. If packagecloud
-          # is unreachable, fall back to whatever local image already exists.
           if docker build -f Dockerfile.pg -t "${PG}:latest" -t "${PG}:${SHA}" \
                -t "localhost/osint-dashboard-pg:latest" .; then
             docker push "${PG}:latest"
             docker push "${PG}:${SHA}"
           elif docker image inspect "localhost/osint-dashboard-pg:latest" >/dev/null 2>&1; then
-            echo "WARN: Dockerfile.pg build failed; retagging existing local pg image into registry"
-            docker tag "localhost/osint-dashboard-pg:latest" "${PG}:latest"
-            docker tag "localhost/osint-dashboard-pg:latest" "${PG}:${SHA}"
-            docker push "${PG}:latest"
-            docker push "${PG}:${SHA}"
+            echo "WARN: Dockerfile.pg build failed; keeping existing local pg image"
           else
             echo "ERROR: cannot build or find osint-dashboard-pg image"
             exit 1
@@ -126,39 +200,50 @@ jobs:
         run: |
           set -ex
           cd "${GITHUB_WORKSPACE}"
-          # Pull from Forgejo registry into local tags compose expects, then up.
-          # Compose file still uses localhost/* for stable local names; we mirror
-          # registry tags so a cold host can recover via docker pull.
-          REG="${{ steps.img.outputs.reg }}"
-          PUB="${{ steps.img.outputs.pub }}"
-          OWN="${{ env.OWNER }}"
-          for name in osint-dashboard osint-dashboard-pg osint-news-scraper osint-news-summarizer; do
-            docker pull "${REG}/${OWN}/${name}:latest" || true
-            docker tag "${REG}/${OWN}/${name}:latest" "localhost/${name}:latest" || true
-            docker tag "${REG}/${OWN}/${name}:latest" "${PUB}/${OWN}/${name}:latest" || true
-          done
-          # Do NOT set COMPOSE_PROJECT_NAME differently — volumes must stay
-          # osint-dashboard_osint-pgdata (pinned by `name:` in compose).
-          docker compose build --no-cache app ingester camera-service news-scraper news-summarizer || \
-            docker compose build app ingester camera-service news-scraper news-summarizer
-          # Name-pinned containers (container_name: osint-dashboard, …) collide
-          # when compose tries to create instead of recreate — e.g. leftover from
-          # a different working_dir or a half-failed previous up. down + rm -f
-          # the known names, then up the full ingest profile.
+          APP="${{ steps.plan.outputs.app }}"
+          SCRAPER="${{ steps.plan.outputs.scraper }}"
+          SUM="${{ steps.plan.outputs.summarizer }}"
+          PG="${{ steps.plan.outputs.pg }}"
+          COMPOSE="${{ steps.plan.outputs.compose }}"
+
+          SVCS=()
+          [ "$APP" = "1" ] && SVCS+=(app ingester camera-service)
+          [ "$SCRAPER" = "1" ] && SVCS+=(news-scraper)
+          [ "$SUM" = "1" ] && SVCS+=(news-summarizer)
+          if [ "$COMPOSE" = "1" ]; then
+            # compose/script change: bounce workers so env/command updates apply.
+            # Still do not bounce Postgres.
+            for s in app ingester camera-service news-scraper news-summarizer; do
+              case " ${SVCS[*]} " in
+                *" $s "*) ;;
+                *) SVCS+=("$s") ;;
+              esac
+            done
+          fi
+
           chmod +x scripts/compose-reup.sh
-          COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \
-            scripts/compose-reup.sh
+          if [ "$PG" = "1" ]; then
+            FORCE_RECREATE_DB=1 COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \
+              scripts/compose-reup.sh "${SVCS[@]}" db
+          elif [ "${#SVCS[@]}" -gt 0 ]; then
+            COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \
+              scripts/compose-reup.sh "${SVCS[@]}"
+          else
+            echo "No image or compose changes — leave running containers alone"
+            docker compose --profile ingest ps
+          fi
           docker image prune -f
-          echo "osint-dashboard deployed; images also on ${PUB}/${OWN}/"
+          echo "osint-dashboard deploy done; db image left in place unless pg=1"
 
       - name: Summary
         if: always()
         run: |
           {
-            echo "## Forgejo registry images"
-            echo "Pushed via ${{ steps.img.outputs.reg }} (loopback). Pull publicly:"
-            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard:latest\`"
-            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard-pg:latest\`"
-            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-scraper:latest\`"
-            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-summarizer:latest\`"
+            echo "## Image plan"
+            echo "- app: \`${{ steps.plan.outputs.app }}\`"
+            echo "- news-scraper: \`${{ steps.plan.outputs.scraper }}\`"
+            echo "- news-summarizer: \`${{ steps.plan.outputs.summarizer }}\`"
+            echo "- pg (Timescale): \`${{ steps.plan.outputs.pg }}\`"
+            echo
+            echo "Postgres is rebuilt/pulled only when \`Dockerfile.pg\` changes (or workflow_dispatch rebuild_pg)."
           } >> "$GITHUB_STEP_SUMMARY"

app/fire_sources.py

@@ -22,6 +22,7 @@ UTC date (YYYY-MM-DD).
 from __future__ import annotations
 
 import csv
+import hashlib
 import io
 import json
 import logging
@@ -44,6 +45,11 @@ from upstream_cache import firms_cache
 
 logger = logging.getLogger("osint.firms")
 
+# In-process poll state: skip byte-identical CSVs, persist only new hotspots.
+# Survives the 15-minute loop; one full ON CONFLICT after process start.
+_csv_digest: dict[tuple, bytes] = {}
+_seen_ids: dict[tuple, set[int]] = {}
+
 # ── FIRMS API ─────────────────────────────────────────────────────────────
 
 FIRMS_AREA_CSV = (
@@ -86,33 +92,34 @@ def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None:
         return None
 
 
-def parse_firms_csv(text: str) -> list[dict]:
-    """Parse a FIRMS area CSV payload into normalized fire messages.
+def _hotspot_id(lat: float, lon: float, acq_iso: str, satellite: str) -> int:
+    return hash((round(lat, 5), round(lon, 5), acq_iso, satellite))
 
-    Returns one dict per hotspot with the fields stored in the ``fires`` table
-    (acq_time already combined into a UTC ISO timestamp). Rows that don't look
-    like valid VIIRS detections are skipped rather than failing the whole poll.
+
+def parse_firms_csv_delta(
+    text: str, skip_ids: set[int] | None = None,
+) -> tuple[list[dict], set[int]]:
+    """Parse FIRMS CSV; optionally drop hotspots already seen this process.
+
+    Returns (new_or_all_points, ids_for_every_valid_row). Streaming — does not
+    materialize the raw CSV as a list of lists.
     """
-    rows = list(csv.reader(io.StringIO(text)))
-    if not rows:
-        return []
-
-    # Locate the real header row. FIRMS normally returns the CSV header first,
-    # but occasionally prepends a legend/info line, so scan until we see the
-    # canonical header.
-    header_idx = 0
-    for i, row in enumerate(rows):
+    reader = csv.reader(io.StringIO(text))
+    header = None
+    for row in reader:
         if row and row[0].strip().lower() == "latitude" and len(row) >= 4:
-            header_idx = i
+            header = [c.strip().lower() for c in row]
             break
-    header = [c.strip().lower() for c in rows[header_idx]]
-    # Guard against a header that isn't actually the FIRMS one.
-    if "latitude" not in header or "longitude" not in header:
-        logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6])
-        return []
+    if not header or "latitude" not in header or "longitude" not in header:
+        logger.warning(
+            "FIRMS payload does not look like a hotspot CSV (first row: %r)",
+            (header or [])[:6],
+        )
+        return [], set()
 
     points: list[dict] = []
-    for row in rows[header_idx + 1:]:
+    ids: set[int] = set()
+    for row in reader:
         if len(row) < len(header):
             continue
         rec = dict(zip(header, row))
@@ -123,13 +130,19 @@ def parse_firms_csv(text: str) -> list[dict]:
         acq_time = normalize_acq_time(rec.get("acq_date"), rec.get("acq_time"))
         if acq_time is None:
             continue
+        sat = str(rec.get("satellite") or "").strip()
+        acq_iso = acq_time.isoformat()
+        hid = _hotspot_id(lat, lon, acq_iso, sat)
+        ids.add(hid)
+        if skip_ids is not None and hid in skip_ids:
+            continue
         points.append({
             "latitude": lat,
             "longitude": lon,
             "brightness": _to_float(rec.get("bright_ti4")),
             "confidence": str(rec.get("confidence") or "").strip(),
-            "acq_time": acq_time.isoformat(),
-            "satellite": str(rec.get("satellite") or "").strip(),
+            "acq_time": acq_iso,
+            "satellite": sat,
             "instrument": str(rec.get("instrument") or "").strip(),
             "bright_ti5": _to_float(rec.get("bright_ti5")),
             "frp": _to_float(rec.get("frp")),
@@ -138,6 +151,17 @@ def parse_firms_csv(text: str) -> list[dict]:
             "track": _to_float(rec.get("track")),
             "version": str(rec.get("version") or "").strip(),
         })
+    return points, ids
+
+
+def parse_firms_csv(text: str) -> list[dict]:
+    """Parse a FIRMS area CSV payload into normalized fire messages.
+
+    Returns one dict per hotspot with the fields stored in the ``fires`` table
+    (acq_time already combined into a UTC ISO timestamp). Rows that don't look
+    like valid VIIRS detections are skipped rather than failing the whole poll.
+    """
+    points, _ids = parse_firms_csv_delta(text)
     return points
 
 
@@ -215,11 +239,18 @@ async def ingest_fires(bbox: str | None = None) -> int:
                     dataset, first_line,
                 )
                 continue
-            points = parse_firms_csv(text)
-            published = await persist_hotspots(points)
+            poll_key = (dataset, area, FIRMS_DAYS)
+            digest = hashlib.sha256(text.encode("utf-8", "surrogatepass")).digest()
+            if _csv_digest.get(poll_key) == digest:
+                logger.info("FIRMS %s CSV unchanged, skip parse/insert", dataset)
+                continue
+            points, ids = parse_firms_csv_delta(text, skip_ids=_seen_ids.get(poll_key))
+            published = await persist_hotspots(points) if points else 0
+            _csv_digest[poll_key] = digest
+            _seen_ids[poll_key] = ids
             total_published += published
             logger.info(
                 "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d",
-                len(points), area, dataset, published,
+                len(ids), area, dataset, published,
             )
     return total_published

app/ingestor.py

@@ -213,7 +213,7 @@ async def ingest_event(msg: dict):
             claimed = await session.execute(dedup)
             if not claimed.rowcount:
                 await session.commit()
-                logger.info("skip duplicate event url=%s", key)
+                logger.debug("skip duplicate event url=%s", key)
                 return None
         result = await session.execute(events_table.insert().values(**event_row))
         await session.commit()

app/sources.py

@@ -55,6 +55,35 @@ def event_dedup_key(msg: dict) -> str | None:
     return url or None
 
 
+async def existing_event_urls(urls: list[str]) -> set[str]:
+    """URLs already claimed in event_dedup. Empty input -> empty set."""
+    if not urls:
+        return set()
+    from sqlalchemy import select
+
+    from database import async_session
+    from models import event_dedup as event_dedup_table
+
+    async with async_session() as session:
+        result = await session.execute(
+            select(event_dedup_table.c.url).where(event_dedup_table.c.url.in_(urls))
+        )
+        return {row[0] for row in result}
+
+
+async def _publish_unknown(subject: str, events: list[dict]) -> int:
+    """Publish only events whose URL is not already in event_dedup."""
+    keys = [event_dedup_key(e) for e in events]
+    known = await existing_event_urls([k for k in keys if k])
+    published = 0
+    for event, key in zip(events, keys):
+        if key and key in known:
+            continue
+        await publish_event(subject, event)
+        published += 1
+    return published
+
+
 def _ua_headers() -> dict[str, str]:
     return {"User-Agent": OSINT_USER_AGENT}
 
@@ -283,10 +312,9 @@ async def ingest_eonet():
         resp.raise_for_status()
         data = resp.json()
     events = parse_eonet_events(data if isinstance(data, dict) else {})
-    for event in events:
-        await publish_event("events.disaster", event)
-    logger.info("Ingested %d EONET events", len(events))
-    return len(events)
+    published = await _publish_unknown("events.disaster", events)
+    logger.info("Ingested %d EONET events (%d already known)", published, len(events) - published)
+    return published
 
 
 # ─── CISA KEV ───────────────────────────────────────────────────────────
@@ -330,10 +358,9 @@ async def ingest_cisa_kev():
         resp.raise_for_status()
         data = resp.json()
     events = parse_cisa_kev(data if isinstance(data, dict) else {})
-    for event in events:
-        await publish_event("events.disaster", event)
-    logger.info("Ingested %d CISA KEV rows", len(events))
-    return len(events)
+    published = await _publish_unknown("events.disaster", events)
+    logger.info("Ingested %d CISA KEV rows (%d already known)", published, len(events) - published)
+    return published
 
 
 # ─── Social Signals (Twitter/X-like placeholder) ────────────────────────

docker-compose.yml

@@ -20,8 +20,10 @@ services:
     # networks (and ISP abuse-mitigation blackholes) block, so rebuilding it
     # on every CI deploy made the pipeline flaky. Rebuild manually when the
     # base image or extensions need bumping:
-    #   docker compose build db && docker compose up -d db
+    #   docker build -f Dockerfile.pg -t localhost/osint-dashboard-pg:latest .
+    #   FORCE_RECREATE_DB=1 scripts/compose-reup.sh db
     image: localhost/osint-dashboard-pg:latest
+    pull_policy: never
     container_name: osint-db
     restart: unless-stopped
     environment:
@@ -52,6 +54,7 @@ services:
 
   nats:
     image: nats:2.10
+    pull_policy: missing
     platform: linux/arm64
     container_name: osint-nats
     restart: unless-stopped
@@ -67,6 +70,7 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
+    pull_policy: never
     container_name: osint-ingester
     restart: unless-stopped
     profiles: ["ingest"]
@@ -111,6 +115,7 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
+    pull_policy: never
     container_name: osint-dashboard
     restart: unless-stopped
     depends_on:
@@ -164,6 +169,7 @@ services:
   # published on host loopback 127.0.0.1:8001 only.
   titiler:
     image: ghcr.io/developmentseed/titiler:latest
+    pull_policy: missing
     container_name: osint-titiler
     platform: linux/arm64
     restart: unless-stopped
@@ -183,6 +189,7 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
+    pull_policy: never
     container_name: osint-camera-scraper
     restart: unless-stopped
     profiles: ["ingest"]
@@ -221,6 +228,7 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-news-scraper:latest
+    pull_policy: never
     container_name: osint-news-scraper
     restart: unless-stopped
     profiles: ["ingest"]
@@ -246,6 +254,7 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-news-summarizer:latest
+    pull_policy: never
     container_name: osint-news-summarizer
     restart: unless-stopped
     profiles: ["ingest"]

scripts/compose-reup.sh

@@ -1,11 +1,15 @@
 #!/usr/bin/env bash
-# Recreate the OSINT compose stack without container_name collisions.
-# Named volumes (osint-dashboard_osint-pgdata, camera-snapshots) are kept —
-# never pass -v to `compose down`.
+# Recreate selected OSINT compose services WITHOUT bouncing Postgres.
 #
-# Usage: scripts/compose-reup.sh
+# The old path was `compose down` + up, which stopped osint-db on every merge
+# even when Dockerfile.pg did not change. Name-pinned leftovers are still
+# removed, but only for the services we are actually replacing.
+#
+# Usage: scripts/compose-reup.sh [compose-service ...]
+#        (default: app ingester camera-service news-scraper news-summarizer)
 # Env:   COMPOSE_PROJECT_NAME (default osint-dashboard)
-#        COMPOSE_PROFILES     (default ingest — nats/ingester/news/cameras)
+#        COMPOSE_PROFILES     (default ingest)
+#        FORCE_RECREATE_DB=1  also recreate db
 set -euo pipefail
 
 ROOT="$(cd "$(dirname "$0")/.." && pwd)"
@@ -14,29 +18,57 @@ cd "$ROOT"
 export COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-osint-dashboard}"
 PROFILE="${COMPOSE_PROFILES:-ingest}"
 
-NAMES=(
-  osint-dashboard
-  osint-db
-  osint-nats
-  osint-ingester
-  osint-camera-scraper
-  osint-news-scraper
-  osint-news-summarizer
+DEFAULT_SVCS=(app ingester camera-service news-scraper news-summarizer)
+if [ "$#" -gt 0 ]; then
+  SVCS=("$@")
+else
+  SVCS=("${DEFAULT_SVCS[@]}")
+fi
+
+if [ "${FORCE_RECREATE_DB:-0}" = "1" ]; then
+  SVCS+=(db)
+fi
+
+# Never recreate db unless it was requested.
+FILTERED=()
+for svc in "${SVCS[@]}"; do
+  if [ "$svc" = "db" ] && [ "${FORCE_RECREATE_DB:-0}" != "1" ]; then
+    echo "compose-reup: skipping db (set FORCE_RECREATE_DB=1 to bounce Postgres)"
+    continue
+  fi
+  FILTERED+=("$svc")
+done
+SVCS=("${FILTERED[@]}")
+
+declare -A CONTAINER_NAME=(
+  [app]=osint-dashboard
+  [ingester]=osint-ingester
+  [camera-service]=osint-camera-scraper
+  [news-scraper]=osint-news-scraper
+  [news-summarizer]=osint-news-summarizer
+  [db]=osint-db
+  [nats]=osint-nats
+  [titiler]=osint-titiler
 )
 
 echo "compose-reup: project=${COMPOSE_PROJECT_NAME} profile=${PROFILE} dir=${ROOT}"
+echo "compose-reup: recreate=${SVCS[*]:-none}"
 
-# Stop compose-owned containers first. Foreign/name-pinned leftovers survive this.
-docker compose --profile "${PROFILE}" down --remove-orphans || true
+# Keep data-plane containers running (db / nats / titiler).
+docker compose --profile "${PROFILE}" up -d --no-build --no-recreate db nats titiler || true
 
-# Drop any leftover name-pinned containers compose does not own (the
-# "Conflict. The container name is already in use" failure mode).
-for c in "${NAMES[@]}"; do
-  if docker inspect "$c" >/dev/null 2>&1; then
-    echo "compose-reup: removing leftover ${c}"
+if [ "${#SVCS[@]}" -eq 0 ]; then
+  docker compose --profile "${PROFILE}" ps
+  exit 0
+fi
+
+for svc in "${SVCS[@]}"; do
+  c="${CONTAINER_NAME[$svc]:-}"
+  if [ -n "$c" ] && docker inspect "$c" >/dev/null 2>&1; then
+    echo "compose-reup: replacing ${c}"
     docker rm -f "$c" >/dev/null
   fi
 done
 
-docker compose --profile "${PROFILE}" up -d --no-build "$@"
+docker compose --profile "${PROFILE}" up -d --no-build --no-deps "${SVCS[@]}"
 docker compose --profile "${PROFILE}" ps

tests/test_fire_sources.py

@@ -131,6 +131,100 @@ def test_ingest_fires_uses_keystore_key(monkeypatch):
     assert any("VIIRS_NOAA21_NRT" in u for u in captured["urls"])
 
 
+def _reset_firms_poll_state():
+    from upstream_cache import firms_cache
+    import fire_sources
+
+    firms_cache.clear()
+    if hasattr(fire_sources, "_csv_digest"):
+        fire_sources._csv_digest.clear()
+    if hasattr(fire_sources, "_seen_ids"):
+        fire_sources._seen_ids.clear()
+
+
+def _fake_firms_http(monkeypatch, bodies_by_call: list[str] | None = None, body: str = SAMPLE_CSV):
+    hits = {"n": 0}
+
+    class FakeResp:
+        def __init__(self, text):
+            self.text = text
+
+        def raise_for_status(self):
+            pass
+
+    class FakeClient:
+        def __init__(self, **kw):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def get(self, url):
+            idx = hits["n"]
+            hits["n"] += 1
+            if bodies_by_call is not None:
+                text = bodies_by_call[min(idx, len(bodies_by_call) - 1)]
+            else:
+                text = body
+            return FakeResp(text)
+
+    monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32)
+    monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"])
+    monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient)
+    return hits
+
+
+def test_ingest_fires_skips_unchanged_csv(monkeypatch):
+    """Same FIRMS CSV must not be re-parsed into a 100k-row ON CONFLICT insert."""
+    _reset_firms_poll_state()
+    hits = _fake_firms_http(monkeypatch)
+    persisted = []
+
+    async def fake_persist(points):
+        persisted.append(len(points))
+        return len(points)
+
+    monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist)
+
+    assert asyncio.run(ingest_fires()) == 5
+    assert persisted == [5]
+    firms_cache_hits = hits["n"]
+    persisted.clear()
+    assert asyncio.run(ingest_fires()) == 0
+    assert persisted == []
+    # TTL cache may skip HTTP; either way we must not persist again.
+    assert hits["n"] >= firms_cache_hits
+
+
+def test_ingest_fires_persists_only_new_hotspots(monkeypatch):
+    """When the CSV grows, persist the delta — not the whole 2-day dump."""
+    _reset_firms_poll_state()
+    extra = (
+        SAMPLE_CSV
+        + "16.00000,-12.00000,340.00,0.40,0.40,2025-06-06,1500,N20,VIIRS,h,2.0NRT,310.00,8.00,D\n"
+    )
+    hits = _fake_firms_http(monkeypatch, bodies_by_call=[SAMPLE_CSV, extra])
+    persisted = []
+
+    async def fake_persist(points):
+        persisted.append([p["latitude"] for p in points])
+        return len(points)
+
+    monkeypatch.setattr("fire_sources.persist_hotspots", fake_persist)
+
+    from upstream_cache import firms_cache
+
+    assert asyncio.run(ingest_fires()) == 5
+    firms_cache.clear()  # force the next poll to see the grown CSV
+    persisted.clear()
+    assert asyncio.run(ingest_fires()) == 1
+    assert persisted == [[16.0]]
+    assert hits["n"] == 2
+
+
 def _async_return(value):
     async def inner():
         return value

tests/test_new_sources.py

@@ -64,3 +64,65 @@ def test_parse_cisa_kev_emits_cve_url_no_coords():
     assert "cisa-kev" in ev["tags"]
     assert "CVE-2024-1234" in ev["tags"]
     assert ev["raw"]["cveID"] == "CVE-2024-1234"
+
+
+def test_ingest_cisa_kev_does_not_republish_known_nist_urls(monkeypatch):
+    """Producer must not push the whole KEV catalog to NATS every cycle."""
+    import asyncio
+
+    from sources import ingest_cisa_kev
+
+    payload = {
+        "vulnerabilities": [
+            {
+                "cveID": "CVE-2024-1111",
+                "vulnerabilityName": "old",
+                "dateAdded": "2024-01-01",
+                "shortDescription": "already in db",
+            },
+            {
+                "cveID": "CVE-2024-2222",
+                "vulnerabilityName": "new",
+                "dateAdded": "2024-06-01",
+                "shortDescription": "not in db yet",
+            },
+        ]
+    }
+
+    class FakeResp:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return payload
+
+    class FakeClient:
+        def __init__(self, **kw):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def get(self, url):
+            return FakeResp()
+
+    published: list[str] = []
+
+    async def fake_publish(subject, event):
+        published.append(event["url"])
+
+    known = {"https://nvd.nist.gov/vuln/detail/CVE-2024-1111"}
+
+    async def fake_existing(urls):
+        return {u for u in urls if u in known}
+
+    monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient)
+    monkeypatch.setattr("sources.publish_event", fake_publish)
+    monkeypatch.setattr("sources.existing_event_urls", fake_existing, raising=False)
+
+    n = asyncio.run(ingest_cisa_kev())
+    assert n == 1
+    assert published == ["https://nvd.nist.gov/vuln/detail/CVE-2024-2222"]

tests/test_upstream_cache.py

@@ -22,7 +22,11 @@ def test_firms_and_rss_caches_are_ttlcache():
 
 
 def test_ingest_fires_hits_http_once_within_ttl(monkeypatch):
+    from fire_sources import _csv_digest, _seen_ids
+
     firms_cache.clear()
+    _csv_digest.clear()
+    _seen_ids.clear()
     monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32)
     monkeypatch.setenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT")
     # fire_sources already imported FIRMS_DATASETS — patch the module attr
@@ -58,7 +62,7 @@ def test_ingest_fires_hits_http_once_within_ttl(monkeypatch):
     monkeypatch.setattr("fire_sources.persist_hotspots", fake_publish)
 
     assert asyncio.run(ingest_fires()) == 5
-    assert asyncio.run(ingest_fires()) == 5
+    assert asyncio.run(ingest_fires()) == 0
     assert hits["n"] == 1