.forgejo/workflows/build.yml

@@ -1,10 +1,6 @@
-# Build changed OSINT images, publish to the Forgejo container registry, then
+# Build all 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).
@@ -23,15 +19,6 @@ 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' }}
@@ -49,68 +36,6 @@ 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
@@ -134,7 +59,6 @@ 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
@@ -147,7 +71,6 @@ 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 }}"
@@ -158,7 +81,6 @@ jobs:
           docker push "${APP}:${SHA}"
 
       - name: Build news-scraper image
-        if: steps.plan.outputs.scraper == '1'
         run: |
           set -ex
           IMG="${{ steps.img.outputs.scraper }}"
@@ -169,7 +91,6 @@ jobs:
           docker push "${IMG}:${SHA}"
 
       - name: Build news-summarizer image
-        if: steps.plan.outputs.summarizer == '1'
         run: |
           set -ex
           IMG="${{ steps.img.outputs.summarizer }}"
@@ -180,17 +101,22 @@ 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; keeping existing local pg image"
+            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}"
           else
             echo "ERROR: cannot build or find osint-dashboard-pg image"
             exit 1
@@ -200,50 +126,39 @@ jobs:
         run: |
           set -ex
           cd "${GITHUB_WORKSPACE}"
-          APP="${{ steps.plan.outputs.app }}"
+          # Pull from Forgejo registry into local tags compose expects, then up.
-          SCRAPER="${{ steps.plan.outputs.scraper }}"
+          # Compose file still uses localhost/* for stable local names; we mirror
-          SUM="${{ steps.plan.outputs.summarizer }}"
+          # registry tags so a cold host can recover via docker pull.
-          PG="${{ steps.plan.outputs.pg }}"
+          REG="${{ steps.img.outputs.reg }}"
-          COMPOSE="${{ steps.plan.outputs.compose }}"
+          PUB="${{ steps.img.outputs.pub }}"
-
+          OWN="${{ env.OWNER }}"
-          SVCS=()
+          for name in osint-dashboard osint-dashboard-pg osint-news-scraper osint-news-summarizer; do
-          [ "$APP" = "1" ] && SVCS+=(app ingester camera-service)
+            docker pull "${REG}/${OWN}/${name}:latest" || true
-          [ "$SCRAPER" = "1" ] && SVCS+=(news-scraper)
+            docker tag "${REG}/${OWN}/${name}:latest" "localhost/${name}:latest" || true
-          [ "$SUM" = "1" ] && SVCS+=(news-summarizer)
+            docker tag "${REG}/${OWN}/${name}:latest" "${PUB}/${OWN}/${name}:latest" || true
-          if [ "$COMPOSE" = "1" ]; then
+          done
-            # compose/script change: bounce workers so env/command updates apply.
+          # Do NOT set COMPOSE_PROJECT_NAME differently — volumes must stay
-            # Still do not bounce Postgres.
+          # osint-dashboard_osint-pgdata (pinned by `name:` in compose).
-            for s in app ingester camera-service news-scraper news-summarizer; do
+          docker compose build --no-cache app ingester camera-service news-scraper news-summarizer || \
-              case " ${SVCS[*]} " in
+            docker compose build app ingester camera-service news-scraper news-summarizer
-                *" $s "*) ;;
+          # Name-pinned containers (container_name: osint-dashboard, …) collide
-                *) SVCS+=("$s") ;;
+          # when compose tries to create instead of recreate — e.g. leftover from
-              esac
+          # a different working_dir or a half-failed previous up. down + rm -f
-            done
+          # the known names, then up the full ingest profile.
-          fi
-
           chmod +x scripts/compose-reup.sh
-          if [ "$PG" = "1" ]; then
+          COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \
-            FORCE_RECREATE_DB=1 COMPOSE_PROJECT_NAME=osint-dashboard COMPOSE_PROFILES=ingest \
+            scripts/compose-reup.sh
-              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 deploy done; db image left in place unless pg=1"
+          echo "osint-dashboard deployed; images also on ${PUB}/${OWN}/"
 
       - name: Summary
         if: always()
         run: |
           {
-            echo "## Image plan"
+            echo "## Forgejo registry images"
-            echo "- app: \`${{ steps.plan.outputs.app }}\`"
+            echo "Pushed via ${{ steps.img.outputs.reg }} (loopback). Pull publicly:"
-            echo "- news-scraper: \`${{ steps.plan.outputs.scraper }}\`"
+            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard:latest\`"
-            echo "- news-summarizer: \`${{ steps.plan.outputs.summarizer }}\`"
+            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-dashboard-pg:latest\`"
-            echo "- pg (Timescale): \`${{ steps.plan.outputs.pg }}\`"
+            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-scraper:latest\`"
-            echo
+            echo "- \`${{ steps.img.outputs.pub }}/sirius/osint-news-summarizer:latest\`"
-            echo "Postgres is rebuilt/pulled only when \`Dockerfile.pg\` changes (or workflow_dispatch rebuild_pg)."
           } >> "$GITHUB_STEP_SUMMARY"

app/fire_sources.py

@@ -22,7 +22,6 @@ UTC date (YYYY-MM-DD).
 from __future__ import annotations
 
 import csv
-import hashlib
 import io
 import json
 import logging
@@ -45,11 +44,6 @@ 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 = (
@@ -92,34 +86,33 @@ def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None:
         return None
 
 
-def _hotspot_id(lat: float, lon: float, acq_iso: str, satellite: str) -> int:
+def parse_firms_csv(text: str) -> list[dict]:
-    return hash((round(lat, 5), round(lon, 5), acq_iso, satellite))
+    """Parse a FIRMS area CSV payload into normalized fire messages.
 
-
+    Returns one dict per hotspot with the fields stored in the ``fires`` table
-def parse_firms_csv_delta(
+    (acq_time already combined into a UTC ISO timestamp). Rows that don't look
-    text: str, skip_ids: set[int] | None = None,
+    like valid VIIRS detections are skipped rather than failing the whole poll.
-) -> 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.
     """
-    reader = csv.reader(io.StringIO(text))
+    rows = list(csv.reader(io.StringIO(text)))
-    header = None
+    if not rows:
-    for row in reader:
+        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):
         if row and row[0].strip().lower() == "latitude" and len(row) >= 4:
-            header = [c.strip().lower() for c in row]
+            header_idx = i
             break
-    if not header or "latitude" not in header or "longitude" not in header:
+    header = [c.strip().lower() for c in rows[header_idx]]
-        logger.warning(
+    # Guard against a header that isn't actually the FIRMS one.
-            "FIRMS payload does not look like a hotspot CSV (first row: %r)",
+    if "latitude" not in header or "longitude" not in header:
-            (header or [])[:6],
+        logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6])
-        )
+        return []
-        return [], set()
 
     points: list[dict] = []
-    ids: set[int] = set()
+    for row in rows[header_idx + 1:]:
-    for row in reader:
         if len(row) < len(header):
             continue
         rec = dict(zip(header, row))
@@ -130,19 +123,13 @@ def parse_firms_csv_delta(
         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_iso,
+            "acq_time": acq_time.isoformat(),
-            "satellite": sat,
+            "satellite": str(rec.get("satellite") or "").strip(),
             "instrument": str(rec.get("instrument") or "").strip(),
             "bright_ti5": _to_float(rec.get("bright_ti5")),
             "frp": _to_float(rec.get("frp")),
@@ -151,17 +138,6 @@ def parse_firms_csv_delta(
             "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
 
 
@@ -239,18 +215,11 @@ async def ingest_fires(bbox: str | None = None) -> int:
                     dataset, first_line,
                 )
                 continue
-            poll_key = (dataset, area, FIRMS_DAYS)
+            points = parse_firms_csv(text)
-            digest = hashlib.sha256(text.encode("utf-8", "surrogatepass")).digest()
+            published = await persist_hotspots(points)
-            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(ids), area, dataset, published,
+                len(points), 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.debug("skip duplicate event url=%s", key)
+                logger.info("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,35 +55,6 @@ 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}
 
@@ -312,9 +283,10 @@ async def ingest_eonet():
         resp.raise_for_status()
         data = resp.json()
     events = parse_eonet_events(data if isinstance(data, dict) else {})
-    published = await _publish_unknown("events.disaster", events)
+    for event in events:
-    logger.info("Ingested %d EONET events (%d already known)", published, len(events) - published)
+        await publish_event("events.disaster", event)
-    return published
+    logger.info("Ingested %d EONET events", len(events))
+    return len(events)
 
 
 # ─── CISA KEV ───────────────────────────────────────────────────────────
@@ -358,9 +330,10 @@ async def ingest_cisa_kev():
         resp.raise_for_status()
         data = resp.json()
     events = parse_cisa_kev(data if isinstance(data, dict) else {})
-    published = await _publish_unknown("events.disaster", events)
+    for event in events:
-    logger.info("Ingested %d CISA KEV rows (%d already known)", published, len(events) - published)
+        await publish_event("events.disaster", event)
-    return published
+    logger.info("Ingested %d CISA KEV rows", len(events))
+    return len(events)
 
 
 # ─── Social Signals (Twitter/X-like placeholder) ────────────────────────

docker-compose.yml

@@ -20,10 +20,8 @@ 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 build -f Dockerfile.pg -t localhost/osint-dashboard-pg:latest .
+    #   docker compose build db && docker compose up -d db
-    #   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:
@@ -54,7 +52,6 @@ services:
 
   nats:
     image: nats:2.10
-    pull_policy: missing
     platform: linux/arm64
     container_name: osint-nats
     restart: unless-stopped
@@ -70,7 +67,6 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
-    pull_policy: never
     container_name: osint-ingester
     restart: unless-stopped
     profiles: ["ingest"]
@@ -115,7 +111,6 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
-    pull_policy: never
     container_name: osint-dashboard
     restart: unless-stopped
     depends_on:
@@ -169,7 +164,6 @@ 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
@@ -189,7 +183,6 @@ services:
       dockerfile: Dockerfile
       platforms: ["linux/arm64"]
     image: localhost/osint-dashboard:latest
-    pull_policy: never
     container_name: osint-camera-scraper
     restart: unless-stopped
     profiles: ["ingest"]
@@ -228,7 +221,6 @@ 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"]
@@ -254,7 +246,6 @@ 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,15 +1,11 @@
 #!/usr/bin/env bash
-# Recreate selected OSINT compose services WITHOUT bouncing Postgres.
+# 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`.
 #
-# The old path was `compose down` + up, which stopped osint-db on every merge
+# Usage: scripts/compose-reup.sh
-# 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)
+#        COMPOSE_PROFILES     (default ingest — nats/ingester/news/cameras)
-#        FORCE_RECREATE_DB=1  also recreate db
 set -euo pipefail
 
 ROOT="$(cd "$(dirname "$0")/.." && pwd)"
@@ -18,57 +14,29 @@ cd "$ROOT"
 export COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-osint-dashboard}"
 PROFILE="${COMPOSE_PROFILES:-ingest}"
 
-DEFAULT_SVCS=(app ingester camera-service news-scraper news-summarizer)
+NAMES=(
-if [ "$#" -gt 0 ]; then
+  osint-dashboard
-  SVCS=("$@")
+  osint-db
-else
+  osint-nats
-  SVCS=("${DEFAULT_SVCS[@]}")
+  osint-ingester
-fi
+  osint-camera-scraper
-
+  osint-news-scraper
-if [ "${FORCE_RECREATE_DB:-0}" = "1" ]; then
+  osint-news-summarizer
-  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}"
 
-# Keep data-plane containers running (db / nats / titiler).
+# Stop compose-owned containers first. Foreign/name-pinned leftovers survive this.
-docker compose --profile "${PROFILE}" up -d --no-build --no-recreate db nats titiler || true
+docker compose --profile "${PROFILE}" down --remove-orphans || true
 
-if [ "${#SVCS[@]}" -eq 0 ]; then
+# Drop any leftover name-pinned containers compose does not own (the
-  docker compose --profile "${PROFILE}" ps
+# "Conflict. The container name is already in use" failure mode).
-  exit 0
+for c in "${NAMES[@]}"; do
-fi
+  if docker inspect "$c" >/dev/null 2>&1; then
-
+    echo "compose-reup: removing leftover ${c}"
-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 --no-deps "${SVCS[@]}"
+docker compose --profile "${PROFILE}" up -d --no-build "$@"
 docker compose --profile "${PROFILE}" ps

tests/test_fire_sources.py

@@ -131,100 +131,6 @@ 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,65 +64,3 @@ 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,11 +22,7 @@ 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
@@ -62,7 +58,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()) == 0
+    assert asyncio.run(ingest_fires()) == 5
     assert hits["n"] == 1