Merge pull request 'feat(map): quiet HUD chrome — VIIRS default, collapsed layer rail' (#49) from feat/hud-chrome-quiet into master

Reviewed-on: #49

.env.example

@@ -31,20 +31,6 @@ NOMINATIM_URL=https://nominatim.openstreetmap.org
 NOMINATIM_MIN_INTERVAL=1.1
 SNAPSHOT_TTL_SECONDS=300
 
-# ── masscan active camera discovery (host-level systemd service, NOT compose) ─
-# Continuous rolling sweep for open RTSP port 554 across a range. Runs on the
-# Pi host via deploy/osint-masscan.service (needs root + raw sockets). Results
-# land in the same `cameras` table as the scraper (discovery_source=masscan).
-# NOTE: 200 pps is the residential-safe default. 1k/10k pps saturated a home
-# uplink. A full 0.0.0.0/0 sweep at 200 pps takes ~8 months (rolling).
-MASSCAN_RANGE=0.0.0.0/0
-MASSCAN_PORTS=554
-MASSCAN_RATE=200
-MASSCAN_RETRIES=1
-MASSCAN_WAIT=0
-MASSCAN_EXCLUDEFILE=/etc/osint-dashboard/masscan-excludes.txt
-MASSCAN_FLUSH_EVERY=250
-
 # ── NASA FIRMS (active fire / hotspot ingest) ──────────────────────────────
 # MAP_KEY is FREE — get one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/
 # (1-minute signup, no payment). Leave blank to keep fire ingest idle.

alembic/versions/009_vessels.py

@@ -0,0 +1,41 @@
+"""vessels — daily VesselAPI snapshots for DVR as-of
+
+Revision ID: 009_vessels
+Revises: 008_summary_kind
+Create Date: 2026-08-29
+"""
+
+from alembic import op
+
+revision = "009_vessels"
+down_revision = "008_summary_kind"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS vessels (
+            mmsi     TEXT NOT NULL,
+            poll_at  TIMESTAMPTZ NOT NULL,
+            lat      DOUBLE PRECISION NOT NULL,
+            lon      DOUBLE PRECISION NOT NULL,
+            heading  DOUBLE PRECISION,
+            speed    DOUBLE PRECISION,
+            label    TEXT,
+            extra    JSONB,
+            PRIMARY KEY (mmsi, poll_at)
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS vessels")

alembic/versions/010_bbox_gist.py

@@ -0,0 +1,29 @@
+"""GIST bbox indexes for events/fires map-pan queries.
+
+Revision ID: 010_bbox_gist
+Revises: 009_vessels
+Create Date: 2026-09-01
+"""
+
+from alembic import op
+
+revision = "010_bbox_gist"
+down_revision = "009_vessels"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_events_geom_gist ON events "
+        "USING gist (ST_SetSRID(ST_MakePoint(location_lon, location_lat), 4326))"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_fires_geom_gist ON fires "
+        "USING gist (ST_SetSRID(ST_MakePoint(longitude, latitude), 4326))"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_fires_geom_gist")
+    op.execute("DROP INDEX IF EXISTS ix_events_geom_gist")

alembic/versions/011_geofence_alerts_fence.py

@@ -0,0 +1,24 @@
+"""geofence_alerts (geofence_id, created_at DESC) for fence-scoped hit log
+
+Revision ID: 011_geofence_alerts_fence
+Revises: 010_bbox_gist
+Create Date: 2026-09-01
+"""
+
+from alembic import op
+
+revision = "011_geofence_alerts_fence"
+down_revision = "010_bbox_gist"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_geofence_alerts_fence_created "
+        "ON geofence_alerts (geofence_id, created_at DESC)"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_geofence_alerts_fence_created")

app/bg_jobs.py

@@ -1,50 +1,19 @@
-"""Background masscan / ffmpeg — never block a FastAPI request on a scan.
+"""Background ffmpeg — never block a FastAPI request on a frame grab.
 
-masscan is capped at 200 pps (home uplink saturates at 1k+). ffmpeg frame
+ffmpeg frame grabs are scheduled with asyncio.create_task and shared per URL.
-grabs are scheduled with asyncio.create_task and shared per URL.
 """
 
 from __future__ import annotations
 
 import asyncio
-import logging
 import shutil
 from cachetools import TTLCache
 
-logger = logging.getLogger("osint.bg_jobs")
-
-MASSCAN_PPS_CAP = 200
-
-_masscan_task: asyncio.Task | None = None
 _ffmpeg_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
 _ffmpeg_tasks: dict[str, asyncio.Task] = {}
 _FFMPEG = shutil.which("ffmpeg")
 
 
-def schedule_masscan_pass() -> bool:
-    """Kick one capped masscan pass. Returns False if a pass is already running."""
-    global _masscan_task
-    if _masscan_task is not None and not _masscan_task.done():
-        return False
-    _masscan_task = asyncio.create_task(_run_masscan_capped())
-    return True
-
-
-async def _run_masscan_capped() -> None:
-    import masscan_config as cfg
-    from run_masscan_service import _verify_excludefile, run_pass
-
-    orig = cfg.MASSCAN_RATE
-    if orig > MASSCAN_PPS_CAP:
-        logger.warning("capping masscan rate %s pps -> %s", orig, MASSCAN_PPS_CAP)
-        cfg.MASSCAN_RATE = MASSCAN_PPS_CAP
-    try:
-        _verify_excludefile()
-        await run_pass()
-    finally:
-        cfg.MASSCAN_RATE = orig
-
-
 def cached_ffmpeg_jpeg(url: str) -> bytes | None:
     return _ffmpeg_cache.get(url)
 

app/camera_config.py

@@ -16,6 +16,8 @@ CALTRANS_CCTV_URLS = tuple(
     f"https://cwwp2.dot.ca.gov/data/d{n}/cctv/cctvStatusD{n:02d}.json"
     for n in range(1, 13)
 )
+# MDOT MiDrive official DOT CCTV list (fields carry rendered HTML).
+MDOT_CAMERA_URL = "https://mdotjboss.state.mi.us/MiDrive/camera/list"
 _DEFAULT_SOURCE_URL = ",".join((
     # Publicly published open-camera list (markdown bullets of stream URLs).
     "https://raw.githubusercontent.com/fury999io/public-ip-cams/main/README.md",
@@ -25,6 +27,10 @@ _DEFAULT_SOURCE_URL = ",".join((
     "https://raw.githubusercontent.com/willytop8/Live-Environment-Streams/main/streams.geojson",
     # Official Caltrans CWWP2 JPEG + HLS CCTV (districts 1–12).
     *CALTRANS_CCTV_URLS,
+    # Oregon DOT TripCheck public CCTV JPEG inventory (Esri JSON).
+    "https://www.tripcheck.com/Scripts/map/data/cctvinventory.js",
+    # Official MDOT MiDrive CCTV (JPEG stills, Michigan).
+    MDOT_CAMERA_URL,
 ))
 CAMERA_SOURCE_URLS = [
     u.strip()
@@ -55,3 +61,15 @@ SNAPSHOT_TIMEOUT = float(os.getenv("SNAPSHOT_TIMEOUT", "8.0"))
 
 # NATS subject cameras are published on (consumed by the shared ingester).
 CAMERA_NATS_SUBJECT = os.getenv("CAMERA_NATS_SUBJECT", "events.camera")
+
+
+# ── UDOT IBI 511 traffic cameras ──────────────────────────────────────────
+# DataTables endpoint (POST form-encoded; server caps at 100 rows/page no
+# matter what `length` is sent). No API key. Snapshot stills live at a stable
+# /map/Cctv/{id} URL — same URL always serves the latest frame, so we store
+# the URL and never scrape every frame ourselves.
+UDOT_IBI_URL = "https://prod-ut.ibi511.com/List/GetData/Cameras"
+UDOT_IBI_BASE = "https://prod-ut.ibi511.com"
+UDOT_IBI_PAGE_SIZE = 100
+# Safety cap on pages per cycle so a runaway recordsTotal cannot fan out.
+UDOT_IBI_MAX_PAGES = int(os.getenv("UDOT_IBI_MAX_PAGES", "40"))

app/camera_preview.py

@@ -1,7 +1,7 @@
 """Resolve a browser-renderable preview for a camera.
 
 HTTP/MJPEG cameras already expose a snapshot_url the existing proxy can
-stream. masscan finds are stored as `rtsp://IP/` with no snapshot_url, so
+stream. Some scraper sources store `rtsp://` URLs with no snapshot_url, so
 the map popup used to skip the <img> entirely and the leftover source link
 handed the browser an rtsp:// URL (which opens VLC).
 
@@ -42,16 +42,6 @@ _HTTP_PATHS = (
     "/tmpfs/auto.jpg",
 )
 
-# Browser-playable MJPEG paths the /stream proxy can pass through.
-_MJPEG_PATHS = (
-    "/mjpg/video.mjpg",
-    "/video.mjpg",
-    "/cgi-bin/mjpg/video.cgi",
-    "/axis-cgi/mjpg/video.cgi",
-    "/nphMotionJpeg",
-    "/mjpeg.cgi",
-)
-
 _FFMPEG = shutil.which("ffmpeg")
 
 
@@ -87,55 +77,6 @@ async def _http_get_image(url: str, timeout: float = 2.5) -> bytes | None:
         return None
 
 
-async def _http_feed_url(url: str, timeout: float = 2.5) -> str | None:
-    """Return url if it looks like an unauthenticated image/MJPEG feed."""
-    try:
-        async with httpx.AsyncClient(
-            timeout=timeout, follow_redirects=True,
-            headers={"User-Agent": USER_AGENT},
-        ) as c:
-            async with c.stream("GET", url) as r:
-                if r.status_code != 200:
-                    return None
-                ctype = (r.headers.get("content-type") or "").lower()
-                if "html" in ctype or ctype.startswith("text/"):
-                    return None
-                if any(x in ctype for x in ("image/", "multipart", "mjpeg", "octet-stream")):
-                    # Read a little to reject empty/error bodies.
-                    chunk = b""
-                    async for b in r.aiter_bytes():
-                        chunk += b
-                        if len(chunk) >= 64:
-                            break
-                    if len(chunk) < 64:
-                        return None
-                    if b"html" in chunk[:64].lower():
-                        return None
-                    return url
-    except Exception:  # noqa: BLE001
-        return None
-    return None
-
-
-async def probe_public_feed(host: str) -> str | None:
-    """Unauthenticated HTTP still or MJPEG URL for this host, or None.
-
-    Used at masscan ingest time so dead RTSP-only hosts never hit the map.
-    No credentials, no RTSP path-walking (too slow / rarely public).
-    """
-    urls = [f"http://{host}{p}" for p in _HTTP_PATHS]
-    urls.append(f"http://{host}:8080/shot.jpg")
-    urls.extend(f"http://{host}{p}" for p in _MJPEG_PATHS)
-    results = await asyncio.gather(
-        *(_http_feed_url(u) for u in urls),
-        return_exceptions=True,
-    )
-    for url, hit in zip(urls, results):
-        if isinstance(hit, str) and hit:
-            return hit
-    return None
-
-
 async def ffmpeg_snapshot(url: str, timeout: float = 8.0) -> bytes | None:
     """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails.
 

app/camera_scraper.py

@@ -25,6 +25,7 @@ import hashlib
 import ipaddress
 import json
 import logging
+import math
 import re
 import time
 from datetime import datetime, timezone
@@ -37,6 +38,7 @@ from camera_config import (
     CAMERA_SOURCE_URLS, CAMERA_REQUEST_DELAY, CAMERA_MAX_PER_SOURCE,
     NOMINATIM_URL, NOMINATIM_MIN_INTERVAL, USER_AGENT,
     SNAPSHOT_CACHE_DIR, SNAPSHOT_TTL_SECONDS, SNAPSHOT_TIMEOUT,
+    UDOT_IBI_URL, UDOT_IBI_BASE, UDOT_IBI_PAGE_SIZE, UDOT_IBI_MAX_PAGES,
 )
 from camera_models import cameras
 from database import async_session
@@ -114,6 +116,15 @@ class RateLimitedClient:
         self._last[host] = time.monotonic()
         return await self.client.get(url, **kw)
 
+    async def post(self, url: str, **kw) -> httpx.Response:
+        host = urlparse(url).netloc
+        now = time.monotonic()
+        wait = self._last.get(host, 0.0) + self._delay - now
+        if wait > 0:
+            await asyncio.sleep(wait)
+        self._last[host] = time.monotonic()
+        return await self.client.post(url, **kw)
+
     async def aclose(self):
         await self.client.aclose()
 
@@ -379,6 +390,201 @@ def parse_caltrans_json(text: str, source_name: str) -> list[dict]:
     return out
 
 
+# ── UDOT IBI 511 ──────────────────────────────────────────────────────────
+# Utah bbox (lat 36.9–42.1, lon -114.2–-108.9). WKT is `POINT (lng lat)`.
+_UDOT_IBI_MIN_LAT, _UDOT_IBI_MAX_LAT = 36.9, 42.1
+_UDOT_IBI_MIN_LON, _UDOT_IBI_MAX_LON = -114.2, -108.9
+_UDOT_WKT_POINT_RE = re.compile(
+    r"POINT\s*\(\s*(-?\d+(?:\.\d+)?)\s+(-?\d+(?:\.\d+)?)\s*\)", re.I,
+)
+
+
+def parse_udot_ibi_page(text: str, source_name: str = "udot") -> list[dict]:
+    """Parse one UDOT IBI 511 DataTables camera page (`{"data": [...]}`).
+
+    Skips rows whose first image is `blocked` or `disabled`, and drops any
+    point outside the Utah bbox. The `/map/Cctv/{id}` URL is a stable identity
+    (always serves the latest frame), so it is stored as both source_url and
+    snapshot_url — we never scrape frames ourselves.
+    """
+    try:
+        payload = json.loads(text)
+    except (json.JSONDecodeError, ValueError):
+        return []
+    rows = payload.get("data") if isinstance(payload, dict) else None
+    if not isinstance(rows, list):
+        return []
+    out: list[dict] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        cam_id = row.get("id")
+        images = row.get("images") or []
+        if cam_id is None or not images:
+            continue
+        img = images[0] or {}
+        if img.get("blocked") or img.get("disabled"):
+            continue
+        lon = lat = None
+        try:
+            wkt = (row.get("latLng") or {}).get("geography") or {}
+            wkt = wkt.get("wellKnownText") or ""
+            m = _UDOT_WKT_POINT_RE.match(str(wkt).strip())
+            if m:
+                lon, lat = float(m.group(1)), float(m.group(2))
+        except (AttributeError, TypeError, ValueError):
+            lon = lat = None
+        if lat is None or lon is None:
+            continue
+        if not (_UDOT_IBI_MIN_LAT <= lat <= _UDOT_IBI_MAX_LAT
+                and _UDOT_IBI_MIN_LON <= lon <= _UDOT_IBI_MAX_LON):
+            continue
+        snap = f"{UDOT_IBI_BASE}/map/Cctv/{cam_id}"
+        roadway, direction, location = (
+            row.get("roadway"), row.get("direction"), row.get("location"),
+        )
+        name = ", ".join(
+            str(b) for b in (roadway, direction, location)
+            if b and str(b).strip() and str(b).strip().lower() != "unknown"
+        ) or None
+        out.append({
+            "source_url": snap,
+            "snapshot_url": snap,
+            "discovery_source": source_name,
+            "location_lat": lat,
+            "location_lon": lon,
+            "location_name": name,
+            "vendor": "UDOT",
+            "device_type": "http",
+            "raw": {
+                "udot_id": cam_id,
+                "agency": row.get("source"),
+                "source_id": row.get("sourceId"),
+                "roadway": roadway,
+                "direction": direction,
+            },
+        })
+    return out
+
+
+# Oregon DOT TripCheck inventory bounding box (approx state extent).
+ODOT_BBOX = (41.9, 46.3, -124.6, -116.4)  # lat_min, lat_max, lon_min, lon_max
+
+
+def parse_odot_json(text: str, source_name: str) -> list[dict]:
+    """Parse Oregon DOT TripCheck cctvinventory Esri-style JSON.
+
+    Store the JPEG still as snapshot_url (map thumbs); never RTSP. Keep only
+    rows with finite coordinates inside Oregon and a usable filename.
+    """
+    try:
+        payload = json.loads(text)
+    except (json.JSONDecodeError, ValueError):
+        return []
+    lat_min, lat_max, lon_min, lon_max = ODOT_BBOX
+    out: list[dict] = []
+    for feat in payload.get("features") or []:
+        attrs = (feat or {}).get("attributes") or {}
+        filename = (attrs.get("filename") or "").strip()
+        if not filename:
+            continue
+        try:
+            lat = float(attrs.get("latitude"))
+            lon = float(attrs.get("longitude"))
+        except (TypeError, ValueError):
+            continue
+        if not (math.isfinite(lat) and math.isfinite(lon)):
+            continue
+        if not (lat_min <= lat <= lat_max and lon_min <= lon <= lon_max):
+            continue
+        jpeg = f"https://tripcheck.com/RoadCams/cams/{filename}"
+        title = (attrs.get("title") or "").strip()
+        out.append({
+            "source_url": jpeg,
+            "snapshot_url": jpeg,
+            "discovery_source": "odot",
+            "location_lat": lat,
+            "location_lon": lon,
+            "location_name": title or None,
+            "vendor": "ODOT",
+            "device_type": "http",
+        })
+    return out
+
+
+# MDOT MiDrive field extractors (fields carry rendered HTML).
+_MDOT_LAT_RE = re.compile(r"lat=(-?\d+(?:\.\d+)?)", re.I)
+_MDOT_LON_RE = re.compile(r"lon=(-?\d+(?:\.\d+)?)", re.I)
+_MDOT_ID_RE = re.compile(r"[?&]id=(\d+)", re.I)
+_MDOT_IMG_RE = re.compile(r'<img[^>]+src=["\']([^"\']+)["\']', re.I)
+
+# Michigan bbox (docs/osiris-ideas.md §3.2): lat 41.6–48.3, lon -90.5–-82.1.
+MDOT_LAT_RANGE = (41.6, 48.3)
+MDOT_LON_RANGE = (-90.5, -82.1)
+
+
+def parse_mdot_json(text: str, source_name: str) -> list[dict]:
+    """Parse MDOT MiDrive `camera/list` JSON (fields carry rendered HTML).
+
+    Coordinates and the stable id live in the `county` field's map link
+    (`/MiDrive/map?...lat=&lon=&id=`); the `image` field carries an `<img>`
+    whose src is the JPEG still. Out-of-bbox and coord-less rows are dropped.
+    """
+    try:
+        payload = json.loads(text)
+    except (json.JSONDecodeError, ValueError):
+        return []
+    if not isinstance(payload, list):
+        return []
+    out: list[dict] = []
+    for row in payload:
+        if not isinstance(row, dict):
+            continue
+        county_html = row.get("county") or ""
+        m_lat = _MDOT_LAT_RE.search(county_html)
+        m_lon = _MDOT_LON_RE.search(county_html)
+        m_id = _MDOT_ID_RE.search(county_html)
+        if not (m_lat and m_lon and m_id):
+            continue  # missing coordinates / stable id → drop
+        try:
+            lat = float(m_lat.group(1))
+            lon = float(m_lon.group(1))
+        except ValueError:
+            continue
+        if not (MDOT_LAT_RANGE[0] <= lat <= MDOT_LAT_RANGE[1]
+                and MDOT_LON_RANGE[0] <= lon <= MDOT_LON_RANGE[1]):
+            continue  # out of Michigan bbox → drop
+        img_m = _MDOT_IMG_RE.search(row.get("image") or "")
+        if not img_m:
+            continue
+        snap = img_m.group(1).strip()
+        low = snap.lower()
+        if not (low.startswith("http://") or low.startswith("https://")):
+            continue
+        if low.startswith("rtsp"):
+            continue
+        cam_id = m_id.group(1)
+        route = (row.get("route") or "").strip()
+        loc = (row.get("location") or "").strip().lstrip("@").strip()
+        county_name = county_html.split("<a", 1)[0].strip()
+        bits = [
+            f"{route} @ {loc}" if (route and loc) else (route or loc or None),
+            county_name or None,
+        ]
+        name = ", ".join(b for b in bits if b) or None
+        out.append({
+            "source_url": f"https://mdotjboss.state.mi.us/MiDrive/camera/{cam_id}",
+            "snapshot_url": snap,
+            "discovery_source": "mdot",
+            "location_lat": lat,
+            "location_lon": lon,
+            "location_name": name,
+            "vendor": "MDOT",
+            "device_type": "http",
+        })
+    return out
+
+
 def parse_live_streams_geojson(text: str, source_name: str) -> list[dict]:
     """Parse willytop8/Live-Environment-Streams GeoJSON.
 
@@ -490,6 +696,10 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder,
     body = resp.text
     if "cwwp2.dot.ca.gov" in src_url or "cctvStatus" in src_url:
         cams = parse_caltrans_json(body, name)
+    elif "cctvinventory" in src_url or "tripcheck.com" in src_url:
+        cams = parse_odot_json(body, name)
+    elif "mdotjboss.state.mi.us" in src_url or "/MiDrive/camera/list" in src_url:
+        cams = parse_mdot_json(body, name)
     elif ("getCameraDataByLoc" in src_url
             or ("json" in ctype and '"locs"' in body[:4000] and '"cams"' in body[:8000])):
         cams = parse_alertwest_json(body, name)
@@ -549,6 +759,54 @@ async def scrape_source(client: RateLimitedClient, geo: Geocoder,
     return out
 
 
+# ── UDOT IBI 511 paginated fetcher ────────────────────────────────────────
+
+async def scrape_udot_ibi(client: RateLimitedClient) -> list[dict]:
+    """Page through the UDOT IBI 511 DataTables endpoint and normalize.
+
+    POSTs `start`/`length` form fields (server caps at 100 rows/page), walking
+    pages until `recordsTotal` is exhausted or UDOT_IBI_MAX_PAGES is hit.
+    """
+    out: list[dict] = []
+    seen: set[str] = set()
+    start = 0
+    for _ in range(UDOT_IBI_MAX_PAGES):
+        try:
+            resp = await client.post(
+                UDOT_IBI_URL,
+                data={
+                    "start": str(start),
+                    "length": str(UDOT_IBI_PAGE_SIZE),
+                    "lang": "en-US",
+                },
+                headers={"X-Requested-With": "XMLHttpRequest"},
+            )
+            resp.raise_for_status()
+            body = resp.text
+        except Exception:  # noqa: BLE001
+            logger.exception("failed to fetch UDOT IBI page start=%d", start)
+            break
+        try:
+            payload = json.loads(body)
+        except ValueError:
+            logger.warning("UDOT IBI non-JSON response at start=%d", start)
+            break
+        total = int(payload.get("recordsTotal") or 0)
+        rows = payload.get("data") or []
+        if not isinstance(rows, list) or not rows:
+            break
+        for cam in parse_udot_ibi_page(body, "udot"):
+            if cam["source_url"] in seen:
+                continue
+            seen.add(cam["source_url"])
+            out.append(cam)
+        if start + len(rows) >= total:
+            break
+        start += len(rows)
+    logger.info("UDOT IBI yielded %d cameras", len(out))
+    return out
+
+
 # ── Persistence ────────────────────────────────────────────────────────────
 
 async def upsert_cameras(cams: list[dict]) -> int:
@@ -600,6 +858,7 @@ async def run_cycle() -> int:
     try:
         results = await asyncio.gather(
             *(scrape_source(client, geo, s) for s in CAMERA_SOURCE_URLS),
+            scrape_udot_ibi(client),
             return_exceptions=True,
         )
         all_cams: list[dict] = []

app/config.py

@@ -71,6 +71,9 @@ FIRMS_DATASETS = [d.strip() for d in _FIRMS_DATASETS_RAW.split(",") if d.strip()
 OSINT_USER_AGENT = os.getenv(
     "OSINT_USER_AGENT", "osint-dashboard/1.0 (self-hosted; lancewalters94@gmail.com)"
 )
+# Nominatim reverse (GET /api/place). Camera scraper has its own copy in camera_config.
+NOMINATIM_URL = os.getenv("NOMINATIM_URL", "https://nominatim.openstreetmap.org")
+NOMINATIM_MIN_INTERVAL = float(os.getenv("NOMINATIM_MIN_INTERVAL", "1.0"))
 
 # Self-hosted TiTiler (warps Sentinel-1 signed COGs into XYZ tiles on the Pi).
 # TITILER_PUBLIC_BASE is the same-origin path prefix the browser hits through

app/conflicts.py

@@ -0,0 +1,168 @@
+"""Curated OSINT conflict-zone catalog + point-in-bbox event counting.
+
+A static, human-curated list of active conflict theatres (war / high /
+elevated). Purely descriptive — this is a catalog, not a live feed and not a
+scrape of LiveUAMap or any other source. Severity and descriptions are
+editorial judgement kept short and factual.
+
+Each zone carries an internal ``bbox`` (``min_lat, min_lon, max_lat, max_lon``)
+used only to count pre-existing geocoded news/GDELT/``/api/news/map`` rows that
+fall inside it. The bbox is not part of the API response; callers get the
+``eventCount`` roll-up instead.
+
+Never call an upstream API from here — event counts come from rows already in
+the local database (``events`` with geocoords + ``news_items`` map pins).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+# id → zone. ``lat``/``lon`` is the fly-to anchor; ``bbox`` is the internal
+# count window in ``min_lat, min_lon, max_lat, max_lon`` order.
+_ZONES: tuple[dict, ...] = (
+    {
+        "id": "ukraine",
+        "label": "Ukraine",
+        "severity": "war",
+        "lat": 48.5,
+        "lon": 31.0,
+        "description": "Full-scale Russian invasion since 2022; active front lines in the east and south.",
+        "bbox": (44.3, 22.1, 52.4, 40.2),
+    },
+    {
+        "id": "gaza",
+        "label": "Gaza",
+        "severity": "war",
+        "lat": 31.4,
+        "lon": 34.4,
+        "description": "Israel–Hamas war; sustained fighting and a severe humanitarian crisis in the Gaza Strip.",
+        "bbox": (31.0, 34.1, 31.8, 34.7),
+    },
+    {
+        "id": "sudan",
+        "label": "Sudan",
+        "severity": "war",
+        "lat": 15.5,
+        "lon": 30.0,
+        "description": "Civil war between the SAF and RSF since 2023, with mass displacement across the country.",
+        "bbox": (8.7, 21.8, 22.0, 38.6),
+    },
+    {
+        "id": "myanmar",
+        "label": "Myanmar",
+        "severity": "war",
+        "lat": 21.5,
+        "lon": 96.0,
+        "description": "Post-2021 coup conflict pitting the junta against resistance and ethnic armed groups.",
+        "bbox": (9.5, 92.2, 28.5, 101.2),
+    },
+    {
+        "id": "drc",
+        "label": "DR Congo",
+        "severity": "war",
+        "lat": -1.5,
+        "lon": 28.0,
+        "description": "Eastern DRC conflict involving M23 and other armed groups; heavy displacement around Goma.",
+        "bbox": (-5.0, 26.0, 3.0, 31.0),
+    },
+    {
+        "id": "yemen",
+        "label": "Yemen",
+        "severity": "war",
+        "lat": 15.5,
+        "lon": 47.5,
+        "description": "Protracted Houthi–government/coalition war with one of the world's worst humanitarian emergencies.",
+        "bbox": (12.6, 42.5, 19.0, 54.0),
+    },
+    {
+        "id": "syria",
+        "label": "Syria",
+        "severity": "war",
+        "lat": 34.5,
+        "lon": 38.5,
+        "description": "Multi-sided civil war; government, opposition, and external actors continue to engage.",
+        "bbox": (32.3, 35.7, 37.3, 42.4),
+    },
+    {
+        "id": "lebanon",
+        "label": "Lebanon",
+        "severity": "high",
+        "lat": 33.9,
+        "lon": 35.9,
+        "description": "Israel–Hezbollah hostilities with periodic escalation along the southern border.",
+        "bbox": (33.0, 35.0, 34.7, 36.6),
+    },
+    {
+        "id": "sahel",
+        "label": "Sahel",
+        "severity": "high",
+        "lat": 14.5,
+        "lon": 0.0,
+        "description": "Jihadist insurgencies across Mali, Burkina Faso, and Niger destabilising the central Sahel.",
+        "bbox": (10.0, -10.0, 20.0, 12.0),
+    },
+    {
+        "id": "somalia",
+        "label": "Somalia",
+        "severity": "high",
+        "lat": 6.0,
+        "lon": 45.0,
+        "description": "Al-Shabaab insurgency against the federal government and security forces.",
+        "bbox": (-2.0, 41.0, 12.0, 51.5),
+    },
+    {
+        "id": "red_sea",
+        "label": "Red Sea",
+        "severity": "high",
+        "lat": 18.0,
+        "lon": 40.0,
+        "description": "Houthi attacks on commercial shipping transiting the Red Sea corridor.",
+        "bbox": (12.0, 34.0, 22.0, 44.0),
+    },
+    {
+        "id": "taiwan_strait",
+        "label": "Taiwan Strait",
+        "severity": "elevated",
+        "lat": 24.5,
+        "lon": 119.5,
+        "description": "Heightened military standoff between China and Taiwan, including deterrence patrols.",
+        "bbox": (21.9, 117.0, 26.5, 122.0),
+    },
+    {
+        "id": "korean_dmz",
+        "label": "Korean DMZ",
+        "severity": "elevated",
+        "lat": 38.3,
+        "lon": 127.0,
+        "description": "Heavily fortified inter-Korean border with periodic tensions and military drills.",
+        "bbox": (37.5, 126.0, 39.0, 128.5),
+    },
+)
+
+SEVERITIES: frozenset[str] = frozenset({"war", "high", "elevated"})
+
+
+def conflict_zones() -> list[dict]:
+    """Return a fresh shallow copy of the catalog (callers must not mutate)."""
+    return [dict(z) for z in _ZONES]
+
+
+def zone_event_stats(
+    points: list[tuple[float, float, datetime | None]],
+    bbox: tuple[float, float, float, float],
+) -> tuple[int, datetime | None]:
+    """Count points inside ``bbox`` and return (count, latest timestamp).
+
+    ``points`` is an iterable of ``(lat, lon, ts)``; ``ts`` may be ``None``.
+    ``bbox`` is ``(min_lat, min_lon, max_lat, max_lon)``.
+    """
+    min_lat, min_lon, max_lat, max_lon = bbox
+    count = 0
+    latest: datetime | None = None
+    for lat, lon, ts in points:
+        if min_lat <= lat <= max_lat and min_lon <= lon <= max_lon:
+            count += 1
+            if ts is not None and (latest is None or ts > latest):
+                latest = ts
+    return count, latest

app/geofence.py

@@ -319,3 +319,154 @@ async def record_and_notify(
         except Exception:
             pass
     return sent
+
+
+async def list_alerts(
+    *,
+    geofence_id: str | None = None,
+    since: datetime | None = None,
+    until: datetime | None = None,
+    source_kind: str | None = None,
+    limit: int = 100,
+) -> list[dict]:
+    """Filterable hit log. Empty list if the DB is down — never raises."""
+    where = ["TRUE"]
+    params: dict[str, Any] = {"limit": int(limit)}
+    if geofence_id:
+        where.append("geofence_id = CAST(:geofence_id AS uuid)")
+        params["geofence_id"] = geofence_id
+    if since is not None:
+        where.append("created_at >= :since")
+        params["since"] = since
+    if until is not None:
+        where.append("created_at <= :until")
+        params["until"] = until
+    if source_kind:
+        where.append("source_kind = :source_kind")
+        params["source_kind"] = source_kind
+    sql = f"""
+        SELECT id::text, geofence_id::text, source_kind, entity_id,
+               lat, lon, payload, created_at
+        FROM geofence_alerts
+        WHERE {' AND '.join(where)}
+        ORDER BY created_at DESC
+        LIMIT :limit
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(text(sql), params)).mappings().all()
+        out = []
+        for r in rows:
+            item = dict(r)
+            if item.get("created_at") is not None:
+                item["created_at"] = item["created_at"].isoformat()
+            out.append(item)
+        return out
+    except Exception:
+        return []
+
+
+async def get_geofence(gid: str) -> dict | None:
+    current = next((f for f in _cache if f["id"] == gid), None)
+    if current is not None:
+        return current
+    try:
+        await refresh_cache()
+    except Exception:
+        return None
+    return next((f for f in _cache if f["id"] == gid), None)
+
+
+def _marker_from_track(row) -> dict:
+    from live_layers import to_marker
+
+    extra = {"bucket": row["bucket"].isoformat() if row.get("bucket") else None, "dvr": True}
+    return to_marker(
+        row["id"], row["lat"], row["lon"],
+        heading=row.get("heading"), speed=row.get("speed"),
+        label=row.get("label") or row["id"],
+        extra=extra,
+    )
+
+
+async def _cagg_inside(gid: str, kind: str, bucket: datetime, limit: int = 2000) -> list[dict]:
+    table = "aircraft_tracks_1min" if kind == "aircraft" else "vessel_tracks_1min"
+    id_col = "hex" if kind == "aircraft" else "mmsi"
+    sql = f"""
+        SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket
+        FROM {table}
+        WHERE bucket = :bucket
+          AND ST_Intersects(
+                (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
+                ST_SetSRID(ST_MakePoint(lon, lat), 4326)
+              )
+        LIMIT :limit
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                text(sql), {"bucket": bucket, "gid": gid, "limit": limit},
+            )).mappings().all()
+        return [
+            _marker_from_track(r)
+            for r in rows
+            if r["lat"] is not None and r["lon"] is not None
+        ]
+    except Exception:
+        return []
+
+
+async def _fires_inside(gid: str, ts: datetime, limit: int = 2000) -> list[dict]:
+    from tracks import minute_bucket
+
+    bucket = minute_bucket(ts)
+    t1 = bucket + timedelta(minutes=1)
+    sql = """
+        SELECT latitude, longitude, brightness, confidence, acq_time, satellite,
+               instrument, bright_ti5, frp, daynight
+        FROM fires
+        WHERE acq_time >= :t0 AND acq_time < :t1
+          AND ST_Intersects(
+                (SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
+                ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)
+              )
+        LIMIT :limit
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                text(sql),
+                {"t0": bucket, "t1": t1, "gid": gid, "limit": limit},
+            )).mappings().all()
+        out = []
+        for r in rows:
+            item = dict(r)
+            if item.get("acq_time") is not None:
+                item["acq_time"] = item["acq_time"].isoformat()
+            out.append(item)
+        return out
+    except Exception:
+        return []
+
+
+async def snapshot_at(gid: str, ts: datetime) -> dict | None:
+    """Positions inside the fence at time T. None if the fence is missing.
+
+    Does not persist or notify. Empty lists if track/fire queries fail.
+    """
+    fence = await get_geofence(gid)
+    if fence is None:
+        return None
+    from tracks import minute_bucket
+
+    bucket = minute_bucket(ts)
+    aircraft = await _cagg_inside(gid, "aircraft", bucket)
+    vessels = await _cagg_inside(gid, "vessel", bucket)
+    fires = await _fires_inside(gid, ts)
+    return {
+        "geofence_id": gid,
+        "timestamp": ts.isoformat(),
+        "aircraft": aircraft,
+        "vessels": vessels,
+        "fires": fires,
+    }

app/live_layers.py

@@ -51,7 +51,9 @@ PC_SAS_TOKEN = "https://planetarycomputer.microsoft.com/api/sas/v1/token/sentine
 # nginx vhost. Relative template — Leaflet resolves it against the page origin,
 # so the browser never touches a raw loopback port or titiler.xyz.
 TITILER_COG_TILES = f"{TITILER_PUBLIC_BASE}/cog/tiles/WebMercatorQuad/{{z}}/{{x}}/{{y}}"
-SENTINEL1_TTL = 20 * 60  # 15–30 min quota-friendly window
+SENTINEL1_TTL = 6 * 3600  # S-1 revisit is 6–12 days; cache the COG all afternoon
+SENTINEL1_CELL = 2.0  # degrees — pan/zoom inside a cell reuses the same scene
+SENTINEL1_STAC_LIMIT = 8
 SENTINEL1_ATTRIBUTION = "Copernicus Sentinel-1 / Microsoft Planetary Computer"
 
 # GPSJAM (John Wiseman / ADS-B Exchange): daily H3 hexes of aircraft nav
@@ -95,6 +97,11 @@ _MAX_VESSELS = 6000
 # Last ADS-B snapshot + WFIGS points for fire↔tanker correlation.
 aircraft_last_known: dict[str, dict] = {}
 fire_last_known: list[dict] = []
+# Last-known counts for the cheap GET /api/stats HUD counter. Updated by the
+# upstream fetchers so the stats endpoint never does its own network/SQL fan-out
+# for these layers; reads are O(1) in-process.
+train_count: int = 0
+nws_alert_count: int = 0
 
 
 def overlay_catalog() -> dict:
@@ -148,6 +155,24 @@ def overlay_catalog() -> dict:
             "endpoint": "/api/map/gpsjam",
             "attribution": "GPSJAM / John Wiseman / ADS-B Exchange",
         },
+        "satellites": {
+            "id": "satellites",
+            "kind": "points",
+            "endpoint": "/api/satellites",
+            "attribution": "CelesTrak (GP JSON / SGP4)",
+        },
+        "infra_nuclear": {
+            "id": "infra_nuclear",
+            "kind": "points",
+            "endpoint": "/api/infrastructure?types=nuclear",
+            "attribution": "OpenStreetMap contributors / Overpass API",
+        },
+        "conflicts": {
+            "id": "conflicts",
+            "kind": "points",
+            "endpoint": "/api/conflicts",
+            "attribution": "Curated OSINT conflict catalog",
+        },
     }
 
 
@@ -964,6 +989,8 @@ async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dic
         return transform_amtraker(await _get_json(AMTRAKER_TRAINS))
 
     rows = await _ttl_get("amtraker:trains", 20.0, _load)
+    global train_count
+    train_count = len(rows)
     if bbox:
         minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
         return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
@@ -1117,6 +1144,8 @@ async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
             logger.warning("NWS alerts fetch failed: %s", exc)
             nws_ok = False
             nws_fc = {"features": []}
+        global nws_alert_count
+        nws_alert_count = len(nws_fc.get("features") or [])
         sbw_fc = await _ttl_get("iem:sbw", 45.0, _load_iem)
         features = []
         for feat in nws_fc.get("features") or []:
@@ -1244,25 +1273,71 @@ def sentinel1_tile_url(signed_cog: str) -> str:
     return f"{TITILER_COG_TILES}?{params}"
 
 
+def feature_bbox(feat: dict) -> list[float] | None:
+    """STAC Feature bbox as [minlon, minlat, maxlon, maxlat], or None."""
+    raw = feat.get("bbox") if isinstance(feat, dict) else None
+    if isinstance(raw, (list, tuple)) and len(raw) >= 4:
+        try:
+            return [float(raw[0]), float(raw[1]), float(raw[2]), float(raw[3])]
+        except (TypeError, ValueError):
+            pass
+    geom = (feat or {}).get("geometry") or {}
+    coords = geom.get("coordinates") if isinstance(geom, dict) else None
+    if not coords:
+        return None
+    lons: list[float] = []
+    lats: list[float] = []
+
+    def _walk(node: Any) -> None:
+        if isinstance(node, (list, tuple)) and node and isinstance(node[0], (int, float)):
+            lons.append(float(node[0]))
+            lats.append(float(node[1]))
+        elif isinstance(node, (list, tuple)):
+            for child in node:
+                _walk(child)
+
+    _walk(coords)
+    if not lons:
+        return None
+    return [min(lons), min(lats), max(lons), max(lats)]
+
+
+def pick_sentinel_feature(features: list, lon: float, lat: float) -> dict | None:
+    """Prefer the scene whose bbox covers the viewport center; else first."""
+    if not features:
+        return None
+    for feat in features:
+        bb = feature_bbox(feat)
+        if bb and bb[0] <= lon <= bb[2] and bb[1] <= lat <= bb[3]:
+            return feat
+    return features[0]
+
+
 async def fetch_sentinel1(bbox: str) -> dict | None:
     """Most recent Sentinel-1 GRD COG for a viewport, signed and TiTiler-ready.
 
     Returns the overlay tile-template dict, or ``None`` when no GRD imagery
     covers the bbox in the last 7 days (caller maps to 404). Queries Planetary
-    Computer only when called; cached per quantized bbox + UTC day.
+    Computer only when called; cached per 2° cell + UTC day so pan/zoom inside
+    the same region reuses the COG instead of picking a neighbouring swath.
     """
     minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+    qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(
+        minlon, minlat, maxlon, maxlat, step=SENTINEL1_CELL,
+    )
     day = datetime.now(timezone.utc).date().isoformat()
-    key = f"sentinel1:{day}:{bbox_cell_key(bbox)}"
+    key = f"sentinel1:{day}:{qminlon:.4f},{qminlat:.4f},{qmaxlon:.4f},{qmaxlat:.4f}"
+    clon = (qminlon + qmaxlon) / 2.0
+    clat = (qminlat + qmaxlat) / 2.0
 
     async def _load() -> dict | None:
         now = datetime.now(timezone.utc)
         week_ago = now - timedelta(days=7)
         payload = {
             "collections": ["sentinel-1-grd"],
-            "bbox": [minlon, minlat, maxlon, maxlat],
+            "bbox": [qminlon, qminlat, qmaxlon, qmaxlat],
             "datetime": f"{week_ago.isoformat()}/{now.isoformat()}",
-            "limit": 1,
+            "limit": SENTINEL1_STAC_LIMIT,
             "sortby": [{"field": "datetime", "direction": "desc"}],
         }
         data = await _pc_call(_post_json(PC_STAC_SEARCH, json=payload))
@@ -1270,7 +1345,9 @@ async def fetch_sentinel1(bbox: str) -> dict | None:
         if not features:
             return None
 
-        feat = features[0]
+        feat = pick_sentinel_feature(features, clon, clat)
+        if not feat:
+            return None
         assets = feat.get("assets") or {}
         chosen_href: str | None = None
         polarization: str | None = None
@@ -1299,6 +1376,7 @@ async def fetch_sentinel1(bbox: str) -> dict | None:
             "datetime": props.get("datetime") or feat.get("datetime"),
             "polarization": polarization,
             "attribution": SENTINEL1_ATTRIBUTION,
+            "bbox": feature_bbox(feat),
         }
 
     return await _ttl_get(key, float(SENTINEL1_TTL), _load)
@@ -1378,3 +1456,104 @@ async def fetch_gpsjam(date: str) -> dict:
         return gpsjam_csv_to_geojson(resp.text)
 
     return await _ttl_get(f"gpsjam:{date}", GPSJAM_TTL, _load)
+# ── Infrastructure (Overpass) ───────────────────────────────────────────────
+
+
+OVERPASS_INTERPRETER = "https://overpass-api.de/api/interpreter"
+# One in-flight query per quantized bbox (the per-key lock in _ttl_get). Overpass
+# asks for a 25s server timeout in-band; the client gives it 30s of headroom.
+OVERPASS_TIMEOUT = httpx.Timeout(30.0, connect=5.0)
+INFRA_TTL = 24 * 3600  # 24h per quantized bbox — static infrastructure
+
+# `types=` enum. Nuclear ships first; military/hospital slot in behind the same
+# query template without touching the transport. Overpass bbox is
+# (south, west, north, east), i.e. (minlat, minlon, maxlat, maxlon).
+_INFRA_QUERIES: dict[str, str] = {
+    "nuclear": (
+        '[out:json][timeout:25];\n'
+        'nwr["power"="plant"]["plant:source"="nuclear"]({bbox});\n'
+        'out center;'
+    ),
+}
+
+
+def infra_query(type_: str, minlon: float, minlat: float, maxlon: float, maxlat: float) -> str:
+    """Render one Overpass query with the bbox substituted in south,west,north,east."""
+    bbox = f"{minlat},{minlon},{maxlat},{maxlon}"
+    return _INFRA_QUERIES[type_].replace("{bbox}", bbox)
+
+
+def normalize_infra_element(elem: dict, type_: str) -> dict | None:
+    """Map one Overpass element to ``{id, name, lat, lon, type, extra}``.
+
+    ``out center`` gives nodes their own ``lat``/``lon`` and ways/relations a
+    ``center``. Elements with no usable coordinate are dropped.
+    """
+    etype = elem.get("type")
+    eid = elem.get("id")
+    if eid is None:
+        return None
+    if etype == "node":
+        lat, lon = elem.get("lat"), elem.get("lon")
+    else:
+        center = elem.get("center") or {}
+        lat, lon = center.get("lat"), center.get("lon")
+    if lat is None or lon is None:
+        return None
+    tags = elem.get("tags") or {}
+    name = tags.get("name") or tags.get("ref") or f"{etype}/{eid}"
+    extra = {k: v for k, v in tags.items() if k != "name"}
+    return {
+        "id": f"{etype}/{eid}",
+        "name": name,
+        "lat": lat,
+        "lon": lon,
+        "type": type_,
+        "extra": extra,
+    }
+
+
+def overpass_nuclear_to_markers(data: dict) -> list[dict]:
+    """Convert an Overpass JSON response to normalized nuclear markers."""
+    markers = []
+    for elem in data.get("elements") or []:
+        marker = normalize_infra_element(elem, "nuclear")
+        if marker is not None:
+            markers.append(marker)
+    return markers
+
+
+async def fetch_infrastructure(types: str, bbox: str) -> list[dict]:
+    """Fetch Overpass infrastructure markers, cached 24h per quantized bbox.
+
+    ``types`` is a single supported enum value (``nuclear`` for now). ``bbox``
+    is ``minlon,minlat,maxlon,maxlat``.
+    """
+    requested = [t.strip() for t in types.split(",") if t.strip()]
+    minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+    key = f"infra:{','.join(requested)}:{bbox_cell_key(bbox)}"
+
+    async def _load() -> list[dict]:
+        # One query per requested type, concatenated. Nuclear is the only type
+        # today; the loop keeps the shape ready for military/hospital.
+        out: list[dict] = []
+        for type_ in requested:
+            query = infra_query(type_, minlon, minlat, maxlon, maxlat)
+            if _http is None:
+                async with httpx.AsyncClient(
+                    timeout=OVERPASS_TIMEOUT, follow_redirects=True,
+                    headers=_headers(),
+                ) as client:
+                    resp = await client.post(OVERPASS_INTERPRETER, data={"data": query})
+                    resp.raise_for_status()
+                    out.extend(overpass_nuclear_to_markers(resp.json()))
+            else:
+                resp = await _http.post(
+                    OVERPASS_INTERPRETER, data={"data": query},
+                    timeout=OVERPASS_TIMEOUT,
+                )
+                resp.raise_for_status()
+                out.extend(overpass_nuclear_to_markers(resp.json()))
+        return out
+
+    return await _ttl_get(key, float(INFRA_TTL), _load)

app/main.py

@@ -15,16 +15,18 @@ import asyncio
 import json
 import logging
 import re
+import time
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal
 from pathlib import Path
 from typing import NoReturn
+from urllib.parse import urlparse
 from uuid import UUID
 
 import httpx
 import structlog
-from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
+from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
@@ -38,10 +40,11 @@ from models import (
 )
 from schemas import (
     AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate,
+    ConflictZoneOut, ConflictsOut,
     DashboardSummary, EntityCreate, EntityKind, EntityOut,
     EventCreate, EventOut, FireOut, NewsArticleOut, NewsMapItemOut,
     NewsSummaryOut, NewsTickerItemOut,
-    FeedSourceCreate, FeedSourceOut,
+    FeedSourceCreate, FeedSourceOut, FeedSourceUpdate,
     KeyOut, KeyValueIn,
     NewsModelsOut, SettingsIn, SettingsOut,
     SearchResult, SentimentSummary, SourceType,
@@ -49,7 +52,8 @@ from schemas import (
     GeofenceCreate, GeofenceUpdate,
 )
 from ingestor import ingest_event, fetch_and_process
-from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
+from camera_scraper import is_public_url
+from sources import GDELT_API, ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
 from fire_sources import ingest_fires
 from keystore import KeyFormatError, delete_key, list_keys, set_key
 from settings_store import SettingsError, get_app_settings, list_models, set_summary_model
@@ -57,8 +61,10 @@ from live_layers import (
     fetch_aircraft, fetch_fire_incidents, fetch_fire_perimeters,
     fetch_gpsjam, fetch_planespotters_photo, fetch_radar_meta, fetch_sentinel1,
     fetch_storms, fetch_trains, fetch_vessels, fetch_weather_alerts,
-    overlay_catalog, parse_bbox, UpstreamRateLimited,
+    fetch_infrastructure, overlay_catalog, parse_bbox, UpstreamRateLimited,
 )
+from satellites import fetch_satellites, parse_groups, DEFAULT_GROUPS
+from place import reverse_geocode
 
 logging.basicConfig(level=logging.INFO)
 logger = structlog.get_logger("osint.dashboard")
@@ -74,6 +80,11 @@ async def _lifespan(app: FastAPI):
         await refresh_cache()
     except Exception:
         pass
+    try:
+        from vesselapi import hydrate_last_known
+        await hydrate_last_known()
+    except Exception:
+        pass
     from config import AISSTREAM_IN_APP, VESSELAPI_IN_APP
     ais_task = None
     vesselapi_task = None
@@ -267,6 +278,67 @@ def overlay_json(data, max_age: int) -> JSONResponse:
     return resp
 
 
+# ── HUD counters ─────────────────────────────────────────────────────────
+
+# Cheap ~100 B–2 KB counts for the layer rail. Cached in-process so the HUD
+# can poll every second without re-hitting SQL or upstream feeds.
+_STATS_TTL = 20.0
+_stats_cache: dict[str, tuple[float, dict]] = {}
+
+
+async def _stats_counts() -> dict:
+    """Fan out to in-memory last-known / cheap SQL counts. Never raises."""
+    from live_layers import (
+        aircraft_last_known, vessel_last_known, train_count, nws_alert_count,
+    )
+
+    counts: dict[str, int | str] = {
+        "aircraft": len(aircraft_last_known),
+        "vessels": len(vessel_last_known),
+        "trains": train_count,
+        "cameras": 0,
+        "fires": 0,
+        "quakes": 0,
+        "alerts": nws_alert_count,
+    }
+
+    # SQL counts are best-effort: a down DB or missing table must not 500 the
+    # rail — the frontend still renders with zeros.
+    try:
+        from camera_models import cameras as cam_table
+        async with async_session() as session:
+            counts["cameras"] = int(
+                (await session.execute(select(func.count()).select_from(cam_table))).scalar() or 0
+            )
+            counts["fires"] = int(
+                (await session.execute(select(func.count()).select_from(fires))).scalar() or 0
+            )
+            counts["quakes"] = int(
+                (await session.execute(
+                    select(func.count()).select_from(events).where(
+                        events.c.source_type == "earthquake"
+                    )
+                )).scalar() or 0
+            )
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("stats_db_failed", error=str(exc))
+
+    counts["timestamp"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+    return counts
+
+
+@app.get("/api/stats")
+async def api_stats():
+    """Cheap HUD counters (counts only — no GeoJSON). Cached ~20 s."""
+    now = time.monotonic()
+    cached = _stats_cache.get("stats")
+    if cached and now - cached[0] < _STATS_TTL:
+        return cached[1]
+    payload = await _stats_counts()
+    _stats_cache["stats"] = (now, payload)
+    return overlay_json(payload, 15)
+
+
 # ── Feed Sources ──────────────────────────────────────────────────────────
 
 @app.get("/api/sources", response_model=list[FeedSourceOut])
@@ -296,20 +368,22 @@ async def create_source(payload: FeedSourceCreate):
 
 
 @app.patch("/api/sources/{source_id}")
-async def update_source(source_id: UUID, payload: dict):
+async def update_source(source_id: UUID, payload: FeedSourceUpdate):
-    """Update a feed source (e.g., toggle enabled)."""
+    """Update a feed source (name/url/config/enabled only)."""
+    values = payload.model_dump(exclude_unset=True)
     async with async_session() as session:
         row = (await session.execute(
             select(feed_sources).where(feed_sources.c.id == source_id)
         )).mappings().one_or_none()
         if not row:
             raise HTTPException(404, "Source not found")
-        await session.execute(
+        if values:
-            feed_sources.update()
+            await session.execute(
-            .where(feed_sources.c.id == source_id)
+                feed_sources.update()
-            .values(**payload)
+                .where(feed_sources.c.id == source_id)
-        )
+                .values(**values)
-        await session.commit()
+            )
+            await session.commit()
     return {"ok": True}
 
 
@@ -752,9 +826,15 @@ async def put_settings(payload: SettingsIn):
 
 # ── Ingestion Triggers ───────────────────────────────────────────────────
 
+def _require_public_url(url: str, field: str) -> None:
+    if not is_public_url(url):
+        raise HTTPException(400, f"{field} is not a public URL")
+
+
 @app.post("/api/ingest/rss")
 async def trigger_rss_ingest(feed_url: str, source_id: str | None = None):
     """Trigger RSS feed ingestion."""
+    _require_public_url(feed_url, "feed_url")
     count = await ingest_rss_feed(feed_url, source_id)
     return {"status": "ok", "items_ingested": count}
 
@@ -762,6 +842,10 @@ async def trigger_rss_ingest(feed_url: str, source_id: str | None = None):
 @app.post("/api/ingest/gdelt")
 async def trigger_gdelt_ingest(query: str = "", max_articles: int = 50):
     """Trigger GDELT data ingestion."""
+    _require_public_url(GDELT_API, "GDELT target")
+    parsed = urlparse(query)
+    if parsed.scheme in ("http", "https") and parsed.hostname:
+        _require_public_url(query, "query")
     count = await ingest_gdelt(query, max_articles)
     return {"status": "ok", "articles_ingested": count}
 
@@ -787,24 +871,16 @@ async def trigger_social_ingest(query: str = "", max_items: int = 50):
     return {"status": "ok", "signals_ingested": count}
 
 
-@app.post("/api/ingest/masscan")
-async def trigger_masscan(background_tasks: BackgroundTasks):
-    """Queue one masscan pass at ≤200 pps. Does not block the request on the scan."""
-    from bg_jobs import MASSCAN_PPS_CAP, schedule_masscan_pass
-
-    async def _kick() -> None:
-        schedule_masscan_pass()
-
-    background_tasks.add_task(_kick)
-    return JSONResponse(
-        {"status": "accepted", "rate_pps": MASSCAN_PPS_CAP},
-        status_code=202,
-    )
-
-
 @app.websocket("/ws/live")
 async def live_ws(ws: WebSocket):
-    """Viewport-filtered AIS/ADS-B fan-out. Client sends {type:viewport,bbox}."""
+    """Viewport-filtered AIS/ADS-B fan-out.
+
+    Client JSON:
+      {"type":"viewport","bbox":"minlon,minlat,maxlon,maxlat"}
+      {"type":"watch_geofences","ids":["<uuid>", ...]}  — empty list = none
+    geofence_alert delivers if the point is in-viewport OR geofence_id is watched.
+    AIS/ADS-B/fire_aircraft stay viewport-only.
+    """
     from ws_manager import manager
 
     client_id = str(id(ws))
@@ -830,6 +906,10 @@ async def live_ws(ws: WebSocket):
                     manager.set_viewport(client_id, parse_bbox(str(data["bbox"])))
                 except ValueError:
                     continue
+            elif data.get("type") == "watch_geofences":
+                ids = data.get("ids") or []
+                if isinstance(ids, list):
+                    manager.set_watched_geofences(client_id, [str(x) for x in ids])
     except WebSocketDisconnect:
         pass
     finally:
@@ -981,7 +1061,7 @@ async def list_cameras(
         True,
         description="Only cameras with a verified HTTP/MJPEG snapshot_url "
                     "(the ones that actually preview). Set false to include "
-                    "unverified masscan port-554 hits.",
+                    "rows without a snapshot_url.",
     ),
     limit: int = Query(500, ge=1, le=5000),
 ):
@@ -1058,7 +1138,7 @@ async def get_camera(camera_id: UUID):
 async def camera_snapshot(camera_id: UUID):
     """Still image for one camera.
 
-    HTTP cameras go through the TTL cache. masscan/RTSP finds have no HTTP
+    HTTP cameras go through the TTL cache. RTSP finds have no HTTP
     snapshot_url — we probe common still-image paths and, failing that, grab
     one JPEG frame from RTSP via ffmpeg. No credentials are tried.
     """
@@ -1263,6 +1343,7 @@ async def list_news_summaries(
 
 
 _FLAGGED = ("critical", "high")
+_LESSER = ("medium", "low")
 
 
 @app.get("/api/news/ticker", response_model=list[NewsTickerItemOut])
@@ -1273,7 +1354,7 @@ async def list_news_ticker(
     ),
     limit: int = Query(20, ge=1, le=50),
 ):
-    """Flagged ticker rows (critical/high), newest first. No LLM required."""
+    """Ticker rows: critical/high first; medium/low if nothing is flagged."""
     async with async_session() as session:
         stmt = (
             select(news_items)
@@ -1287,6 +1368,19 @@ async def list_news_ticker(
             stmt = stmt.where(news_items.c.created_at >= since)
         stmt = stmt.limit(limit)
         rows = (await session.execute(stmt)).mappings().all()
+        if not rows:
+            stmt = (
+                select(news_items)
+                .where(
+                    news_items.c.kind == "ticker",
+                    news_items.c.importance.in_(_LESSER),
+                )
+                .order_by(news_items.c.created_at.desc())
+            )
+            if since:
+                stmt = stmt.where(news_items.c.created_at >= since)
+            stmt = stmt.limit(limit)
+            rows = (await session.execute(stmt)).mappings().all()
     return [
         NewsTickerItemOut(
             id=r["id"], headline=r["headline"], importance=r["importance"],
@@ -1413,6 +1507,58 @@ async def map_chokepoints():
     return {"chokepoints": chokepoints()}
 
 
+async def _fetch_geocoded_points() -> list[tuple[float, float, datetime | None]]:
+    """Collect geocoded ``(lat, lon, ts)`` rows from the local DB.
+
+    Sources are the flagged map pins (``news_items`` kind=map) and geocoded
+    news/GDELT events (``events`` with ``location_lat/lon``). This is the
+    pre-existing geocoded corpus the conflict-zone counters roll up — no
+    upstream scraping and no generated/jittered coordinates.
+    """
+    async with async_session() as session:
+        map_rows = (
+            await session.execute(
+                select(news_items.c.lat, news_items.c.lon, news_items.c.created_at)
+                .where(
+                    news_items.c.kind == "map",
+                    news_items.c.lat.isnot(None),
+                    news_items.c.lon.isnot(None),
+                )
+            )
+        ).all()
+        event_rows = (
+            await session.execute(
+                select(events.c.location_lat, events.c.location_lon, events.c.source_timestamp)
+                .where(
+                    events.c.source_type.in_(["rss", "gdel-t2"]),
+                    events.c.location_lat.isnot(None),
+                    events.c.location_lon.isnot(None),
+                )
+            )
+        ).all()
+    return [tuple(r) for r in map_rows] + [tuple(r) for r in event_rows]
+
+
+@app.get("/api/conflicts", response_model=ConflictsOut)
+async def list_conflicts():
+    """Curated conflict-zone catalog with per-zone event counts.
+
+    Static catalogue (severity + short factual description) merged with a live
+    ``eventCount`` roll-up of pre-existing geocoded news/GDELT//api/news/map
+    rows inside each zone bbox. Empty DB → ``eventCount=0`` (never 502).
+    """
+    from conflicts import conflict_zones, zone_event_stats
+
+    points = await _fetch_geocoded_points()
+    timestamp = datetime.now(timezone.utc)
+    zones = []
+    for z in conflict_zones():
+        bbox = z.pop("bbox")
+        count, latest = zone_event_stats(points, bbox)
+        zones.append({**z, "eventCount": count, "lastUpdated": latest})
+    return {"zones": zones, "timestamp": timestamp}
+
+
 def _upstream_or_502(exc: Exception, name: str) -> NoReturn:
     logger.warning("live_layer_upstream_failed", layer=name, error=str(exc))
     raise HTTPException(502, f"{name} upstream unavailable: {exc}") from exc
@@ -1530,7 +1676,17 @@ async def list_vessels(
         from tracks import fetch_positions_at, parse_timestamp
         ts = parse_timestamp(timestamp)
         if ts is not None:
-            return overlay_json(await fetch_positions_at("vessel", ts, bbox, limit), 5)
+            from vesselapi import fetch_vessels_as_of
+            if src == "vesselapi":
+                return overlay_json(await fetch_vessels_as_of(ts, bbox, limit), 5)
+            ais = await fetch_positions_at("vessel", ts, bbox, limit)
+            if src == "aisstream":
+                return overlay_json(ais, 5)
+            va = await fetch_vessels_as_of(ts, bbox, limit)
+            by_id = {m["id"]: m for m in ais}
+            for m in va:
+                by_id[m["id"]] = m
+            return overlay_json(list(by_id.values())[:limit], 5)
         return overlay_json(await fetch_vessels(bbox, limit, src=src), 5)
     except ValueError as exc:
         raise HTTPException(422, str(exc)) from exc
@@ -1612,33 +1768,68 @@ async def api_update_geofence(gid: str, payload: GeofenceUpdate):
 @app.delete("/api/geofences/{gid}", status_code=204)
 async def api_delete_geofence(gid: str):
     from geofence import delete_geofence
-    await delete_geofence(gid)
+    ok = await delete_geofence(gid)
+    if not ok:
+        raise HTTPException(404, "geofence not found")
     return None
 
 
-@app.get("/api/geofence-alerts")
+@app.get("/api/geofences/{gid}/at")
-async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)):
+async def api_geofence_at(
-    from sqlalchemy import text as sql_text
+    gid: str,
+    timestamp: str = Query(..., description="ISO-8601 instant for the 1-minute DVR bucket"),
+):
+    """Aircraft/vessels/fires inside this fence at time T. Never writes."""
+    from geofence import snapshot_at
+    from tracks import parse_timestamp
+
     try:
-        async with async_session() as session:
+        ts = parse_timestamp(timestamp)
-            rows = (await session.execute(sql_text(
+    except ValueError as exc:
-                """
+        raise HTTPException(422, str(exc)) from exc
-                SELECT id::text, geofence_id::text, source_kind, entity_id,
+    if ts is None:
-                       lat, lon, payload, created_at
+        raise HTTPException(422, "timestamp required")
-                FROM geofence_alerts
+    try:
-                ORDER BY created_at DESC
+        body = await snapshot_at(gid, ts)
-                LIMIT :limit
-                """
-            ), {"limit": limit})).mappings().all()
-        out = []
-        for r in rows:
-            item = dict(r)
-            if item.get("created_at") is not None:
-                item["created_at"] = item["created_at"].isoformat()
-            out.append(item)
-        return out
     except Exception:
-        return []
+        body = {
+            "geofence_id": gid,
+            "timestamp": ts.isoformat(),
+            "aircraft": [],
+            "vessels": [],
+            "fires": [],
+        }
+    if body is None:
+        raise HTTPException(404, "geofence not found")
+    return body
+
+
+@app.get("/api/geofence-alerts")
+async def api_geofence_alerts(
+    geofence_id: UUID | None = Query(None),
+    since: str | None = Query(None, description="ISO-8601 inclusive lower bound"),
+    until: str | None = Query(None, description="ISO-8601 inclusive upper bound"),
+    source_kind: str | None = Query(None, description="firms|ais|adsb"),
+    limit: int = Query(100, ge=1, le=500),
+):
+    """Hit log for drawn fences. Not /api/alerts (entity/keyword)."""
+    from geofence import list_alerts
+    from tracks import parse_timestamp
+
+    if source_kind is not None and source_kind not in ("firms", "ais", "adsb"):
+        raise HTTPException(422, "source_kind must be one of: firms, ais, adsb")
+    try:
+        since_ts = parse_timestamp(since) if since else None
+        until_ts = parse_timestamp(until) if until else None
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    return await list_alerts(
+        geofence_id=str(geofence_id) if geofence_id else None,
+        since=since_ts,
+        until=until_ts,
+        source_kind=source_kind,
+        limit=limit,
+    )
 
 
 @app.get("/api/fire-aircraft")
@@ -1700,6 +1891,25 @@ async def list_storms():
         _upstream_or_502(exc, "storms")
 
 
+@app.get("/api/place")
+async def get_place(
+    lat: float = Query(..., ge=-90, le=90),
+    lon: float = Query(..., ge=-180, le=180),
+):
+    """Nominatim reverse geocode for the map \"What's here?\" dossier.
+
+    Identifying ``OSINT_USER_AGENT``, 1 req/s, 60s cache, 500 keys. The HUD
+    lists already-loaded overlay entities client-side — this route does not
+    refetch aircraft/vessels/cameras/fires.
+    """
+    try:
+        return overlay_json(await reverse_geocode(lat, lon), 60)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    except Exception as exc:
+        _upstream_or_502(exc, "nominatim")
+
+
 _GPSJAM_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 
 
@@ -1735,6 +1945,68 @@ async def map_gpsjam(
     return overlay_json(fc, 3600)
 
 
+@app.get("/api/satellites")
+async def list_satellites(
+    groups: str | None = Query(None, description="Comma-separated CelesTrak groups"),
+    bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
+    limit: int = Query(2000, ge=1, le=5000),
+):
+    """Last-known satellite positions from CelesTrak GP JSON, SGP4-propagated.
+
+    Default groups are ``stations,weather`` (tens of objects). The GP element
+    blob is fetched at most once per 2 hours per group and cached; positions
+    are re-propagated on every request. Falls back to the last good blob on a
+    CelesTrak 403 / stale response, and to SatNOGS TLE only when the cache is
+    empty. Unknown groups 400.
+    """
+    try:
+        group_list = parse_groups(groups if groups is not None else ",".join(DEFAULT_GROUPS))
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+    if bbox:
+        _parse_bbox_query(bbox)
+    try:
+        payload = await fetch_satellites(group_list, bbox=bbox, limit=limit)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    except Exception as exc:
+        _upstream_or_502(exc, "satellites")
+    return overlay_json(payload, 30)
+
+
+_INFRA_TYPES = frozenset({"nuclear"})
+
+
+@app.get("/api/infrastructure")
+async def api_infrastructure(
+    types: str = Query(..., description="comma-separated enum (nuclear)"),
+    bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
+):
+    """Overpass-derived static infrastructure markers (nuclear power plants).
+
+    ``bbox`` is required; ``types`` is a comma-separated subset of ``nuclear``.
+    Fetched from Overpass (identifying UA, 25s query) and cached 24h per
+    quantized bbox. Markers are ``{id, name, lat, lon, type, extra}``.
+    """
+    if not bbox:
+        raise HTTPException(400, "bbox required (minlon,minlat,maxlon,maxlat)")
+    requested = [t.strip() for t in (types or "").split(",") if t.strip()]
+    if not requested:
+        raise HTTPException(422, "types required (e.g. nuclear)")
+    unknown = [t for t in requested if t not in _INFRA_TYPES]
+    if unknown:
+        raise HTTPException(
+            422, f"unsupported types: {', '.join(unknown)} (supported: nuclear)"
+        )
+    try:
+        markers = await fetch_infrastructure(",".join(requested), bbox)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    except Exception as exc:
+        _upstream_or_502(exc, "infrastructure")
+    return overlay_json(markers, 86400)
+
+
 @app.get("/api/map/times")
 async def map_layer_times(
     layer: str = Query(..., description="GIBS layer identifier, e.g. VIIRS_SNPP_CorrectedReflectance_TrueColor"),
@@ -1765,4 +2037,4 @@ app.mount("/static", CachedStaticFiles(directory=str(STATIC_DIR)), name="static"
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000, workers=1)  # single worker: in-memory WS/pubsub + layer caches

app/masscan_config.py

@@ -1,65 +0,0 @@
-"""Active camera-discovery configuration (masscan-based, env-driven).
-
-All knobs read from the environment with safe defaults. The scanner targets
-open TCP port 554 (RTSP — the typical IP-camera port) across a configured
-range and feeds results into the same `cameras` table as the passive scraper
-(discovery_source='masscan'), deduped by URL hash.
-
-ETHICS / SCOPE (mirrors camera_scraper.py):
-  * Detection only — a SYN port scan for OPEN hosts. No credential guessing,
-    no login attempts, no banner grabbing, and no access to camera feeds.
-  * Private / reserved ranges are excluded via MASSCAN_EXCLUDEFILE so the
-    scanner never probes RFC1918, loopback, link-local, multicast, or the
-    bogons. Fail closed if the excludefile is missing.
-
-TIMING REALITY: at the residential-safe default of 200 pps a full IPv4
-sweep (0.0.0.0/0, ~4.29B addresses) takes ~8 months. This is therefore a
-CONTINUOUS ROLLING SWEEP, not a "finish in a day" job: masscan streams
-open hosts to stdout and the runner ingests them incrementally, then
-restarts the sweep when a pass completes. New cameras are detected as they
-appear on each pass. 1k/10k pps saturated a home uplink — do not raise the
-rate unless you are on a VPS / unmetered link.
-"""
-
-from __future__ import annotations
-
-import os
-
-# Path to the masscan binary (installed on the Pi host).
-MASSCAN_BIN = os.getenv("MASSCAN_BIN", "masscan")
-
-# CIDR(s) to sweep. Default = the whole public IPv4 space.
-MASSCAN_RANGE = os.getenv("MASSCAN_RANGE", "0.0.0.0/0")
-
-# Port(s) to probe. Default 554 = RTSP, the typical IP-camera port.
-MASSCAN_PORTS = os.getenv("MASSCAN_PORTS", "554")
-
-# Packets/sec. 200 is the residential-safe default — 1k/10k pps saturated
-# a home uplink. Raise only on a VPS / unmetered link.
-MASSCAN_RATE = int(os.getenv("MASSCAN_RATE", "200"))
-
-# Retransmission count. 1 maximizes unique-host coverage at low rate; the
-# default (10) spends most of the budget re-probing the same hosts.
-MASSCAN_RETRIES = int(os.getenv("MASSCAN_RETRIES", "1"))
-
-# Seconds to keep listening for straggler responses after the last probe.
-# 0 avoids a 10s tail per pass; tiny loss of the very last hosts is fine
-# since the sweep repeats.
-MASSCAN_WAIT = int(os.getenv("MASSCAN_WAIT", "0"))
-
-# Excludefile path on the Pi host. Must contain RFC1918/loopback/link-local/
-# multicast/bogons so the scanner never probes private ranges. Fail closed if
-# the file is absent (the runner refuses to start rather than scan wide).
-MASSCAN_EXCLUDEFILE = os.getenv(
-    "MASSCAN_EXCLUDEFILE", "/etc/osint-dashboard/masscan-excludes.txt"
-)
-
-# Ingest batch size — flush this many newly-seen hosts to the DB per round.
-MASSCAN_FLUSH_EVERY = int(os.getenv("MASSCAN_FLUSH_EVERY", "250"))
-
-# NATS subject newly-found cameras are published on (same feed as the
-# passive scraper so the shared ingester persists them).
-MASSCAN_NATS_SUBJECT = os.getenv("MASSCAN_NATS_SUBJECT", "events.camera")
-
-# discovery_source tag written into the cameras table.
-MASSCAN_DISCOVERY_SOURCE = os.getenv("MASSCAN_DISCOVERY_SOURCE", "masscan")

app/masscan_scanner.py

@@ -1,226 +0,0 @@
-"""masscan result parsing + ingestion for the OSINT dashboard.
-
-Turns a stream of masscan JSON-lines (open port 554 hosts) into rows in the
-`cameras` table with discovery_source='masscan', deduped by URL hash against
-whatever the passive scraper already found. Newly discovered hosts are also
-published to NATS (`events.camera`) so the shared ingester pipeline persists
-them exactly like scraper finds.
-
-Scope: detection of OPEN hosts only. No credentials, no banners, no feed
-access. Private/reserved ranges never enter masscan (see excludefile).
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-import logging
-from datetime import datetime, timezone
-
-from camera_models import cameras
-from camera_scraper import url_hash, geolocate_ips
-from database import async_session
-
-from masscan_config import (
-    MASSCAN_NATS_SUBJECT, MASSCAN_DISCOVERY_SOURCE,
-)
-
-logger = logging.getLogger("osint.masscan_scanner")
-
-
-# ── URL building ──────────────────────────────────────────────────────────
-
-def build_rtsp_url(ip: str) -> str:
-    """Canonical URL for an open-RTSP host. Used as the dedupe key."""
-    return f"rtsp://{ip}/"
-
-
-# ── masscan JSON parsing ──────────────────────────────────────────────────
-# masscan --output-format=json --output-file=- emits line-delimited JSON on a
-# pipe (a bare object per open host), not the array form used for seekable
-# files. We parse per-line and tolerate an accidental leading '['.
-
-def parse_masscan_line(line: str) -> list[dict]:
-    """Parse one masscan stdout line into a list of host records.
-
-    A line may contain one JSON object or, defensively, be wrapped in an
-    array. Returns [] on anything unparseable (harmless — the sweep repeats).
-    """
-    s = line.strip()
-    if not s:
-        return []
-    s = s.lstrip("[").rstrip("]").strip()
-    if not s:
-        return []
-    # Multiple records may share a line separated by '},{'.
-    if s.endswith(","):
-        s = s[:-1].rstrip()
-    out: list[dict] = []
-    for cand in _split_records(s):
-        try:
-            obj = json.loads(cand)
-        except (json.JSONDecodeError, ValueError):
-            continue
-        if isinstance(obj, dict) and obj.get("ip"):
-            out.append(obj)
-    return out
-
-
-def _split_records(s: str) -> list[str]:
-    """Split a buffer into individual JSON object strings, honoring nesting."""
-    records, depth, start = [], 0, 0
-    for i, ch in enumerate(s):
-        if ch == "{":
-            if depth == 0:
-                start = i
-            depth += 1
-        elif ch == "}":
-            depth -= 1
-            if depth == 0:
-                records.append(s[start:i + 1])
-    return records
-
-
-def extract_open_ips(records: list[dict], port: int) -> list[str]:
-    """Return the list of IPs from records that have `port` open."""
-    ips: list[str] = []
-    for rec in records:
-        for p in rec.get("ports", []):
-            if p.get("port") == port and p.get("status") == "open":
-                ips.append(rec["ip"])
-                break
-    return ips
-
-
-# ── Persistence ───────────────────────────────────────────────────────────
-
-async def ingest_open_hosts(ips: list[str]) -> tuple[int, list[str]]:
-    """Insert-or-refresh camera rows for open RTSP hosts that have a public feed.
-
-    A host only lands in the table (and therefore on the map) if an
-    unauthenticated HTTP still or MJPEG URL responds. Port-554-only hosts
-    are skipped. Returns (newly_inserted, hosts_with_working_feed).
-    """
-    if not ips:
-        return 0, []
-    from camera_preview import probe_public_feed
-
-    unique = list(dict.fromkeys(ips))
-    sem = asyncio.Semaphore(20)
-
-    async def _probe(ip: str) -> tuple[str, str | None]:
-        async with sem:
-            return ip, await probe_public_feed(ip)
-
-    probed = await asyncio.gather(*(_probe(ip) for ip in unique))
-    live = [(ip, feed) for ip, feed in probed if feed]
-    if not live:
-        logger.info("masscan ingest: 0 working feeds of %d open-554 hosts",
-                    len(unique))
-        return 0, []
-
-    now = datetime.now(timezone.utc)
-    coords = await geolocate_ips([ip for ip, _ in live])
-    new = 0
-    async with async_session() as session:
-        for ip, feed in live:
-            url = build_rtsp_url(ip)
-            h = url_hash(url)
-            lat, lon = coords.get(ip, (None, None))
-            existing = (await session.execute(
-                cameras.select().where(cameras.c.url_hash == h)
-            )).one_or_none()
-            if existing is None:
-                await session.execute(cameras.insert().values(
-                    url_hash=h,
-                    source_url=url,
-                    snapshot_url=feed,
-                    discovery_source=MASSCAN_DISCOVERY_SOURCE,
-                    location_lat=lat,
-                    location_lon=lon,
-                    location_name=f"{ip} (IP-geo)" if lat is not None else None,
-                    vendor=None,
-                    device_type="rtsp",
-                    first_seen=now,
-                    last_seen=now,
-                    raw={"discovered_via": "masscan", "port": 554,
-                         "public_feed": feed},
-                ))
-                new += 1
-            else:
-                await session.execute(cameras.update().where(
-                    cameras.c.url_hash == h
-                ).values(
-                    last_seen=now,
-                    snapshot_url=feed,
-                    location_lat=lat,
-                    location_lon=lon,
-                    location_name=f"{ip} (IP-geo)" if lat is not None else None,
-                ))
-        await session.commit()
-    logger.info("masscan ingest: %d new working feeds (%d probed, %d open-554)",
-                new, len(live), len(unique))
-    return new, [ip for ip, _ in live]
-
-
-# ── NATS publish ──────────────────────────────────────────────────────────
-
-async def publish_new_hosts(ips: list[str]) -> int:
-    """Publish newly-found open hosts to NATS for the shared ingester.
-
-    Returns the number of messages published (0 if NATS is down).
-    """
-    import json as _json
-    import nats
-    from config import NATS_URL
-
-    if not ips:
-        return 0
-    try:
-        nc = await nats.connect(NATS_URL)
-    except Exception:  # noqa: BLE001
-        logger.warning("NATS unavailable — skipping publish pass")
-        return 0
-    published = 0
-    try:
-        js = nc.jetstream()
-        for ip in dict.fromkeys(ips):
-            url = build_rtsp_url(ip)
-            msg = {
-                "source_type": "camera",
-                "title": f"Open RTSP camera ({ip})",
-                "url": url,
-                "location_lat": None,
-                "location_lon": None,
-                "location_name": None,
-                "tags": ["osint", "camera", MASSCAN_DISCOVERY_SOURCE],
-                "raw": {
-                    "url_hash": url_hash(url),
-                    "source_url": url,
-                    "snapshot_url": None,
-                    "vendor": None,
-                    "device_type": "rtsp",
-                    "discovered_via": "masscan",
-                    "port": 554,
-                },
-                "source_timestamp": datetime.now(timezone.utc).isoformat(),
-            }
-            await js.publish(MASSCAN_NATS_SUBJECT, _json.dumps(msg).encode())
-            published += 1
-    finally:
-        await nc.close()
-    logger.info("published %d masscan finds to %s", published, MASSCAN_NATS_SUBJECT)
-    return published
-
-
-# ── Batch drain helper used by the runner ─────────────────────────────────
-
-async def flush(seen: set[str], new_accum: int) -> tuple[int, int]:
-    """Ingest + publish the accumulated host set; return (new, published)."""
-    if not seen:
-        return 0, 0
-    ips = list(seen)
-    new, live = await ingest_open_hosts(ips)
-    published = await publish_new_hosts(live)
-    seen.clear()
-    return new, published

app/place.py

@@ -0,0 +1,99 @@
+"""Nominatim reverse-geocode proxy for the map place dossier.
+
+Browser clients cannot set an identifying User-Agent, and Nominatim typically
+blocks CORS — so the HUD calls GET /api/place instead of talking to OSM
+directly. Cache 60s / 500 keys; never exceed 1 req/s upstream.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+import httpx
+from cachetools import TTLCache
+
+from config import NOMINATIM_MIN_INTERVAL, NOMINATIM_URL, OSINT_USER_AGENT
+
+_NOMINATIM = NOMINATIM_URL.rstrip("/")
+
+place_cache: TTLCache = TTLCache(maxsize=500, ttl=60)
+
+_lock = asyncio.Lock()
+_last_req = 0.0
+
+_ADDR_KEEP = (
+    "house_number", "road", "neighbourhood", "suburb", "city", "town",
+    "village", "hamlet", "county", "state", "postcode", "country", "country_code",
+)
+
+
+def cache_key(lat: float, lon: float) -> str:
+    return f"{lat:.4f},{lon:.4f}"
+
+
+def slim_place(lat: float, lon: float, data: dict | None) -> dict:
+    data = data or {}
+    raw_addr = data.get("address")
+    addr_in: dict = raw_addr if isinstance(raw_addr, dict) else {}
+    address = {k: addr_in[k] for k in _ADDR_KEEP if addr_in.get(k)}
+    err = data.get("error")
+    display = None if err else (data.get("display_name") or None)
+    name = None if err else (data.get("name") or address.get("city")
+                             or address.get("town") or address.get("village") or None)
+    return {
+        "lat": lat,
+        "lon": lon,
+        "display_name": display,
+        "name": name,
+        "address": address,
+        "osm_type": None if err else data.get("osm_type"),
+        "osm_id": None if err else data.get("osm_id"),
+        "attribution": "© OpenStreetMap contributors",
+    }
+
+
+async def reverse_geocode(lat: float, lon: float) -> dict:
+    """Reverse-geocode a point. Cache hits skip Nominatim entirely."""
+    if not (-90.0 <= lat <= 90.0 and -180.0 <= lon <= 180.0):
+        raise ValueError("lat/lon out of range")
+    key = cache_key(lat, lon)
+    qlat, qlon = (float(p) for p in key.split(","))
+    async with _lock:
+        hit = place_cache.get(key)
+        if hit is not None:
+            return hit
+        global _last_req
+        wait = _last_req + NOMINATIM_MIN_INTERVAL - time.monotonic()
+        if wait > 0:
+            await asyncio.sleep(wait)
+        body = await _fetch_nominatim(qlat, qlon)
+        _last_req = time.monotonic()
+        place_cache[key] = body
+        return body
+
+
+async def _fetch_nominatim(lat: float, lon: float) -> dict:
+    headers = {
+        "User-Agent": OSINT_USER_AGENT,
+        "Accept": "application/json",
+    }
+    url = f"{_NOMINATIM}/reverse"
+    params = {
+        "lat": f"{lat:.6f}",
+        "lon": f"{lon:.6f}",
+        "format": "jsonv2",
+        "addressdetails": "1",
+        "zoom": "18",
+    }
+    async with _http_client(timeout=10.0, follow_redirects=True) as client:
+        r = await client.get(url, params=params, headers=headers)
+        r.raise_for_status()
+        data = r.json()
+    if not isinstance(data, dict):
+        data = {}
+    return slim_place(lat, lon, data)
+
+
+def _http_client(**kwargs):
+    return httpx.AsyncClient(**kwargs)

app/requirements.txt

@@ -13,3 +13,4 @@ structlog>=24.4
 websockets>=14
 cachetools>=5.5
 h3>=4.0
+sgp4>=2.23

app/run_masscan_service.py

@@ -1,149 +0,0 @@
-"""Continuous masscan rolling-sweep service for the OSINT dashboard.
-
-Runs masscan against the configured range for open port 554 (RTSP), streams
-the JSON-lines output, and ingests open hosts into the `cameras` table (new
-finds only) plus publishes them to NATS — exactly like the passive scraper.
-
-Because a full IPv4 sweep at a conservative rate takes days, this runs
-masscan CONTINUOUSLY: each pass streams results in as they're found, and when
-a pass completes the sweep restarts from the top. New cameras are picked up
-on every pass.
-
-Ethics: detection-only (open-port SYN scan). Private/reserved ranges are
-excluded and the service REFUSES to start if the excludefile is missing, so
-we never probe private space by accident.
-
-Run once (for a manual/test pass):  python app/run_masscan_service.py --once
-Run forever (systemd):            python app/run_masscan_service.py
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import os
-import sys
-from pathlib import Path
-
-sys_path = str(Path(__file__).parent)
-sys.path.insert(0, sys_path)
-
-import masscan_config as cfg  # noqa: E402
-from database import init_extensions  # noqa: E402
-from masscan_scanner import (  # noqa: E402
-    parse_masscan_line, extract_open_ips, flush,
-)
-
-logging.basicConfig(level=logging.INFO,
-                    format="%(asctime)s %(levelname)s %(name)s: %(message)s")
-logger = logging.getLogger("osint.masscan_service")
-
-ONCE = "--once" in sys.argv[1:]
-
-
-def _verify_excludefile() -> None:
-    """Fail closed: refuse to sweep the wide range without an excludefile."""
-    if not cfg.MASSCAN_EXCLUDEFILE:
-        raise SystemExit("MASSCAN_EXCLUDEFILE is empty — refusing to run")
-    if not Path(cfg.MASSCAN_EXCLUDEFILE).is_file():
-        raise SystemExit(
-            f"excludefile {cfg.MASSCAN_EXCLUDEFILE!r} missing — refusing to "
-            f"run (would risk probing private ranges). Install the excludefile "
-            f"first (see deploy/masscan-excludes.txt)."
-        )
-
-
-def build_command() -> list[str]:
-    cmd = [
-        cfg.MASSCAN_BIN,
-        cfg.MASSCAN_RANGE,
-        f"-p{cfg.MASSCAN_PORTS}",
-        f"--rate={cfg.MASSCAN_RATE}",
-        f"--retries={cfg.MASSCAN_RETRIES}",
-        f"--wait={cfg.MASSCAN_WAIT}",
-        "--output-format=json",
-        "--output-file=-",
-    ]
-    if cfg.MASSCAN_EXCLUDEFILE:
-        cmd.append(f"--excludefile={cfg.MASSCAN_EXCLUDEFILE}")
-    return cmd
-
-
-async def _drain_stderr(stream: asyncio.StreamReader) -> None:
-    """Consume masscan's progress chatter so its stderr pipe never fills."""
-    while True:
-        line = await stream.readline()
-        if not line:
-            break
-        text = line.decode(errors="ignore").strip()
-        if text and not text.startswith("rate:"):
-            logger.debug("masscan: %s", text)
-
-
-async def run_pass() -> tuple[int, int]:
-    """Run one full sweep pass, ingesting incrementally.
-
-    Returns (new_hosts, total_hosts_seen) for the whole pass.
-    """
-    cmd = build_command()
-    logger.info("starting masscan pass: %s", " ".join(cmd))
-    proc = await asyncio.create_subprocess_exec(
-        *cmd,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    if proc.stderr is not None:
-        asyncio.ensure_future(_drain_stderr(proc.stderr))
-
-    seen: set[str] = set()
-    total_seen = 0
-    total_new = 0
-    try:
-        while True:
-            raw = await proc.stdout.readline()
-            if not raw:
-                break
-            records = parse_masscan_line(raw.decode(errors="ignore"))
-            for ip in extract_open_ips(records, 554):
-                if ip in seen:
-                    continue
-                seen.add(ip)
-                if len(seen) >= cfg.MASSCAN_FLUSH_EVERY:
-                    new, _published = await flush(seen, total_new)
-                    total_new += new
-                    total_seen += new
-        # Drain the final partial batch.
-        if seen:
-            new, _published = await flush(seen, total_new)
-            total_new += new
-        rc = await proc.wait()
-    except asyncio.CancelledError:
-        proc.kill()
-        raise
-    logger.info("masscan pass finished (rc=%s): %d new hosts ingested",
-                rc, total_new)
-    return total_new, total_seen
-
-
-async def main() -> None:
-    _verify_excludefile()
-    await init_extensions()
-    logger.info(
-        "masscan service starting: range=%s ports=%s rate=%s pps (full sweep "
-        "~%.0fh at this rate)",
-        cfg.MASSCAN_RANGE, cfg.MASSCAN_PORTS, cfg.MASSCAN_RATE,
-        4.29e9 / cfg.MASSCAN_RATE / 3600,
-    )
-    while True:
-        try:
-            await run_pass()
-        except Exception:  # noqa: BLE001
-            logger.exception("masscan pass error")
-        if ONCE:
-            return
-        # Small gap between passes so the restart is visible in logs.
-        await asyncio.sleep(5)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

app/satellites.py

@@ -0,0 +1,289 @@
+"""CelesTrak satellites last-known overlay.
+
+Fetches GP **JSON** (OMM mean elements — not TLE) per group at most once per
+2 hours, caches the element blob, and propagates positions with a real SGP4
+library on every request. Positions move every second; the *element set* is
+what we cache, not the derived lat/lon.
+
+Catalog numbers >= 100000 only fit OMM/JSON, never a 5-column TLE field, so
+elements are initialized through :func:`sgp4.omm.initialize` (which consumes
+the CelesTrak GP JSON fields verbatim) rather than round-tripping to TLE.
+
+CelesTrak usage policy is non-negotiable: fetch the GP JSON blob at most once
+per 2 hours per group, never fan out every GROUP, never also fetch
+``GROUP=active`` plus subsets, and identify with ``OSINT_USER_AGENT``.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from datetime import datetime, timezone
+from urllib.parse import quote
+
+logger = logging.getLogger("osint.satellites")
+
+CELESTRAK_GP = "https://celestrak.org/NORAD/elements/gp.php"
+SATNOGS_TLE = "https://db.satnogs.org/api/tle/"
+DEFAULT_GROUPS = ("stations", "weather")
+ALLOWED_GROUPS = ("stations", "weather", "gps-ops", "starlink")
+# CelesTrak policy: do not hit gp.php more than once per 2 hours per group.
+SATELLITE_TTL = 2 * 3600.0
+SOURCE_CELESTRAK = "celestrak"
+SOURCE_SATNOGS = "satnogs"
+DEFAULT_LIMIT = 2000
+
+# WGS-84 ellipsoid for TEME -> geodetic.
+_WGS84_A = 6378.137
+_WGS84_F = 1.0 / 298.257223563
+
+# Last-good element blob per group, kept past TTL so a 403 / "has not updated
+# since ..." still serves the previous set instead of failing the overlay.
+_last_good: dict[str, list[dict]] = {}
+
+
+def parse_groups(raw: str | None) -> list[str]:
+    """Validate + normalize a comma-separated group list. Raises ValueError.
+
+    Starlink is allowed only when explicitly requested (never in the default);
+    it is a large supplemental feed, not part of the stations/weather default.
+    """
+    groups = [g.strip().lower() for g in (raw or "").split(",") if g.strip()]
+    if not groups:
+        raise ValueError("groups must be a non-empty comma-separated list")
+    bad = [g for g in groups if g not in ALLOWED_GROUPS]
+    if bad:
+        raise ValueError(f"unknown group(s): {', '.join(bad)}")
+    # Dedup, preserve order.
+    seen: set[str] = set()
+    out: list[str] = []
+    for g in groups:
+        if g not in seen:
+            seen.add(g)
+            out.append(g)
+    return out
+
+
+def _teme_to_geodetic(
+    r: tuple[float, float, float],
+    jd: float,
+    fr: float,
+) -> tuple[float, float, float]:
+    """SGP4 TEME position (km) -> geodetic (lat_deg, lon_deg, alt_km).
+
+    Rotate TEME into an Earth-fixed frame via GMST, then iterate the WGS-84
+    geodetic conversion. Good to well under a km for a ground-track overlay.
+    """
+    # GMST (radians) from UT1 ~= UTC here (sub-second error is negligible).
+    d = (jd + fr) - 2451545.0
+    t = d / 36525.0
+    gmst_s = (
+        67310.54841
+        + (876600.0 * 3600.0 + 8640184.812866) * t
+        + 0.093104 * t * t
+        - 6.2e-6 * t * t * t
+    )
+    theta = math.radians((gmst_s % 86400.0) / 240.0)
+
+    x, y, z = r
+    xe = x * math.cos(theta) + y * math.sin(theta)
+    ye = -x * math.sin(theta) + y * math.cos(theta)
+    ze = z
+
+    e2 = _WGS84_F * (2.0 - _WGS84_F)
+    p = math.sqrt(xe * xe + ye * ye)
+    lon = math.atan2(ye, xe)
+    lat = math.atan2(ze, p * (1.0 - e2))
+    alt = 0.0
+    for _ in range(10):
+        n = _WGS84_A / math.sqrt(1.0 - e2 * math.sin(lat) ** 2)
+        alt = p / math.cos(lat) - n
+        lat = math.atan2(ze, p * (1.0 - e2 * n / (n + alt)))
+    n = _WGS84_A / math.sqrt(1.0 - e2 * math.sin(lat) ** 2)
+    alt = p / math.cos(lat) - n
+    return math.degrees(lat), math.degrees(lon), alt
+
+
+def propagate_gp(
+    elements: list[dict],
+    group: str,
+    now: datetime,
+) -> list[dict]:
+    """Propagate CelesTrak GP JSON elements to geodetic positions at ``now``.
+
+    Pure and deterministic given ``now``. Returns ``[{id, name, lat, lon,
+    alt_km, group}]``; malformed elements and propagation errors are skipped.
+    """
+    from sgp4.api import Satrec, jday
+    import sgp4.omm as omm
+
+    jd, fr = jday(
+        now.year, now.month, now.day,
+        now.hour, now.minute, now.second + now.microsecond / 1e6,
+    )
+    out: list[dict] = []
+    for rec in elements:
+        if not isinstance(rec, dict):
+            continue
+        sat = Satrec()
+        try:
+            omm.initialize(sat, rec)
+        except (KeyError, ValueError, TypeError):
+            continue
+        err, r, _v = sat.sgp4(jd, fr)
+        if err != 0:
+            continue
+        lat, lon, alt = _teme_to_geodetic(r, jd, fr)
+        norad = rec.get("NORAD_CAT_ID")
+        out.append({
+            "id": str(norad) if norad is not None else "",
+            "name": rec.get("OBJECT_NAME") or str(norad or ""),
+            "lat": round(lat, 5),
+            "lon": round(lon, 5),
+            "alt_km": round(alt, 2),
+            "group": group,
+        })
+    return out
+
+
+def _max_epoch(elements: list[dict]) -> str | None:
+    """Most recent EPOCH across an element set (ISO-8601 lexical max)."""
+    epochs = [
+        str(e["EPOCH"]) for e in elements
+        if isinstance(e, dict) and e.get("EPOCH")
+    ]
+    return max(epochs) if epochs else None
+
+
+def propagate_satnogs_tle(
+    payload: list[dict],
+    group: str,
+    now: datetime,
+) -> tuple[list[dict], str | None]:
+    """Fallback parser for SatNOGS TLE JSON (``[{tle0,tle1,tle2,updated}]``).
+
+    Returns ``(satellites, epoch)`` where epoch is the max ``updated`` time.
+    Only used when the CelesTrak cache is completely empty.
+    """
+    from sgp4.api import Satrec, jday
+
+    jd, fr = jday(
+        now.year, now.month, now.day,
+        now.hour, now.minute, now.second + now.microsecond / 1e6,
+    )
+    out: list[dict] = []
+    epochs: list[str] = []
+    for rec in payload or []:
+        if not isinstance(rec, dict):
+            continue
+        line1 = rec.get("tle1")
+        line2 = rec.get("tle2")
+        if not line1 or not line2:
+            continue
+        try:
+            sat = Satrec.twoline2rv(line1, line2)
+        except (ValueError, TypeError):
+            continue
+        e, r, _v = sat.sgp4(jd, fr)
+        if e != 0:
+            continue
+        lat, lon, alt = _teme_to_geodetic(r, jd, fr)
+        satnum = getattr(sat, "satnum_str", None) or rec.get("norad_cat_id")
+        name = (rec.get("tle0") or "").strip().lstrip("0").strip() or str(satnum or "")
+        out.append({
+            "id": str(satnum).strip() or "",
+            "name": name,
+            "lat": round(lat, 5),
+            "lon": round(lon, 5),
+            "alt_km": round(alt, 2),
+            "group": group,
+        })
+        if rec.get("updated"):
+            epochs.append(str(rec["updated"]))
+    return out, (max(epochs) if epochs else None)
+
+
+async def _group_elements(group: str) -> tuple[list[dict], str | None]:
+    """CelesTrak GP blob for one group, TTL-cached with a last-good fallback.
+
+    Returns ``(elements, epoch)``. On a fetch failure (403 / "has not updated
+    since ...") falls back to the previous successful blob for that group.
+    """
+    from live_layers import _get_json, _ttl_get
+
+    url = f"{CELESTRAK_GP}?GROUP={quote(group)}&FORMAT=JSON"
+
+    async def _load() -> list[dict]:
+        data = await _get_json(url)
+        if not isinstance(data, list):
+            raise ValueError(f"unexpected CelesTrak payload for {group}")
+        if data:
+            _last_good[group] = data
+        return data
+
+    key = f"celestrak:gp:{group}"
+    try:
+        elements = await _ttl_get(key, SATELLITE_TTL, _load)
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("celestrak_fetch_failed group=%s: %s", group, exc)
+        elements = _last_good.get(group, [])
+        if not elements:
+            return [], None
+    return elements, _max_epoch(elements)
+
+
+async def fetch_satellites(
+    groups: list[str],
+    bbox: str | None = None,
+    limit: int = DEFAULT_LIMIT,
+) -> dict:
+    """Assemble the ``/api/satellites`` payload for the requested groups."""
+    from live_layers import _get_json, _ttl_get, filter_points_bbox, parse_bbox
+
+    now = datetime.now(timezone.utc)
+    satellites: list[dict] = []
+    epoch: str | None = None
+    source = SOURCE_CELESTRAK
+
+    for group in groups:
+        elements, group_epoch = await _group_elements(group)
+        if not elements:
+            continue
+        if group_epoch and (epoch is None or group_epoch > epoch):
+            epoch = group_epoch
+        satellites.extend(propagate_gp(elements, group, now))
+
+    if not satellites:
+        # Fallback only when the CelesTrak cache is entirely empty — never
+        # poll both providers every cycle.
+        async def _load_satnogs() -> list[dict]:
+            data = await _get_json(SATNOGS_TLE, params={"format": "json"})
+            return data if isinstance(data, list) else []
+
+        try:
+            satnogs = await _ttl_get("satnogs:tle", SATELLITE_TTL, _load_satnogs)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("satnogs_fetch_failed: %s", exc)
+            satnogs = []
+        if satnogs:
+            source = SOURCE_SATNOGS
+            for group in groups:
+                rows, sn_epoch = propagate_satnogs_tle(satnogs, group, now)
+                if sn_epoch and (epoch is None or sn_epoch > epoch):
+                    epoch = sn_epoch
+                satellites.extend(rows)
+
+    if bbox:
+        minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+        satellites = filter_points_bbox(
+            satellites, minlon, minlat, maxlon, maxlat, limit,
+        )
+    else:
+        satellites = satellites[:limit]
+
+    return {
+        "satellites": satellites,
+        "source": source,
+        "tle_epoch": epoch,
+        "timestamp": now.isoformat(),
+    }

app/schemas.py

@@ -4,10 +4,10 @@ from __future__ import annotations
 
 from datetime import datetime
 from enum import Enum
-from typing import Optional
+from typing import Literal, Optional
 from uuid import UUID
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 # ─── Enums ───────────────────────────────────────────────────────────────
@@ -63,6 +63,17 @@ class FeedSourceCreate(BaseModel):
     config: Optional[dict] = None
 
 
+class FeedSourceUpdate(BaseModel):
+    """PATCH /api/sources/{id} — only these keys may be set."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: Optional[str] = None
+    url: Optional[str] = None
+    config: Optional[dict] = None
+    enabled: Optional[bool] = None
+
+
 class FeedSourceOut(BaseModel):
     id: UUID
     name: str
@@ -386,3 +397,24 @@ class GeofenceUpdate(BaseModel):
     geojson: Optional[dict] = None
     active: Optional[bool] = None
 
+
+class ConflictZoneOut(BaseModel):
+    """One curated conflict theatre as exposed by GET /api/conflicts."""
+
+    id: str
+    label: str
+    severity: Literal["war", "high", "elevated"]
+    lat: float
+    lon: float
+    description: str
+    eventCount: int
+    lastUpdated: Optional[datetime] = None
+
+
+class ConflictsOut(BaseModel):
+    """Response envelope for GET /api/conflicts."""
+
+    zones: list[ConflictZoneOut]
+    timestamp: datetime
+
+

app/tracks.py

@@ -186,15 +186,14 @@ async def track_range() -> dict:
         async with async_session() as session:
             row = (await session.execute(text(
                 """
-                SELECT
+                SELECT min(t) AS tmin, max(t) AS tmax FROM (
-                  LEAST(
+                  SELECT min(bucket) AS t FROM vessel_tracks_1min
-                    (SELECT min(bucket) FROM vessel_tracks_1min),
+                  UNION ALL SELECT max(bucket) FROM vessel_tracks_1min
-                    (SELECT min(bucket) FROM aircraft_tracks_1min)
+                  UNION ALL SELECT min(bucket) FROM aircraft_tracks_1min
-                  ) AS tmin,
+                  UNION ALL SELECT max(bucket) FROM aircraft_tracks_1min
-                  GREATEST(
+                  UNION ALL SELECT min(poll_at) FROM vessels
-                    (SELECT max(bucket) FROM vessel_tracks_1min),
+                  UNION ALL SELECT max(poll_at) FROM vessels
-                    (SELECT max(bucket) FROM aircraft_tracks_1min)
+                ) s
-                  ) AS tmax
                 """
             ))).mappings().first()
         if not row or row["tmin"] is None:

app/vesselapi.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import asyncio
 import calendar
+import json
 import logging
 import os
 from datetime import date, datetime, timezone
@@ -32,7 +33,7 @@ from config import (
     VESSELAPI_MAX_CALLS_PER_DAY,
 )
 from database import async_session, engine, metadata
-from live_layers import to_marker, upsert_vessel
+from live_layers import parse_bbox, to_marker, upsert_vessel, vessel_last_known, vessel_lock
 
 logger = logging.getLogger("osint.vesselapi")
 
@@ -176,6 +177,49 @@ def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
     return out
 
 
+def utc_day_start(now: datetime) -> datetime:
+    """Floor ``now`` to 00:00:00 UTC."""
+    if now.tzinfo is None:
+        now = now.replace(tzinfo=timezone.utc)
+    now = now.astimezone(timezone.utc)
+    return now.replace(hour=0, minute=0, second=0, microsecond=0)
+
+
+def pick_poll_at(poll_times: list[datetime], as_of: datetime) -> datetime | None:
+    """Latest poll timestamp at or before ``as_of`` (DVR as-of)."""
+    if as_of.tzinfo is None:
+        as_of = as_of.replace(tzinfo=timezone.utc)
+    else:
+        as_of = as_of.astimezone(timezone.utc)
+    eligible: list[datetime] = []
+    for raw in poll_times:
+        ts = raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc)
+        ts = ts.astimezone(timezone.utc)
+        if ts <= as_of:
+            eligible.append(ts)
+    return max(eligible) if eligible else None
+
+
+def snapshot_as_of(rows: list[dict], as_of: datetime) -> list[dict]:
+    """Keep only rows from the latest poll_at ≤ ``as_of``."""
+    chosen = pick_poll_at(
+        [r["poll_at"] for r in rows if r.get("poll_at") is not None],
+        as_of,
+    )
+    if chosen is None:
+        return []
+    out = []
+    for row in rows:
+        ts = row.get("poll_at")
+        if ts is None:
+            continue
+        if ts.tzinfo is None:
+            ts = ts.replace(tzinfo=timezone.utc)
+        if ts.astimezone(timezone.utc) == chosen:
+            out.append(row)
+    return out
+
+
 # ── Durable daily quota (Postgres, survives restarts) ─────────────────────
 # Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.
 
@@ -261,6 +305,183 @@ class PgQuotaStore:
         return (int(existing) if existing else 0) + 1
 
 
+# ── Daily VesselAPI snapshots (DVR as-of + survive restarts) ──────────────
+# Cleared at the UTC day boundary so the table holds today's 5 polls only.
+
+_CREATE_VESSELS_SQL = text(
+    """
+    CREATE TABLE IF NOT EXISTS vessels (
+        mmsi     TEXT NOT NULL,
+        poll_at  TIMESTAMPTZ NOT NULL,
+        lat      DOUBLE PRECISION NOT NULL,
+        lon      DOUBLE PRECISION NOT NULL,
+        heading  DOUBLE PRECISION,
+        speed    DOUBLE PRECISION,
+        label    TEXT,
+        extra    JSONB,
+        PRIMARY KEY (mmsi, poll_at)
+    )
+    """
+)
+_CREATE_VESSELS_POLL_IDX = text(
+    "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)"
+)
+_CREATE_VESSELS_BBOX_IDX = text(
+    "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)"
+)
+
+_vessels_lock = asyncio.Lock()
+_vessels_ensured = False
+
+
+async def ensure_vessels_table() -> None:
+    global _vessels_ensured
+    if _vessels_ensured:
+        return
+    async with _vessels_lock:
+        if _vessels_ensured:
+            return
+        async with engine.begin() as conn:
+            await conn.execute(_CREATE_VESSELS_SQL)
+            await conn.execute(_CREATE_VESSELS_POLL_IDX)
+            await conn.execute(_CREATE_VESSELS_BBOX_IDX)
+        _vessels_ensured = True
+
+
+def _marker_from_vessel_row(r) -> dict:
+    extra = r.get("extra") or {}
+    if isinstance(extra, str):
+        try:
+            extra = json.loads(extra)
+        except (TypeError, ValueError):
+            extra = {}
+    if not isinstance(extra, dict):
+        extra = {}
+    extra.setdefault("src", "vesselapi")
+    poll_at = r.get("poll_at")
+    if poll_at is not None and hasattr(poll_at, "isoformat"):
+        extra["poll_at"] = poll_at.isoformat()
+    marker = to_marker(
+        str(r["id"]), r["lat"], r["lon"],
+        heading=r.get("heading"), speed=r.get("speed"),
+        label=r.get("label") or str(r["id"]),
+        extra=extra,
+    )
+    marker["seen_at"] = extra.get("poll_at") or datetime.now(timezone.utc).isoformat()
+    return marker
+
+
+async def persist_vessel_snapshot(markers: list[dict], poll_at: datetime) -> None:
+    """Write one VesselAPI poll into ``vessels`` (today's snapshots)."""
+    await ensure_vessels_table()
+    if not markers:
+        return
+    async with async_session() as session:
+        for m in markers:
+            vid = str(m.get("id") or "")
+            lat, lon = m.get("lat"), m.get("lon")
+            if not vid or lat is None or lon is None:
+                continue
+            extra = dict(m.get("extra") or {})
+            extra.setdefault("src", "vesselapi")
+            await session.execute(
+                text(
+                    """
+                    INSERT INTO vessels
+                        (mmsi, poll_at, lat, lon, heading, speed, label, extra)
+                    VALUES
+                        (:mmsi, :poll_at, :lat, :lon, :heading, :speed, :label,
+                         CAST(:extra AS jsonb))
+                    ON CONFLICT (mmsi, poll_at) DO UPDATE SET
+                        lat = EXCLUDED.lat,
+                        lon = EXCLUDED.lon,
+                        heading = EXCLUDED.heading,
+                        speed = EXCLUDED.speed,
+                        label = EXCLUDED.label,
+                        extra = EXCLUDED.extra
+                    """
+                ),
+                {
+                    "mmsi": vid,
+                    "poll_at": poll_at,
+                    "lat": float(lat),
+                    "lon": float(lon),
+                    "heading": m.get("heading"),
+                    "speed": m.get("speed"),
+                    "label": m.get("label") or vid,
+                    "extra": json.dumps(extra),
+                },
+            )
+        await session.commit()
+
+
+async def purge_old_vessels(before: datetime | None = None) -> None:
+    """Drop snapshots from before the current UTC day (or ``before``)."""
+    await ensure_vessels_table()
+    cutoff = before or utc_day_start(datetime.now(timezone.utc))
+    async with async_session() as session:
+        await session.execute(
+            text("DELETE FROM vessels WHERE poll_at < :cutoff"),
+            {"cutoff": cutoff},
+        )
+        await session.commit()
+
+
+async def fetch_vessels_as_of(
+    ts: datetime,
+    bbox: str | None = None,
+    limit: int = 2000,
+) -> list[dict]:
+    """Latest VesselAPI poll at or before ``ts`` (DVR as-of, not exact minute)."""
+    try:
+        await ensure_vessels_table()
+        async with async_session() as session:
+            poll = (await session.execute(
+                text("SELECT max(poll_at) FROM vessels WHERE poll_at <= :ts"),
+                {"ts": ts},
+            )).scalar()
+            if poll is None:
+                return []
+            sql = """
+                SELECT mmsi AS id, lat, lon, heading, speed, label, extra, poll_at
+                FROM vessels
+                WHERE poll_at = :poll
+            """
+            params: dict = {"poll": poll, "limit": limit}
+            if bbox:
+                minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+                sql += (
+                    " AND lon BETWEEN :minlon AND :maxlon"
+                    " AND lat BETWEEN :minlat AND :maxlat"
+                )
+                params.update(
+                    minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat,
+                )
+            sql += " LIMIT :limit"
+            rows = (await session.execute(text(sql), params)).mappings().all()
+        return [_marker_from_vessel_row(r) for r in rows]
+    except Exception:
+        logger.exception("VesselAPI snapshot fetch failed")
+        return []
+
+
+async def hydrate_last_known() -> int:
+    """Seed in-memory last-known from today's latest poll (app boot)."""
+    try:
+        rows = await fetch_vessels_as_of(datetime.now(timezone.utc))
+    except Exception:
+        logger.exception("VesselAPI hydrate failed")
+        return 0
+    if not rows:
+        return 0
+    async with vessel_lock:
+        for m in rows:
+            vid = str(m.get("id") or "")
+            if vid:
+                vessel_last_known[vid] = m
+    return len(rows)
+
+
 # ── Budget / scheduling (pure, unit-testable) ─────────────────────────────
 
 def days_left_in_month(now: datetime) -> int:
@@ -406,6 +627,11 @@ async def poll_once(store, boxes: list[tuple[float, float, float, float]], key:
     markers = transform_vesselapi_payload(data)
     for m in markers:
         await upsert_vessel(m)
+    try:
+        await persist_vessel_snapshot(markers, now)
+        await purge_old_vessels(utc_day_start(now))
+    except Exception:  # noqa: BLE001 — live overlay must not die on persist
+        logger.exception("VesselAPI snapshot persist failed")
     logger.info(
         "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
         len(markers), remaining, calls,

app/ws_manager.py

@@ -8,10 +8,18 @@ from __future__ import annotations
 
 import asyncio
 from typing import Any
+from uuid import UUID
 
 BBox = tuple[float, float, float, float]  # minlon, minlat, maxlon, maxlat
 
 
+def _uuid_str(value: object) -> str | None:
+    try:
+        return str(UUID(str(value)))
+    except (ValueError, TypeError, AttributeError):
+        return None
+
+
 def point_in_bbox(lon: float, lat: float, bbox: BBox | None) -> bool:
     """True if (lon, lat) sits inside an axis-aligned viewport."""
     if bbox is None:
@@ -26,6 +34,7 @@ class ConnectionManager:
     def __init__(self) -> None:
         self._queues: dict[str, asyncio.Queue] = {}
         self._viewports: dict[str, BBox] = {}
+        self._watched: dict[str, set[str]] = {}
 
     def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue:
         q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
@@ -35,6 +44,21 @@ class ConnectionManager:
     def unregister(self, client_id: str) -> None:
         self._queues.pop(client_id, None)
         self._viewports.pop(client_id, None)
+        self._watched.pop(client_id, None)
+
+    def set_watched_geofences(self, client_id: str, ids: list[str]) -> None:
+        """Watch these fence UUIDs so geofence_alert delivers off-viewport.
+
+        Invalid UUIDs are ignored. Empty list = watch none (viewport-only).
+        """
+        if client_id not in self._queues:
+            return
+        watched: set[str] = set()
+        for raw in ids:
+            uid = _uuid_str(raw)
+            if uid is not None:
+                watched.add(uid)
+        self._watched[client_id] = watched
 
     def set_viewport(self, client_id: str, bbox: BBox) -> None:
         if client_id in self._queues:
@@ -59,13 +83,21 @@ class ConnectionManager:
     ) -> int:
         """Enqueue `{type, payload}` for clients whose viewport contains the point.
 
-        Drops the oldest queued message if a client's buffer is full so a slow
+        kind=geofence_alert also delivers when payload.geofence_id is in the
-        tab cannot stall ingest. Returns the number of clients that got a copy.
+        client's watch set (even if the point is off-viewport). Other kinds
+        stay viewport-only. Drops the oldest queued message if a client's
+        buffer is full. Returns the number of clients that got a copy.
         """
         msg = {"type": kind, "payload": payload}
         sent = 0
+        gid = _uuid_str(payload.get("geofence_id")) if kind == "geofence_alert" else None
         for client_id, queue in list(self._queues.items()):
-            if not point_in_bbox(lon, lat, self._viewports.get(client_id)):
+            in_view = point_in_bbox(lon, lat, self._viewports.get(client_id))
+            if kind == "geofence_alert":
+                watching = gid is not None and gid in self._watched.get(client_id, set())
+                if not in_view and not watching:
+                    continue
+            elif not in_view:
                 continue
             if queue.full():
                 try:

deploy/README.md

@@ -1,31 +0,0 @@
-# systemd unit template — copy to /etc/systemd/system/osint-masscan.service
-#
-# The masscan service is a CONTINUOUS rolling sweep (a full IPv4 pass at a
-# conservative rate takes ~5 days), so it runs as a long-lived service, NOT a
-# daily timer. The [Install] WantedBy means it starts at boot and Restart=always
-# keeps it up. Install steps (run once on the Pi, as root):
-#
-#   apt install -y masscan      # or: apt-get install masscan
-#   mkdir -p /etc/osint-dashboard /opt/siriusdevops
-#   cp deploy/masscan-excludes.txt /etc/osint-dashboard/masscan-excludes.txt
-#
-#   # Optional tuning (override env in this file; the DB_* values in the unit
-#   # already point at the host-published Postgres on 127.0.0.1:5432):
-#   cat > /etc/osint-dashboard/masscan.env <<'EOF'
-#   MASSCAN_RANGE=0.0.0.0/0
-#   MASSCAN_PORTS=554
-#   MASSCAN_RATE=1000
-#   EOF
-#
-#   # Venv for the scanner (host-level, not the compose image):
-#   cd /opt/siriusdevops/osint-dashboard
-#   python3 -m venv .venv-masscan
-#   .venv-masscan/bin/pip install -r app/requirements.txt
-#
-#   install -m 644 deploy/osint-masscan.service /etc/systemd/system/
-#   systemctl daemon-reload
-#   systemctl enable --now osint-masscan
-#
-# Watch:  journalctl -u osint-masscan -f
-# DB:     writes into the same Postgres the compose stack uses (127.0.0.1:5432)
-#         so findings appear on the dashboard camera map automatically.

deploy/masscan-excludes.txt

@@ -1,33 +0,0 @@
-# masscan excludefile — never probe these ranges.
-# RFC1918 private + loopback + link-local + multicast + documentation/bogons.
-# The service refuses to start if this file is missing (fail closed).
-
-# Loopback
-127.0.0.0/8
-
-# RFC1918 private
-10.0.0.0/8
-172.16.0.0/12
-192.168.0.0/16
-
-# Link-local
-169.254.0.0/16
-
-# CGNAT (RFC 6598)
-100.64.0.0/10
-
-# Multicast + reserved
-224.0.0.0/4
-240.0.0.0/4
-
-# Documentation / benchmark / example ranges (never real hosts)
-0.0.0.0/8
-192.0.2.0/24
-198.51.100.0/24
-203.0.113.0/24
-192.0.0.0/24
-198.18.0.0/15
-255.255.255.255/32
-
-# Carrier NAT / TEST-NET leftovers
-233.252.0.0/24

deploy/osint-masscan.service

@@ -1,29 +0,0 @@
-[Unit]
-Description=OSINT dashboard — masscan rolling sweep (open RTSP port 554)
-Documentation=https://forgejo.siriusdevops.com/sirius/osint-dashboard
-After=network-online.target
-Wants=network-online.target
-
-[Service]
-Type=simple
-# masscan needs raw sockets (CAP_NET_RAW) — run as root on the Pi host.
-User=root
-WorkingDirectory=/opt/siriusdevops/osint-dashboard
-EnvironmentFile=-/etc/osint-dashboard/masscan.env
-# Point at the compose-published Postgres on the HOST (127.0.0.1:5432), not the
-# docker service name 'postgres' which doesn't resolve outside the compose net.
-Environment=DB_HOST=127.0.0.1
-Environment=DB_PORT=5432
-Environment=DB_USER=osint
-Environment=DB_PASSWORD=osint
-Environment=DB_NAME=osint_data
-Environment=MASSCAN_EXCLUDEFILE=/etc/osint-dashboard/masscan-excludes.txt
-ExecStart=/opt/siriusdevops/osint-dashboard/.venv-masscan/bin/python app/run_masscan_service.py
-Restart=always
-RestartSec=10
-# Log the sweep to journald (read with: journalctl -u osint-masscan -f)
-StandardOutput=journal
-StandardError=journal
-
-[Install]
-WantedBy=multi-user.target

docker-compose.yml

@@ -138,6 +138,8 @@ services:
       FIRMS_DATASETS: ${FIRMS_DATASETS:-VIIRS_NOAA20_NRT,VIIRS_NOAA21_NRT}
       FIRMS_BBOX: ${FIRMS_BBOX:--180,-60,180,75}
       OSINT_USER_AGENT: ${OSINT_USER_AGENT:-osint-dashboard/1.0 (self-hosted; lancewalters94@gmail.com)}
+      NOMINATIM_URL: ${NOMINATIM_URL:-https://nominatim.openstreetmap.org}
+      NOMINATIM_MIN_INTERVAL: ${NOMINATIM_MIN_INTERVAL:-1.0}
       AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-}
       AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66}
       AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1}
@@ -168,7 +170,7 @@ services:
   # Listens on 8000 INSIDE the container (the app already owns host 8000);
   # published on host loopback 127.0.0.1:8001 only.
   titiler:
-    image: ghcr.io/developmentseed/titiler:latest
+    image: ghcr.io/developmentseed/titiler:latest@sha256:1809958d063543e3ec858259536002b2de78e9f8f09a22a8d9591bdc2b550b14
     pull_policy: missing
     container_name: osint-titiler
     platform: linux/arm64

docs/free-data-streams.md

@@ -2,7 +2,7 @@
 
 Builder brief for backend + frontend. Researched 2026-08-27. Every endpoint below was either live-probed from this machine or taken from the provider’s current docs. Prefer **free, no-key, CORS-open** sources first. Keys are called out explicitly.
 
-This is **not** a camera-discovery / masscan change. Existing camera rules still apply: never emit `rtsp://` hrefs; masscan pins go through `/api/cameras/{id}/snapshot`; HTTP directory cams use `/stream` MJPEG.
+This is **not** a camera-discovery change. Existing camera rules still apply: never emit `rtsp://` hrefs; camera pins go through `/api/cameras/{id}/snapshot`; HTTP directory cams use `/stream` MJPEG.
 
 ---
 
@@ -13,7 +13,7 @@ This is **not** a camera-discovery / masscan change. Existing camera rules still
 | NASA FIRMS VIIRS hotspots | Ingested (`app/fire_sources.py` → NATS `events.fire` → `fires` hypertable → `GET /api/fires`) | Needs free `FIRMS_MAP_KEY`. See `docs/firms.md`. |
 | NASA GIBS basemaps | Frontend tiles via `app/gibs_map.py` | No key. CORS `*`. |
 | GIBS VIIRS thermal tiles | Documented, not wired as overlay | Same GIBS stack; no key. |
-| Cameras | Scraper + masscan → `cameras` table | Defaults already include ALERTWest JPEGs + Live-Environment-Streams HLS/YouTube GeoJSON. |
+| Cameras | Scraper → `cameras` table | Defaults already include ALERTWest JPEGs + Live-Environment-Streams HLS/YouTube GeoJSON. |
 | News / RSS / GDELT / USGS quakes | Ingest | Out of scope for this brief. |
 
 **Action for existing fire ingest:** NASA will stop Suomi NPP product delivery on **2026-11-01**. Switch `FIRMS_DATASET` from `VIIRS_SNPP_NRT` to `VIIRS_NOAA20_NRT` and/or `VIIRS_NOAA21_NRT` before then.[20]
@@ -261,7 +261,7 @@ Use later if you want commuter rail / subway vehicle positions (LA Metro, MTA, e
 
 ## 6. Open video / camera feeds (official public only)
 
-Do **not** add Insecam-style random IP cams as a new source. The scraper already has a public list + masscan; this section is **agency-published** JPEG/HLS.
+Do **not** add Insecam-style random IP cams as a new source. The scraper already has a public list; this section is **agency-published** JPEG/HLS.
 
 ### 6.1 Already wired
 
@@ -304,7 +304,7 @@ Do not call the YouTube Data API unless you want search. Embedding existing stre
 
 ### 6.5 Skip
 
-- Insecam / random “public IP cam” aggregators — ToS / privacy / already covered by masscan ethics.
+- Insecam / random “public IP cam” aggregators — ToS / privacy.
 - TrafficLand, EarthCam commercial APIs.
 - SkylineWebcams — scraping, not an API.
 
@@ -523,7 +523,7 @@ Attribution bar (required): OpenSky / ADSB.lol ODbL / Amtraker / RainViewer / IE
 
 ## 12. Legal / ethics (non-negotiable)
 
-- Masscan / RTSP policy unchanged.
+- RTSP policy unchanged (never emit `rtsp://` hrefs).
 - AISStream: server-side only; do not put the key in JS.[5]
 - OpenSky: non-commercial unless licensed; cite if you publish.[2]
 - ADSB.lol: ODbL share-alike on derived databases.[4]

docs/news.md

@@ -128,11 +128,12 @@ Key set **unchanged** (no `lat`/`lon` on articles; geo lives on `/api/news/map`)
 `?kind=daily_recap` pins the nightly 24h recap. Empty DB → `[]` (no crash).
 Malformed `kind` → `422`.
 
-### GET /api/news/ticker — flagged HUD headlines
+### GET /api/news/ticker — HUD headlines
 
-Critical/high `news_items` with `kind=ticker` only. Do **not** reuse
+Critical/high `news_items` with `kind=ticker` first. If none are flagged,
-`GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a dump of
+medium/low ticker rows fill the tape so the dock is not blank. Do **not**
-the whole brief.
+reuse `GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a
+dump of the whole brief.
 
 | Query param | Meaning | Default |
 |---|---|---|
@@ -226,10 +227,10 @@ markdown json fences, then brace-slices:
 }
 ```
 
-Persist ticker/map only for `importance` in `critical`/`high`. Map rows also
+Persist ticker for critical/high first; if none, persist medium/low so the
-need valid coords; Unknown / invented places are dropped. Caps: 12 ticker
+tape is not empty. Map rows stay critical/high with valid coords; Unknown /
-(≤140 chars, no markdown), 20 map. Empty ticker is allowed. `summary_en`
+invented places are dropped. Caps: 12 ticker (≤140 chars, no markdown), 20
-lands in `article_summaries.summary_text`.
+map. `summary_en` lands in `article_summaries.summary_text`.
 
 ## Configuration (all via env / `.env`)
 

news/summerizer/intel.py

@@ -7,6 +7,7 @@ import re
 
 _EMPTY = {"summary_en": "", "ticker": [], "map_items": []}
 _KEEP = frozenset({"critical", "high"})
+_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3}
 _THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
 _FENCE_RE = re.compile(r"```(?:json)?", re.IGNORECASE)
 
@@ -58,19 +59,29 @@ def _trimmed_headline(row: dict, limit: int) -> str:
 
 
 def select_ticker(rows: list) -> list:
-    out = []
+    flagged = []
+    medium = []
+    low = []
     for row in rows:
-        if row.get("importance") not in _KEEP:
+        imp = row.get("importance")
+        if imp not in _RANK:
             continue
         headline = _trimmed_headline(row, TICKER_HEADLINE_MAX)
         if not headline:
             continue
         item = dict(row)
         item["headline"] = headline
-        out.append(item)
+        if imp in _KEEP:
-        if len(out) >= TICKER_CAP:
+            flagged.append(item)
+        elif imp == "medium":
+            medium.append(item)
+        else:
+            low.append(item)
+        if len(flagged) >= TICKER_CAP:
             break
-    return out
+    if flagged:
+        return flagged[:TICKER_CAP]
+    return (medium + low)[:TICKER_CAP]
 
 
 def select_map(items: list) -> list:

news/summerizer/nous_client.py

@@ -8,6 +8,10 @@ import httpx
 
 _DEFAULT_UA = "osint-dashboard-news-summarizer"
 _DEFAULT_BASE = "https://inference-api.nousresearch.com/v1"
+_JSON_SYSTEM = (
+    "You are an OSINT executive briefer. Reply with a single complete JSON object. "
+    "Never truncate mid-sentence. If you run out of room, drop the lowest-priority item."
+)
 
 
 def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str:
@@ -17,20 +21,37 @@ def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str:
         "Authorization": f"Bearer {api_key}",
         "User-Agent": os.environ.get("OSINT_USER_AGENT") or _DEFAULT_UA,
     }
+    max_tokens = 8192 if json_mode else 4096
+    timeout = 120.0 if json_mode else 60.0
+    messages = [{"role": "user", "content": prompt}]
+    if json_mode:
+        messages = [
+            {"role": "system", "content": _JSON_SYSTEM},
+            {"role": "user", "content": prompt},
+        ]
     payload = {
         "model": model,
-        "messages": [{"role": "user", "content": prompt}],
+        "messages": messages,
         "temperature": 0.2,
-        "max_tokens": 4096,
+        "max_tokens": max_tokens,
     }
     if json_mode:
         payload["response_format"] = {"type": "json_object"}
+    last_content = ""
     try:
-        with httpx.Client(timeout=60.0) as client:
+        for attempt in range(2):
-            resp = client.post(url, headers=headers, json=payload)
+            with httpx.Client(timeout=timeout) as client:
-        if resp.status_code == 401 or resp.status_code >= 500:
+                resp = client.post(url, headers=headers, json=payload)
-            return ""
+            if resp.status_code == 401 or resp.status_code >= 500:
-        data = resp.json()
+                return ""
-        return data["choices"][0]["message"]["content"]
+            data = resp.json()
+            choice = (data.get("choices") or [{}])[0]
+            last_content = (choice.get("message") or {}).get("content") or ""
+            finish = choice.get("finish_reason")
+            if finish == "length" and attempt == 0:
+                payload["max_tokens"] = min(int(payload["max_tokens"]) * 2, 16384)
+                continue
+            return last_content
+        return last_content
     except Exception:
         return ""

news/summerizer/summarizer.py

@@ -94,7 +94,7 @@ FUTURES_TICKERS = {
 MAP_PROMPT_DEFAULT = """\
 You are a precise, factual OSINT news processor. Your ONLY source of information is the articles provided below. Do NOT add external knowledge, assumptions, training data, or invented facts.
 
-Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event.
+Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event. If the batch has no critical/high stories, still extract minor incidents and crime reports.
 
 Write every field in English. Translate if the article is not English.
 
@@ -133,7 +133,11 @@ You are writing an English operator HUD brief from the article facts in DATA bel
 
 Always write a real summary_en that recaps the most important stories present in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.
 
-ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts.
+Lead with critical and high breaking events. If DATA has no critical/high stories, fill the brief with minor incidents and crime reports rather than writing an empty or unfinished brief. Never truncate mid-sentence; finish every sentence. If you run out of room, drop the lowest-priority item instead of cutting a line short.
+
+ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.
+
+map_items may be empty if no located critical/high event is explicit in the data.
 
 Demand a single JSON object (no markdown fences) with this exact shape:
 
@@ -143,9 +147,9 @@ Demand a single JSON object (no markdown fences) with this exact shape:
   "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
 }
 
-ticker: only critical and high, max 12, ≤140 chars, no markdown.
+ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
-map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
+map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
-summary_en: English markdown brief of breaking important news for an operator HUD (bullets or short paragraphs). Cover the actual stories in DATA.
+summary_en: English markdown executive brief for an operator HUD (4–8 complete bullets or short paragraphs). Cover the actual stories in DATA. Complete — never an unfinished sentence.
 
 DATA:
 {final_input}
@@ -156,7 +160,9 @@ You are writing a daily recap of the last 24 hours of news for an OSINT operator
 
 Always write a real summary_en daily recap of the most important stories in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.
 
-ticker and map_items may be empty if nothing is critical or high. Never replace summary_en with a canned empty-brief sentence when DATA contains article facts.
+Lead with critical and high breaking events. If DATA has no critical/high stories, fill the recap with minor incidents and crime reports rather than writing an empty or unfinished recap. Never truncate mid-sentence; finish every sentence.
+
+ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.
 
 Demand a single JSON object (no markdown fences) with this exact shape:
 
@@ -166,9 +172,9 @@ Demand a single JSON object (no markdown fences) with this exact shape:
   "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
 }
 
-ticker: only critical and high, max 12, ≤140 chars, no markdown.
+ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
-map_items: only critical and high where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
+map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
-summary_en: English markdown daily recap of the last 24 hours of breaking important news. Cover the actual stories in DATA.
+summary_en: English markdown daily recap of the last 24 hours. Complete sentences. Cover the actual stories in DATA.
 
 DATA:
 {final_input}

news/summerizer/tests/test_intel.py

@@ -32,6 +32,16 @@ def test_select_ticker_keeps_critical_high_caps_12():
     assert len(out) == 12
     assert all(r["importance"] in ("critical", "high") for r in out)
 
+
+def test_select_ticker_falls_back_to_medium_low_when_nothing_flagged():
+    rows = [
+        {"headline": "shop theft", "importance": "low"},
+        {"headline": "highway crash", "importance": "medium"},
+        {"headline": "none", "importance": "none"},
+    ]
+    out = select_ticker(rows)
+    assert [r["headline"] for r in out] == ["highway crash", "shop theft"]
+
 def test_select_map_requires_valid_coords_and_flag():
     items = [
         {"headline": "A", "importance": "critical", "lat": 50.45, "lon": 30.52, "location_name": "Kyiv"},

news/summerizer/tests/test_nous_client.py

@@ -67,6 +67,33 @@ def test_json_mode_sets_response_format(monkeypatch):
     captured = _install_fake(monkeypatch, lambda *a: _ok_response("{}"))
     chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
     assert captured["json"]["response_format"] == {"type": "json_object"}
+    assert captured["json"]["max_tokens"] >= 8192
+    roles = [m["role"] for m in captured["json"]["messages"]]
+    assert "system" in roles
+    assert "user" in roles
+
+
+def test_retries_once_when_finish_reason_is_length(monkeypatch):
+    calls = {"n": 0}
+
+    def post_impl(*a):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            resp = MagicMock()
+            resp.status_code = 200
+            resp.json.return_value = {
+                "choices": [{
+                    "message": {"content": "{\"summary_en\": \"cut off"},
+                    "finish_reason": "length",
+                }]
+            }
+            return resp
+        return _ok_response('{"summary_en": "complete brief."}')
+
+    _install_fake(monkeypatch, post_impl)
+    out = chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
+    assert calls["n"] == 2
+    assert "complete brief" in out
 
 
 def test_401_returns_empty_string(monkeypatch):

news/summerizer/tests/test_prompts.py

@@ -22,6 +22,14 @@ def test_summary_prompt_focuses_on_breaking_news_not_futures():
     assert "commodity" in p or "market" in p
 
 
+def test_summary_prompt_covers_critical_then_incidents():
+    p = SUMMARY_PROMPT_DEFAULT.lower()
+    assert "critical" in p
+    assert "crime" in p
+    assert "incident" in p
+    assert "complete" in p or "truncat" in p or "unfinished" in p or "mid-sentence" in p
+
+
 def test_summary_prompt_does_not_bail_out_with_canned_empty_brief():
     p = SUMMARY_PROMPT_DEFAULT
     assert "AND STOP" not in p

tests/test_aircraft_popup_frontend.py

@@ -0,0 +1,57 @@
+"""Aircraft popup enrichment + emergency/MIL layer contract (static HTML)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def _fn(name: str, nxt: str) -> str:
+    return HTML.split(f"function {name}", 1)[1].split(f"function {nxt}", 1)[0]
+
+
+def test_popup_has_required_adsb_fields_and_photo():
+    js = _fn("pointPopup", "loadPlanePhoto")
+    for field in ("callsign", "hex", "registration", "type", "alt", "gs", "squawk"):
+        assert f"add('{field}'" in js
+    assert "class=\"ps-photo\"" in js or "class='ps-photo'" in js
+    assert "wikipedia" not in js.lower()
+    assert "ceo" not in js.lower()
+
+
+def test_emergency_badge_and_squawk_codes():
+    assert "role-badge emergency" in HTML
+    assert "hdg-emerg" in HTML
+    assert "EMERG_SQUAWK" in HTML
+    assert "['7700', '7600', '7500']" in HTML
+    emerg = HTML.split("function acIsEmergency", 1)[1].split("function acVisible", 1)[0]
+    assert "EMERG_SQUAWK.has(sq)" in emerg
+    color = HTML.split("function acColor", 1)[1].split("function connectLiveWs", 1)[0]
+    assert "acIsEmergency(p)" in color
+    assert "#ff5d5d" in color
+
+
+def test_mil_toggle_hidden_until_role_flag_and_never_hits_adsb_lol():
+    assert 'id="lp-ac-mil-row"' in HTML
+    assert 'id="lp-ac-mil-on"' in HTML
+    row = HTML.split('id="lp-ac-mil-row"', 1)[1].split(">", 1)[0]
+    assert "hidden" in row
+    on = HTML.split('id="lp-ac-mil-on"', 1)[1].split(">", 1)[0]
+    assert "checked" not in on
+    load = HTML.split("async function loadAircraft", 1)[1].split("async function toggleTrains", 1)[0]
+    assert "/api/aircraft?bbox=" in load
+    assert "api.adsb.lol" not in load
+    assert "noteMilSupport" in load
+    assert "acMilOn" in load
+    note = HTML.split("function noteMilSupport", 1)[1].split("function acColor", 1)[0]
+    assert "extra.role" in note
+    assert "lp-ac-mil-row" in note
+    assert "hidden = false" in note
+
+
+def test_planespotters_lazy_photo_still_wired():
+    assert "function loadPlanePhoto" in HTML
+    assert "/api/aircraft/photo?" in HTML
+    assert "map.on('popupopen', (e) => { loadPlanePhoto(e.popup); });" in HTML

tests/test_api_news.py

@@ -230,6 +230,21 @@ def test_api_news_ticker_returns_only_flagged(clean_news):
     assert item["url"] == "https://example.com/ticker"
 
 
+@requires_db
+def test_api_news_ticker_falls_back_to_lesser_when_nothing_flagged(clean_news):
+    sid = _seed_summary("quiet brief", "2026-08-27T18:05:00+00:00", "Hermes-4.3-36B")
+    _seed_news_item(
+        sid, "ticker", "Shop theft downtown", "low",
+        location_name="Raleigh", url="https://example.com/theft",
+    )
+    resp = _get("/api/news/ticker")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert len(body) == 1
+    assert body[0]["headline"] == "Shop theft downtown"
+    assert body[0]["importance"] == "low"
+
+
 @requires_db
 def test_api_news_map_returns_only_flagged_with_coords(clean_news):
     _seed_flagged_items()

tests/test_api_place.py

@@ -0,0 +1,127 @@
+"""GET /api/place — Nominatim reverse proxy (60s cache, 500 keys, 1 req/s)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+
+from main import app
+from place import cache_key, place_cache, slim_place
+
+BASE = "http://test"
+
+SAMPLE = {
+    "display_name": "Raleigh, Wake County, North Carolina, United States",
+    "name": "Raleigh",
+    "osm_type": "relation",
+    "osm_id": 123,
+    "address": {
+        "city": "Raleigh",
+        "state": "North Carolina",
+        "country": "United States",
+        "country_code": "us",
+        "tourism": "ignore-me",
+    },
+}
+
+
+class _FakeResp:
+    def __init__(self, payload, status=200):
+        self._payload = payload
+        self.status_code = status
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            req = httpx.Request("GET", "https://nominatim.openstreetmap.org/reverse")
+            raise httpx.HTTPStatusError(
+                "upstream", request=req,
+                response=httpx.Response(self.status_code, request=req),
+            )
+
+    def json(self):
+        return self._payload
+
+
+class _FakeNominatim:
+    calls: list[dict] = []
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    async def get(self, url, params=None, headers=None):
+        _FakeNominatim.calls.append({"url": url, "params": params, "headers": headers})
+        return _FakeResp(SAMPLE)
+
+
+def _nominatim_client(**kwargs):
+    return _FakeNominatim()
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+@pytest.fixture(autouse=True)
+def _reset_place(monkeypatch):
+    place_cache.clear()
+    _FakeNominatim.calls = []
+    monkeypatch.setattr("place._http_client", _nominatim_client)
+    monkeypatch.setattr("place.NOMINATIM_MIN_INTERVAL", 0.0)
+    monkeypatch.setattr("place._last_req", 0.0)
+    yield
+    place_cache.clear()
+
+
+def test_slim_place_keeps_address_subset():
+    body = slim_place(35.78, -78.64, SAMPLE)
+    assert body["display_name"].startswith("Raleigh")
+    assert body["name"] == "Raleigh"
+    assert body["address"]["city"] == "Raleigh"
+    assert "tourism" not in body["address"]
+    assert body["attribution"].startswith("© OpenStreetMap")
+
+
+def test_cache_key_quantizes_to_4_decimals():
+    assert cache_key(35.77961, -78.63821) == cache_key(35.77964, -78.63819)
+
+
+def test_place_requires_lat_lon():
+    resp = asyncio.run(_get("/api/place"))
+    assert resp.status_code == 422
+
+
+def test_place_rejects_out_of_range():
+    assert asyncio.run(_get("/api/place?lat=99&lon=0")).status_code == 422
+    assert asyncio.run(_get("/api/place?lat=0&lon=200")).status_code == 422
+
+
+def test_place_reverse_and_cache():
+    r1 = asyncio.run(_get("/api/place?lat=35.7796&lon=-78.6382"))
+    assert r1.status_code == 200
+    body = r1.json()
+    assert body["display_name"].startswith("Raleigh")
+    assert body["lat"] == pytest.approx(35.7796, abs=0.001)
+    assert "max-age=60" in (r1.headers.get("cache-control") or "").lower()
+    assert len(_FakeNominatim.calls) == 1
+    ua = _FakeNominatim.calls[0]["headers"]["User-Agent"]
+    assert "osint-dashboard" in ua.lower() or "@" in ua
+    r2 = asyncio.run(_get("/api/place?lat=35.77961&lon=-78.63821"))
+    assert r2.status_code == 200
+    assert len(_FakeNominatim.calls) == 1  # cache hit, same 4-decimal key
+
+
+def test_place_cache_cap_500():
+    from cachetools import TTLCache
+    assert isinstance(place_cache, TTLCache)
+    assert place_cache.maxsize == 500
+    assert place_cache.ttl == 60

tests/test_api_stats.py

@@ -0,0 +1,87 @@
+"""GET /api/stats HUD counter contract (counts only, small, never 500)."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from datetime import timezone
+
+import httpx
+
+from main import app, _stats_counts
+
+BASE = "http://test"
+
+EXPECTED_KEYS = ("aircraft", "vessels", "trains", "cameras",
+                 "fires", "quakes", "alerts", "timestamp")
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+def test_stats_200_all_keys_present():
+    resp = asyncio.run(_get("/api/stats"))
+    assert resp.status_code == 200
+    body = resp.json()
+    for key in EXPECTED_KEYS:
+        assert key in body, f"missing key {key}"
+    assert "max-age" in (resp.headers.get("cache-control") or "").lower()
+
+
+def test_stats_counters_are_ints():
+    body = asyncio.run(_get("/api/stats")).json()
+    for key in EXPECTED_KEYS:
+        if key == "timestamp":
+            continue
+        assert isinstance(body[key], int), f"{key} is not an int: {body[key]!r}"
+
+
+def test_stats_timestamp_is_iso8601_z():
+    body = asyncio.run(_get("/api/stats")).json()
+    ts = body["timestamp"]
+    # ISO8601 with a trailing Z (we normalize +00:00 -> Z).
+    assert isinstance(ts, str) and ts.endswith("Z")
+    assert re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", ts)
+
+
+def test_stats_payload_is_tiny():
+    resp = asyncio.run(_get("/api/stats"))
+    assert len(resp.content) < 2048, "stats payload must be counts-only, not GeoJSON"
+
+
+def test_stats_counts_reflect_last_known(monkeypatch):
+    """aircraft/vessels/trains/alerts come from in-memory last-known state."""
+    import live_layers
+
+    monkeypatch.setattr(live_layers, "aircraft_last_known", {str(i): {} for i in range(7)})
+    monkeypatch.setattr(live_layers, "vessel_last_known", {str(i): {} for i in range(3)})
+    monkeypatch.setattr(live_layers, "train_count", 11)
+    monkeypatch.setattr(live_layers, "nws_alert_count", 5)
+
+    # _stats_counts imports the dicts/counters inside the function from live_layers,
+    # so monkeypatching the module attributes is what it observes.
+    from main import _stats_counts as fn
+
+    body = asyncio.run(fn())
+    assert body["aircraft"] == 7
+    assert body["vessels"] == 3
+    assert body["trains"] == 11
+    assert body["alerts"] == 5
+
+
+def test_stats_db_failure_degrades_to_zero(monkeypatch):
+    """A down DB yields zeros for the SQL-backed counters, never a 500."""
+    # Make the session factory raise synchronously so the try/except in
+    # _stats_counts degrades the SQL counters to zero (no dangling coroutine).
+    def _raise(*args, **kwargs):
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr("main.async_session", _raise)
+    body = asyncio.run(_stats_counts())
+    assert body["cameras"] == 0
+    assert body["fires"] == 0
+    assert body["quakes"] == 0
+    assert isinstance(body["timestamp"], str)

tests/test_bg_jobs.py

@@ -1,4 +1,4 @@
-"""masscan/ffmpeg stay off the request path (asyncio.create_task)."""
+"""ffmpeg snapshots stay off the request path (asyncio.create_task)."""
 
 from __future__ import annotations
 
@@ -7,36 +7,27 @@ import asyncio
 import bg_jobs
 
 
-def test_schedule_masscan_pass_returns_without_awaiting_scan(monkeypatch):
+def test_bg_jobs_has_no_pps_cap():
-    started = {"n": 0}
+    assert not any(name.endswith("_PPS_CAP") for name in dir(bg_jobs))
-
-    async def slow_pass():
-        started["n"] += 1
-        await asyncio.sleep(30)
-
-    monkeypatch.setattr(bg_jobs, "_run_masscan_capped", slow_pass)
-    bg_jobs._masscan_task = None
-
-    async def run():
-        launched = bg_jobs.schedule_masscan_pass()
-        assert launched is True
-        # Must not have blocked for the 30s pass.
-        assert bg_jobs._masscan_task is not None
-        assert not bg_jobs._masscan_task.done()
-        launched2 = bg_jobs.schedule_masscan_pass()
-        assert launched2 is False  # already running
-        bg_jobs._masscan_task.cancel()
-        try:
-            await bg_jobs._masscan_task
-        except (asyncio.CancelledError, Exception):
-            pass
-        bg_jobs._masscan_task = None
-
-    asyncio.run(run())
 
 
-def test_masscan_rate_cap_is_200():
+def test_camera_preview_has_no_public_feed_probe():
-    assert bg_jobs.MASSCAN_PPS_CAP == 200
+    import camera_preview
+
+    assert not hasattr(camera_preview, "probe_public_feed")
+    assert not hasattr(camera_preview, "_http_feed_url")
+
+
+def test_ingest_routes_exclude_active_discovery():
+    from main import app
+
+    ingest = [
+        getattr(r, "path", "")
+        for r in app.routes
+        if getattr(r, "path", "").startswith("/api/ingest/")
+    ]
+    assert "/api/ingest/fires" in ingest
+    assert all("scan" not in path for path in ingest)
 
 
 def test_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch):

tests/test_conflicts.py

@@ -0,0 +1,132 @@
+"""GET /api/conflicts — curated conflict-zone catalog + event-count roll-up.
+
+No outbound HTTP: event counts come from geocoded rows already (or not) in the
+DB, and the API tests monkeypatch ``main._fetch_geocoded_points`` so no database
+is required for the contract checks.
+"""
+
+from datetime import datetime, timezone
+
+import httpx
+
+from conflicts import SEVERITIES, conflict_zones, zone_event_stats
+from live_layers import overlay_catalog
+from main import app
+
+BASE = "http://test"
+
+
+def _get(path: str, monkeypatch=None, points=None) -> httpx.Response:
+    import asyncio
+
+    async def run() -> httpx.Response:
+        if monkeypatch is not None:
+            async def fake():
+                return points or []
+
+            monkeypatch.setattr("main._fetch_geocoded_points", fake)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+            return await client.get(path)
+
+    return asyncio.run(run())
+
+
+# ── Catalog shape ──────────────────────────────────────────────────────
+
+
+def test_catalog_length():
+    zones = conflict_zones()
+    assert len(zones) == 13
+
+
+def test_catalog_severity_enum():
+    zones = conflict_zones()
+    sevs = {z["severity"] for z in zones}
+    assert sevs.issubset(SEVERITIES)
+    # All three tiers are represented.
+    assert sevs == SEVERITIES
+
+
+def test_catalog_fields_factual_and_complete():
+    zones = conflict_zones()
+    ids = [z["id"] for z in zones]
+    assert len(set(ids)) == len(ids)  # unique ids
+    for z in zones:
+        assert z["label"]
+        assert z["description"].strip()
+        assert -90.0 <= z["lat"] <= 90.0
+        assert -180.0 <= z["lon"] <= 180.0
+        # internal-only bbox is well-formed: (min_lat, min_lon, max_lat, max_lon)
+        min_lat, min_lon, max_lat, max_lon = z["bbox"]
+        assert min_lat <= max_lat and min_lon <= max_lon
+        assert min_lat <= z["lat"] <= max_lat and min_lon <= z["lon"] <= max_lon
+
+
+def test_overlay_catalog_has_conflicts():
+    entry = overlay_catalog()["conflicts"]
+    assert entry["kind"] == "points"
+    assert entry["endpoint"] == "/api/conflicts"
+
+
+# ── Pure counting ──────────────────────────────────────────────────────
+
+TS1 = datetime(2026, 8, 30, 12, 0, tzinfo=timezone.utc)
+TS2 = datetime(2026, 8, 30, 13, 0, tzinfo=timezone.utc)
+
+
+def test_zone_event_stats_counts_and_picks_latest():
+    bbox = (40.0, 20.0, 52.0, 40.0)  # roughly Ukraine
+    points = [
+        (50.45, 30.52, TS1),  # inside
+        (48.0, 25.0, TS2),    # inside, later
+        (0.0, -60.0, TS1),    # outside
+        (15.0, 45.0, TS2),    # outside (lat ok, lon out)
+    ]
+    count, latest = zone_event_stats(points, bbox)
+    assert count == 2
+    assert latest == TS2
+
+
+def test_zone_event_stats_empty_bbox():
+    count, latest = zone_event_stats([], (0.0, 0.0, 1.0, 1.0))
+    assert count == 0
+    assert latest is None
+
+
+# ── API contract (mocked map items, no DB) ─────────────────────────────
+
+
+def test_conflicts_returns_catalog_with_mocked_counts(monkeypatch):
+    points = [
+        (50.45, 30.52, TS1),   # Ukraine
+        (25.03, 121.56, TS2),  # Taiwan Strait
+        (0.0, -60.0, TS1),     # nowhere
+    ]
+    resp = _get("/api/conflicts", monkeypatch=monkeypatch, points=points)
+    assert resp.status_code == 200
+    body = resp.json()
+    assert "zones" in body and "timestamp" in body
+    by_id = {z["id"]: z for z in body["zones"]}
+    assert len(body["zones"]) == 13
+
+    zone = by_id["ukraine"]
+    assert zone["eventCount"] == 1
+    assert zone["lastUpdated"] == TS1.isoformat().replace("+00:00", "Z")
+    assert zone["severity"] == "war"
+
+    assert by_id["taiwan_strait"]["eventCount"] == 1
+    assert by_id["gaza"]["eventCount"] == 0
+    # exact per-zone key contract the frontend consumes
+    assert set(zone.keys()) == {
+        "id", "label", "severity", "lat", "lon",
+        "description", "eventCount", "lastUpdated",
+    }
+
+
+def test_conflicts_empty_db_yields_zero_counts(monkeypatch):
+    resp = _get("/api/conflicts", monkeypatch=monkeypatch, points=[])
+    assert resp.status_code == 200
+    body = resp.json()
+    assert all(z["eventCount"] == 0 for z in body["zones"])
+    assert all(z["lastUpdated"] is None for z in body["zones"])

tests/test_conflicts_frontend.py

@@ -0,0 +1,66 @@
+"""Conflicts Leaflet overlay: default-off toggle, catalog fetch, no jitter."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def _fn(name: str, until: str | None = None) -> str:
+    chunk = HTML.split(f"function {name}", 1)[1]
+    if until:
+        chunk = chunk.split(until, 1)[0]
+    return chunk
+
+
+def test_conflicts_toggle_default_off():
+    assert 'id="lp-conflicts-on"' in HTML
+    assert 'id="conflicts-layer"' in HTML
+    assert "> Conflicts<" in HTML or "> Conflicts</" in HTML
+    on = HTML.split('id="lp-conflicts-on"', 1)[1].split(">", 1)[0]
+    assert "checked" not in on
+
+
+def test_conflicts_fetches_catalog_not_liveuamap():
+    js = _fn("loadConflicts", "/* ═══════════════ INITIAL LOAD")
+    assert "/api/conflicts" in js
+    assert "liveuamap.com" not in HTML.lower()
+    assert "Math.random" not in js
+    assert "jitter" not in js.lower()
+
+
+def test_conflicts_not_refetched_on_moveend():
+    refresh = HTML.split("function refreshLiveOverlays", 1)[1].split(
+        "function addExtraAttrib", 1
+    )[0]
+    assert "loadConflicts" not in refresh
+    assert "probeConflicts" not in refresh
+    init = HTML.split("function initMap", 1)[1].split("function readMapPrefs", 1)[0]
+    assert "probeConflicts()" in init
+    assert "loadConflicts(true)" not in init
+    assert "paintConflicts()" not in init
+
+
+def test_conflicts_hides_toggle_on_404():
+    js = _fn("loadConflicts", "/* ═══════════════ INITIAL LOAD")
+    assert "r.status === 404" in js
+    assert "hideConflictsToggle()" in js
+    hide = _fn("hideConflictsToggle", "function paintConflicts")
+    assert "row.hidden = true" in hide
+    assert "lp-conflicts-on" in hide
+
+
+def test_conflicts_popup_and_severity_colors():
+    paint = _fn("paintConflicts", "async function probeConflicts")
+    assert "z.label" in paint
+    assert "z.description" in paint
+    assert "eventCount" in paint
+    assert "L.circleMarker" in paint
+    assert "z.lat == null || z.lon == null" in paint
+    assert "Number.isFinite(lat)" in paint
+    color = _fn("conflictSeverityColor", "function hideConflictsToggle")
+    assert "war" in color and "#ff2a6d" in color
+    assert "high" in color and "#fb923c" in color
+    assert "elevated" in color and "#facc15" in color

tests/test_frontend_reliability.py

@@ -72,6 +72,14 @@ def test_chokepoint_skips_aisstream_subscribe_outside_conus():
     assert "minlat,minlon,maxlat,maxlon" in HTML.split("function chokepointLeafletBounds")[1][:400]
 
 
+def test_news_ticker_polls_more_often_than_summarizer_cycle():
+    assert "NEWS_REFRESH_MS" in HTML
+    # Summarizer is 15 min; ticker should refresh on a shorter cadence so
+    # lesser-news fills show up without waiting for the next brief.
+    line = [ln for ln in HTML.splitlines() if "NEWS_REFRESH_MS" in ln][0]
+    assert "900000" not in line
+
+
 def test_phone_chokepoints_use_select_not_buttons():
     mobile = HTML.split("@media (max-width: 820px)")[1].split("@media (prefers-reduced-motion")[0]
     assert "#chokepoint-select { display: block; }" in mobile

tests/test_geofence.py

@@ -51,23 +51,33 @@ def test_matching_geofences_only_active_hits():
     assert matching_geofences(-122.4, 37.7, fences) == []
 
 
+FENCE_ID = "11111111-1111-1111-1111-111111111111"
+NC_VIEW = (-80.0, 35.0, -78.0, 36.0)
+SF_VIEW = (-123.0, 37.0, -121.0, 38.0)
+
+
+def _alert_payload(gid=FENCE_ID):
+    return {
+        "geofence_id": gid,
+        "geofence_name": "NC",
+        "source_kind": "ais",
+        "entity_id": "366123456",
+        "lat": 35.5,
+        "lon": -79.0,
+    }
+
+
 def test_geofence_alert_fans_out_only_to_viewport_clients():
     mgr = ConnectionManager()
     q_nc = mgr.register("nc")
     q_sf = mgr.register("sf")
-    mgr.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
+    mgr.set_viewport("nc", NC_VIEW)
-    mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0))
+    mgr.set_viewport("sf", SF_VIEW)
 
     async def run():
-        payload = {
+        n = await mgr.publish_point(
-            "geofence_id": "a",
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
-            "geofence_name": "NC",
+        )
-            "source_kind": "ais",
-            "entity_id": "366123456",
-            "lat": 35.5,
-            "lon": -79.0,
-        }
-        n = await mgr.publish_point("geofence_alert", payload, lat=35.5, lon=-79.0)
         assert n == 1
         msg = q_nc.get_nowait()
         assert msg["type"] == "geofence_alert"
@@ -77,6 +87,109 @@ def test_geofence_alert_fans_out_only_to_viewport_clients():
     asyncio.run(run())
 
 
+def test_off_viewport_watch_receives_geofence_alert():
+    mgr = ConnectionManager()
+    q_sf = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+    mgr.set_watched_geofences("sf", [FENCE_ID])
+
+    async def run():
+        n = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n == 1
+        msg = q_sf.get_nowait()
+        assert msg["type"] == "geofence_alert"
+        assert msg["payload"]["geofence_id"] == FENCE_ID
+
+    asyncio.run(run())
+
+
+def test_off_viewport_without_watch_does_not_receive_geofence_alert():
+    mgr = ConnectionManager()
+    q_sf = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+
+    async def run():
+        n = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n == 0
+        assert q_sf.empty()
+
+    asyncio.run(run())
+
+
+def test_on_viewport_receives_geofence_alert_without_watch():
+    mgr = ConnectionManager()
+    q_nc = mgr.register("nc")
+    mgr.set_viewport("nc", NC_VIEW)
+
+    async def run():
+        n = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n == 1
+        assert q_nc.get_nowait()["type"] == "geofence_alert"
+
+    asyncio.run(run())
+
+
+def test_ais_stays_viewport_only_even_when_watching():
+    mgr = ConnectionManager()
+    q_sf = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+    mgr.set_watched_geofences("sf", [FENCE_ID])
+
+    async def run():
+        n = await mgr.publish_point("ais", {"id": "366123456"}, lat=35.5, lon=-79.0)
+        assert n == 0
+        assert q_sf.empty()
+
+    asyncio.run(run())
+
+
+def test_invalid_watch_uuids_ignored_empty_list_clears():
+    mgr = ConnectionManager()
+    q = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+    mgr.set_watched_geofences("sf", ["not-a-uuid", FENCE_ID, "also-bad"])
+
+    async def run():
+        n = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n == 1
+        q.get_nowait()
+        mgr.set_watched_geofences("sf", [])
+        n2 = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n2 == 0
+        assert q.empty()
+
+    asyncio.run(run())
+
+
+def test_unregister_clears_watched_geofences():
+    mgr = ConnectionManager()
+    q = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+    mgr.set_watched_geofences("sf", [FENCE_ID])
+    mgr.unregister("sf")
+    q2 = mgr.register("sf")
+    mgr.set_viewport("sf", SF_VIEW)
+
+    async def run():
+        n = await mgr.publish_point(
+            "geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
+        )
+        assert n == 0
+        assert q2.empty()
+
+    asyncio.run(run())
+
+
 def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
     """FIRMS ingest in the ingester has an empty in-process cache — still ST_Intersects."""
     import geofence
@@ -134,3 +247,117 @@ def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
     inserts = [p for p in executed if isinstance(p, dict)]
     assert inserts and inserts[0]["source_kind"] == "firms"
     assert "commit" in executed
+
+
+def test_list_alerts_sql_filters(monkeypatch):
+    captured: dict = {}
+
+    class FakeResult:
+        def mappings(self):
+            return self
+
+        def all(self):
+            return []
+
+    class FakeSession:
+        async def execute(self, stmt, params=None):
+            captured["sql"] = str(stmt)
+            captured["params"] = params
+            return FakeResult()
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    monkeypatch.setattr(geofence, "async_session", FakeSession)
+    from datetime import datetime, timezone
+
+    since = datetime(2026, 8, 28, tzinfo=timezone.utc)
+    until = datetime(2026, 8, 29, tzinfo=timezone.utc)
+
+    async def run():
+        return await geofence.list_alerts(
+            geofence_id=FENCE_ID, since=since, until=until,
+            source_kind="firms", limit=5,
+        )
+
+    assert asyncio.run(run()) == []
+    sql = captured["sql"].lower()
+    assert "geofence_id" in sql
+    assert "created_at >=" in sql
+    assert "created_at <=" in sql
+    assert "source_kind" in sql
+    assert captured["params"]["geofence_id"] == FENCE_ID
+    assert captured["params"]["source_kind"] == "firms"
+    assert captured["params"]["limit"] == 5
+
+
+def test_alembic_fence_created_index_exists():
+    from pathlib import Path
+    text = Path(__file__).resolve().parent.parent.joinpath(
+        "alembic/versions/011_geofence_alerts_fence.py",
+    ).read_text()
+    assert "ix_geofence_alerts_fence_created" in text
+    assert "010_bbox_gist" in text
+
+
+def test_snapshot_at_404_when_fence_missing(monkeypatch):
+    geofence._cache.clear()
+
+    async def boom():
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(geofence, "refresh_cache", boom)
+
+    async def run():
+        from datetime import datetime, timezone
+        return await geofence.snapshot_at(
+            FENCE_ID, datetime(2026, 8, 28, 12, 4, tzinfo=timezone.utc),
+        )
+
+    assert asyncio.run(run()) is None
+
+
+def test_snapshot_queries_st_intersects(monkeypatch):
+    geofence._cache[:] = [{
+        "id": FENCE_ID, "name": "NC", "geojson": NC_BOX, "active": True,
+    }]
+    sqls: list[str] = []
+
+    class FakeResult:
+        def mappings(self):
+            return self
+
+        def all(self):
+            return []
+
+    class FakeSession:
+        async def execute(self, stmt, params=None):
+            sqls.append(str(stmt))
+            return FakeResult()
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    monkeypatch.setattr(geofence, "async_session", FakeSession)
+
+    async def run():
+        from datetime import datetime, timezone
+        return await geofence.snapshot_at(
+            FENCE_ID, datetime(2026, 8, 28, 12, 4, 30, tzinfo=timezone.utc),
+        )
+
+    body = asyncio.run(run())
+    assert body["aircraft"] == []
+    assert body["vessels"] == []
+    assert body["fires"] == []
+    blob = "\n".join(sqls).lower()
+    assert "st_intersects" in blob
+    assert "aircraft_tracks_1min" in blob
+    assert "vessel_tracks_1min" in blob
+    assert "from fires" in blob

tests/test_geofence_frontend.py

@@ -0,0 +1,56 @@
+"""Geofence layer panel: draw, watch, inbox, delete (HTML contract)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def test_geofence_panel_has_list_and_delete_hook():
+    assert 'id="gf-draw"' in HTML
+    assert 'id="gf-list"' in HTML
+    assert "function deleteGeofence" in HTML
+    assert "method: 'DELETE'" in HTML or 'method: "DELETE"' in HTML
+    assert "/api/geofences/" in HTML
+
+
+def test_load_geofences_renders_delete_controls():
+    js = HTML.split("async function loadGeofences", 1)[1].split(
+        "async function loadFireAircraftHits", 1
+    )[0]
+    assert "gf-list" in js
+    assert "deleteGeofence" in js
+    assert "onEachFeature" in js
+    assert "bindPopup" in js
+
+
+def test_finish_cancel_draw_controls():
+    assert 'id="gf-finish"' in HTML
+    assert 'id="gf-cancel"' in HTML
+    assert "function cancelGeofenceDraw" in HTML
+    assert "function onGfClose" in HTML
+
+
+def test_watch_geofences_ws_payload():
+    assert "watch_geofences" in HTML
+    assert "function sendWatchGeofences" in HTML
+
+
+def test_geofence_alert_inbox():
+    assert 'id="gf-inbox"' in HTML
+    assert "/api/geofence-alerts" in HTML
+    assert "function loadGfInbox" in HTML
+    assert "function pushGfInbox" in HTML
+
+
+def test_delete_geofence_still_present():
+    assert "function deleteGeofence" in HTML
+    assert "method: 'DELETE'" in HTML or 'method: "DELETE"' in HTML
+
+
+def test_fence_dvr_at_endpoint():
+    assert "/at?timestamp=" in HTML or "/at?timestamp=${" in HTML
+    assert "function dvrScrubFence" in HTML
+    assert "gfSelectedId" in HTML

tests/test_hud_chrome.py

@@ -0,0 +1,113 @@
+"""Quiet HUD chrome: VIIRS default, collapsed rail, no Orbitron/MKT dashes."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def _attr(html: str, elem_id: str) -> str:
+    chunk = html.split(f'id="{elem_id}"', 1)[1].split(">", 1)[0]
+    return chunk
+
+
+def test_initmap_prefers_viirs_true_color():
+    init = HTML.split("async function initMap", 1)[1].split("function readMapPrefs", 1)[0]
+    assert "VIIRS_SNPP_CorrectedReflectance_TrueColor" in init
+    assert init.index("VIIRS_SNPP_CorrectedReflectance_TrueColor") < init.index(
+        "MODIS_Terra_CorrectedReflectance_TrueColor"
+    )
+    assert init.index("MODIS_Terra_CorrectedReflectance_TrueColor") < init.index(
+        "BlueMarble_ShadedRelief_Bathymetry"
+    )
+
+
+def test_orbitron_gone():
+    assert "Orbitron" not in HTML
+    assert "IBM Plex Sans" in HTML
+    assert "IBM Plex Mono" in HTML
+
+
+def test_lp_note_stripped_from_layer_list():
+    assert 'class="lp-note"' not in HTML
+    body = HTML.split('class="lp-body"', 1)[1].split("lp-legend", 1)[0]
+    assert "lp-note" not in body
+
+
+def test_default_overlays_basemap_and_firms_only():
+    fires = _attr(HTML, "lp-fires-on")
+    assert "checked" in fires
+    for eid in (
+        "lp-cams-on",
+        "lp-blips-on",
+        "lp-news-on",
+        "lp-radar-on",
+        "lp-alerts-on",
+        "lp-perim-on",
+        "lp-ac-on",
+        "lp-trains-on",
+        "lp-storms-on",
+    ):
+        assert "checked" not in _attr(HTML, eid), eid
+
+
+def test_geofence_markup_before_cameras():
+    assert 'id="gf-draw"' in HTML
+    assert HTML.index('id="gf-draw"') < HTML.index('id="lp-cams-on"')
+    assert HTML.index('id="lp-base-on"') < HTML.index('id="gf-draw"')
+
+
+def test_parent_geofence_hud_survives():
+    assert "watch_geofences" in HTML
+    assert "function deleteGeofence" in HTML
+    assert 'id="gf-finish"' in HTML
+    assert 'id="gf-cancel"' in HTML
+    assert 'id="gf-inbox"' in HTML
+
+
+def test_layer_rail_collapsed_on_load():
+    head = HTML.split('class="lp-head"', 1)[1].split("</div>", 1)[0]
+    assert 'aria-expanded="false"' in head
+    assert 'id="layer-panel" class="collapsed"' in HTML
+
+
+def test_market_ticker_hidden_no_poll():
+    mkt = HTML.split('class="ticker market"', 1)[1].split(">", 1)[0]
+    assert "hidden" in mkt
+    assert "setInterval(probeMarket" not in HTML
+    assert "setInterval(loadMarket" not in HTML
+    init = HTML.split("function initMarketTicker", 1)[1].split("function ", 1)[0]
+    assert "/api/market" in init or "404-poll" in init
+    assert "setInterval" not in init
+
+
+def test_news_ticker_fills_news_only_dock():
+    css = HTML.split("</style>", 1)[0]
+    compact = css.replace(" ", "").replace("\n", "")
+    assert ".dock.news-only{height:32px;}" in compact
+    assert ".dock.news-only.ticker{height:100%;}" in compact
+    assert ".ticker{display:flex;align-items:stretch;height:50%;" in compact
+
+
+def test_news_pins_are_circle_markers():
+    js = HTML.split("async function loadNewsPins", 1)[1].split("function refreshLiveOverlays", 1)[0]
+    assert "L.circleMarker" in js
+    assert "fillOpacity: 0.7" in js or "fillOpacity:0.7" in js
+    assert "rotate(45deg)" not in js
+    assert "L.divIcon" not in js
+
+
+def test_chokepoint_buttons_not_in_toolbar_flow():
+    assert 'id="chokepoint-select"' in HTML
+    css = HTML.split("</style>", 1)[0]
+    assert ".chokepoint-btns { display: none; }" in css or ".chokepoint-btns{display:none" in css.replace(
+        " ", ""
+    )
+
+
+def test_brand_is_osint_slash():
+    assert "GLOBAL SITUATIONAL AWARENESS TERMINAL" not in HTML
+    assert "OSINT" in HTML
+    assert 'class="accent">//</span>' in HTML

tests/test_hud_osiris.py

@@ -0,0 +1,90 @@
+"""HUD: layer-rail stats, shortcuts, terminator, zoom-gated cams, SWPC chip."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def _fn(name: str, nxt: str | None = None) -> str:
+    start = HTML.index(f"function {name}")
+    if nxt:
+        return HTML[start : HTML.index(f"function {nxt}", start + 1)]
+    return HTML[start : start + 4000]
+
+
+def test_stats_poll_uses_api_then_falls_back():
+    assert "/api/stats" in HTML
+    assert "30000" in HTML.split("pollLayerStats")[1][:2500] or "STATS_POLL_MS" in HTML
+    poll = HTML.split("async function pollLayerStats")[1].split("async function ")[0]
+    assert "404" in poll
+    assert "catch" in poll
+    ids = HTML.split("STATS_COUNT_IDS")[1].split("};")[0]
+    assert "aircraft" in ids and "cameras" in ids and "fires" in ids and "vessels" in ids
+    # Overlay loaders still write array lengths when stats is down.
+    assert "setLayerCount('lp-fires-count'" in HTML or 'setLayerCount("lp-fires-count"' in HTML
+    assert "setLayerCount('lp-cams-count'" in HTML or 'setLayerCount("lp-cams-count"' in HTML
+    assert "setLayerCount('lp-ac-count'" in HTML or 'setLayerCount("lp-ac-count"' in HTML
+    assert "setLayerCount('lp-vessels-count'" in HTML or 'setLayerCount("lp-vessels-count"' in HTML
+
+
+def test_keyboard_shortcuts_do_not_steal_osiris_fs():
+    keys = HTML.split("function initHudKeys")[1].split("function ")[0]
+    assert "Escape" in keys
+    assert "cheat-sheet" in keys or "toggleCheatSheet" in keys
+    assert "mapResetView" in keys
+    assert "toggleLayerPanel" in keys or "closeLayerPanel" in keys
+    # Do not bind Osiris's conflicting F/S (flights vs fullscreen / search).
+    assert "e.key === 'f'" not in keys.lower()
+    assert "e.key === 's'" not in keys.lower()
+    assert "case 'f'" not in keys.lower()
+    assert "case 's'" not in keys.lower()
+    assert 'id="cheat-sheet"' in HTML
+    assert "?" in keys or "Shift" in keys
+
+
+def test_terminator_toggle_defaults_off():
+    assert 'id="lp-terminator-on"' in HTML
+    row = HTML.split('id="lp-terminator-on"')[0][-120:] + HTML.split('id="lp-terminator-on"')[1][:80]
+    assert "checked" not in row.split(">")[0]
+    assert "function toggleTerminator" in HTML
+    assert "subsolarPoint" in HTML or "terminator" in HTML.lower()
+
+
+def test_camera_thumbs_gated_at_zoom_12():
+    assert "CAM_THUMB_MIN_ZOOM" in HTML
+    assert "CAM_THUMB_MIN_ZOOM = 12" in HTML
+    thumb = _fn("camThumb", "camPopupHtml")
+    assert "camThumbsAllowed" in thumb or "CAM_THUMB_MIN_ZOOM" in thumb
+    assert "zoom in for preview" in HTML or "zoom for preview" in HTML
+    assert "preview unavailable" in HTML
+    # RTSP still proxy through snapshot; never emit rtsp hrefs.
+    src = _fn("camSourceLink", "youtubeId")
+    assert "rtsp://" in src
+    assert "href=" not in src.split("rtsp://")[1].split("return")[0] or "Never emit" in src
+    assert 'href="${esc(url)}"' in src or "href=\"${esc(url)}\"" in src
+    assert src.index("rtsp://") < src.index("href=")
+
+
+def test_swpc_chip_browser_direct_correct_urls():
+    assert 'id="swpc-chip"' in HTML
+    assert "services.swpc.noaa.gov/json/planetary_k_index_1m.json" in HTML
+    assert "services.swpc.noaa.gov/json/goes/primary/xray-flares-latest.json" in HTML
+    assert "services.swpc.noaa.gov/products/alerts.json" in HTML
+    assert "services.swpc.noaa.gov/json/alerts.json" not in HTML
+    sw = HTML.split("async function pollSwpc")[1].split("async function ")[0]
+    assert "hidden" in sw
+    assert "kp_index" in sw
+    assert "90000" in HTML or "SWPC_POLL_MS" in HTML
+
+
+def test_new_chrome_does_not_cover_mobile_layers_zoom():
+    mobile = HTML.split("@media (max-width: 820px)")[1].split("@media (prefers-reduced-motion")[0]
+    assert "#layer-panel" in mobile
+    assert ".leaflet-top.leaflet-right .leaflet-control-zoom" in mobile
+    assert 'id="cheat-sheet"' in HTML
+    cheat = HTML.split(".cheat-sheet")[1][:500]
+    assert "z-index" in cheat
+    assert "calc(100% - 96px)" in cheat or "96px" in cheat

tests/test_infrastructure.py

@@ -0,0 +1,145 @@
+"""GET /api/infrastructure — Overpass nuclear markers."""
+
+import asyncio
+
+import httpx
+
+from live_layers import (
+    normalize_infra_element,
+    overlay_catalog,
+    overpass_nuclear_to_markers,
+    _cache,
+)
+from main import app
+
+BASE = "http://test"
+
+OVERPASS = {
+    "version": 0.6,
+    "generator": "Overpass API",
+    "elements": [
+        {
+            "type": "node",
+            "id": 12345,
+            "lat": 44.0,
+            "lon": -1.5,
+            "tags": {"name": "Test NPP", "operator": "EDF", "plant:source": "nuclear"},
+        },
+        {
+            "type": "way",
+            "id": 67890,
+            "center": {"lat": 43.5, "lon": -1.25},
+            "tags": {"name": "Test Plant Way", "plant:source": "nuclear"},
+        },
+        {
+            "type": "relation",
+            "id": 999,
+            "center": {"lat": 43.0, "lon": -1.0},
+            "tags": {},
+        },
+    ],
+}
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+def test_normalize_node_to_marker():
+    m = normalize_infra_element(OVERPASS["elements"][0], "nuclear")
+    assert m["id"] == "node/12345"
+    assert m["name"] == "Test NPP"
+    assert m["lat"] == 44.0
+    assert m["lon"] == -1.5
+    assert m["type"] == "nuclear"
+    assert m["extra"]["operator"] == "EDF"
+    assert "name" not in m["extra"]
+
+
+def test_way_center_and_unnamed_fallback():
+    way = normalize_infra_element(OVERPASS["elements"][1], "nuclear")
+    assert way["lat"] == 43.5
+    assert way["lon"] == -1.25
+    rel = normalize_infra_element(OVERPASS["elements"][2], "nuclear")
+    assert rel["name"] == "relation/999"
+
+
+def test_overpass_json_to_markers():
+    markers = overpass_nuclear_to_markers(OVERPASS)
+    assert len(markers) == 3
+    assert markers[0]["id"] == "node/12345"
+
+
+def test_missing_bbox_400():
+    resp = asyncio.run(_get("/api/infrastructure?types=nuclear"))
+    assert resp.status_code == 400
+
+
+def test_unknown_type_422():
+    resp = asyncio.run(_get("/api/infrastructure?types=military&bbox=-2,43,-1,44"))
+    assert resp.status_code == 422
+
+
+def test_map_infrastructure_returns_markers(monkeypatch):
+    async def fake_fetch(types, bbox):
+        return [
+            {"id": "node/1", "name": "X", "lat": 1.0, "lon": 2.0,
+             "type": "nuclear", "extra": {}}
+        ]
+
+    monkeypatch.setattr("main.fetch_infrastructure", fake_fetch)
+    resp = asyncio.run(_get("/api/infrastructure?types=nuclear&bbox=-2,43,-1,44"))
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body[0]["name"] == "X"
+    assert body[0]["type"] == "nuclear"
+    assert "max-age" in (resp.headers.get("cache-control") or "").lower()
+
+
+def test_overlay_catalog_has_infra_nuclear():
+    entry = overlay_catalog()["infra_nuclear"]
+    assert entry["kind"] == "points"
+    assert "nuclear" in entry["endpoint"]
+
+
+def test_fetch_infrastructure_cache_hit_no_refetch(monkeypatch):
+    _cache.clear()
+    hits = {"n": 0}
+
+    class FakeResp:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return OVERPASS
+
+    class FakeClient:
+        def __init__(self, **kw):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def post(self, url, data=None, timeout=None):
+            hits["n"] += 1
+            assert "overpass-api.de" in url
+            assert "plant:source" in data["data"]
+            assert "nuclear" in data["data"]
+            return FakeResp()
+
+    monkeypatch.setattr("live_layers.httpx.AsyncClient", FakeClient)
+    monkeypatch.setattr("live_layers._http", None)
+
+    from live_layers import fetch_infrastructure
+
+    m1 = asyncio.run(fetch_infrastructure("nuclear", "-2,43,-1,44"))
+    m2 = asyncio.run(fetch_infrastructure("nuclear", "-2,43,-1,44"))
+    assert len(m1) == 3
+    assert m2 == m1
+    assert hits["n"] == 1
+    _cache.clear()

tests/test_ingest_ssrf_and_sources.py

@@ -0,0 +1,67 @@
+"""SSRF guard on ingest triggers + PATCH /api/sources allowlist."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+from pydantic import ValidationError
+
+from main import app
+
+BASE = "http://test"
+LINK_LOCAL_META = "http://169.254.169.254/latest/meta-data/"
+LOOPBACK = "http://127.0.0.1/secret"
+
+
+async def _req(method: str, path: str, **kw) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.request(method, path, **kw)
+
+
+def test_rss_ingest_rejects_link_local_metadata_url(monkeypatch):
+    called = {"n": 0}
+
+    async def _boom(*_a, **_k):
+        called["n"] += 1
+        raise AssertionError("ingest_rss_feed must not run for a private URL")
+
+    monkeypatch.setattr("main.ingest_rss_feed", _boom)
+    resp = asyncio.run(_req("POST", "/api/ingest/rss", params={"feed_url": LINK_LOCAL_META}))
+    assert resp.status_code == 400
+    assert called["n"] == 0
+
+
+def test_gdelt_ingest_rejects_private_query_url(monkeypatch):
+    called = {"n": 0}
+
+    async def _boom(*_a, **_k):
+        called["n"] += 1
+        raise AssertionError("ingest_gdelt must not run for a private URL query")
+
+    monkeypatch.setattr("main.ingest_gdelt", _boom)
+    resp = asyncio.run(_req("POST", "/api/ingest/gdelt", params={"query": LOOPBACK}))
+    assert resp.status_code == 400
+    assert called["n"] == 0
+
+
+def test_update_source_rejects_unknown_fields():
+    sid = "00000000-0000-0000-0000-000000000001"
+    resp = asyncio.run(_req("PATCH", f"/api/sources/{sid}", json={"enabled": True, "source_type": "rss"}))
+    assert resp.status_code == 422
+
+
+def test_feed_source_update_allowlist_only():
+    from schemas import FeedSourceUpdate
+
+    payload = FeedSourceUpdate(name="n", url="https://example.com/rss", config={"k": 1}, enabled=False)
+    assert payload.model_dump(exclude_unset=True) == {
+        "name": "n",
+        "url": "https://example.com/rss",
+        "config": {"k": 1},
+        "enabled": False,
+    }
+    with pytest.raises(ValidationError):
+        FeedSourceUpdate.model_validate({"enabled": True, "id": "00000000-0000-0000-0000-000000000001"})

tests/test_live_layers.py

@@ -1,5 +1,7 @@
 """Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans)."""
 
+import json
+
 from live_layers import (
     MARKER_FIELDS,
     bbox_center_radius_nm,
@@ -7,6 +9,7 @@ from live_layers import (
     filter_points_bbox,
     parse_bbox,
     quantize_bbox,
+    pick_sentinel_feature,
     rainviewer_tile_url,
     sign_cog_url,
     sentinel1_tile_url,
@@ -24,7 +27,7 @@ from live_layers import (
     _wfigs_params,
 )
 
-from camera_scraper import parse_caltrans_json
+from camera_scraper import parse_caltrans_json, parse_udot_ibi_page, parse_odot_json, parse_mdot_json
 
 
 def test_parse_bbox_and_radius_clamps_to_150_nm():
@@ -256,6 +259,186 @@ def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
     assert "rtsp://" not in cam["snapshot_url"].lower()
 
 
+# ── UDOT IBI 511 parser ──────────────────────────────────────────────────
+
+def _udot_row(cam_id, lng, lat, **img_overrides):
+    img = {
+        "id": cam_id, "cameraSiteId": cam_id,
+        "imageUrl": f"/map/Cctv/{cam_id}", "disabled": False, "blocked": False,
+    }
+    img.update(img_overrides)
+    return {
+        "id": cam_id, "sourceId": "102771", "source": "ADX",
+        "roadway": "Unknown", "direction": "Unknown",
+        "location": "Freedom Blvd / 200 W @ 1100 N, PVO",
+        "latLng": {"geography": {
+            "coordinateSystemId": 4326,
+            "wellKnownText": f"POINT ({lng} {lat})"}},
+        "images": [img],
+    }
+
+
+def _udot_page(rows):
+    import json
+    return json.dumps({"draw": 0, "recordsTotal": len(rows),
+                       "recordsFiltered": len(rows), "data": rows})
+
+
+def test_parse_udot_wkt_maps_lng_lat():
+    cams = parse_udot_ibi_page(_udot_page([_udot_row(112731, -111.66204, 40.24863)]))
+    assert len(cams) == 1
+    cam = cams[0]
+    # WKT is `POINT (lng lat)` — order must not be swapped.
+    assert cam["location_lat"] == 40.24863
+    assert cam["location_lon"] == -111.66204
+    assert cam["discovery_source"] == "udot"
+    assert cam["vendor"] == "UDOT"
+    assert cam["source_url"] == "https://prod-ut.ibi511.com/map/Cctv/112731"
+    assert cam["snapshot_url"] == cam["source_url"]
+    assert "rtsp://" not in cam["source_url"].lower()
+    assert cam["raw"]["udot_id"] == 112731
+
+
+def test_parse_udot_skips_blocked_and_disabled():
+    rows = [
+        _udot_row(1, -111.0, 40.0),
+        _udot_row(2, -111.1, 40.1, blocked=True),
+        _udot_row(3, -111.2, 40.2, disabled=True),
+    ]
+    rows.append(_udot_row(4, -111.3, 40.3))
+    rows[3]["images"] = []  # no images → drop
+    cams = parse_udot_ibi_page(_udot_page(rows))
+    assert [c["raw"]["udot_id"] for c in cams] == [1]
+
+
+def test_parse_udot_drops_out_of_bbox():
+    rows = [
+        _udot_row(1, -111.0, 40.0),   # inside Utah
+        _udot_row(2, -100.0, 40.0),   # east of -108.9
+        _udot_row(3, -120.0, 40.0),   # west of -114.2
+        _udot_row(4, -111.0, 44.0),   # north of 42.1
+        _udot_row(5, -111.0, 30.0),   # south of 36.9
+    ]
+    cams = parse_udot_ibi_page(_udot_page(rows))
+    assert [c["raw"]["udot_id"] for c in cams] == [1]
+
+
+def test_parse_udot_bad_payload_returns_empty():
+    import json
+    assert parse_udot_ibi_page("not json") == []
+    assert parse_udot_ibi_page(json.dumps({"data": None})) == []
+    assert parse_udot_ibi_page(json.dumps({"data": "nope"})) == []
+
+
+def test_parse_udot_missing_wkt_skipped():
+    row = _udot_row(1, -111.0, 40.0)
+    row["latLng"] = {}
+    assert parse_udot_ibi_page(_udot_page([row])) == []
+
+
+def test_parse_odot_tripcheck_keeps_valid_skips_missing_and_oob():
+    payload = """
+    {"features":[
+      {"attributes":{
+        "cameraId":277,"filename":"AstoriaUS101_pid392.jpg",
+        "latitude":46.18785,"longitude":-123.85347,
+        "route":"US101 ","title":"US101 at Astoria"
+      }},
+      {"attributes":{
+        "cameraId":200,"filename":"","latitude":45.0,"longitude":-122.0,
+        "route":"I-5","title":"missing filename"
+      }},
+      {"attributes":{
+        "cameraId":300,"filename":"nocal_pid1.jpg",
+        "latitude":40.0,"longitude":-122.0,
+        "route":"US97","title":"out of bbox"
+      }},
+      {"attributes":{
+        "cameraId":400,"filename":"badcoord_pid2.jpg",
+        "latitude":null,"longitude":-122.0,
+        "route":"OR22","title":"null coord"
+      }}
+    ]}
+    """
+    cams = parse_odot_json(payload, "www.tripcheck.com")
+    assert len(cams) == 1
+    cam = cams[0]
+    assert cam["discovery_source"] == "odot"
+    assert cam["snapshot_url"] == (
+        "https://tripcheck.com/RoadCams/cams/AstoriaUS101_pid392.jpg")
+    assert cam["source_url"] == cam["snapshot_url"]
+    assert cam["location_lat"] == 46.18785
+    assert cam["location_lon"] == -123.85347
+    assert "US101 at Astoria" in cam["location_name"]
+    assert cam["vendor"] == "ODOT"
+    assert cam["device_type"] == "http"
+    assert "rtsp://" not in cam["snapshot_url"].lower()
+
+
+def test_parse_odot_tripcheck_handles_malformed():
+    assert parse_odot_json("not json", "www.tripcheck.com") == []
+    assert parse_odot_json('{"features":null}', "www.tripcheck.com") == []
+
+
+def test_parse_mdot_extracts_html_fields_and_bbox_filters():
+    rows = [
+        # In-bbox, full fields.
+        {
+            "route": "11 Mile",
+            "county": 'Wayne County <a href="/MiDrive/map?cameras=true&lat=42.491304&lon=-83.04479&zoom=15&id=1129"target="_blank">Go to</a>',
+            "location": " @ Mound NB",
+            "direction": "Traffic closest to camera is traveling north.",
+            "image": '<img alt="x" class="cameraImageForActivePane" id="1129Img" src="https://micamerasimages.net/thumbs/semtoc_cam_253.flv.jpg?item=1" height="170" width="250" onerror="cameraImageBroken(this)">',
+        },
+        # Out of bbox (lat 50) → drop.
+        {
+            "route": "Far",
+            "county": 'Nowhere <a href="/MiDrive/map?lat=50.0&lon=-83.0&zoom=15&id=9999">Go to</a>',
+            "location": "",
+            "image": '<img src="https://micamerasimages.net/thumbs/x.jpg">',
+        },
+        # Missing coordinates → drop.
+        {
+            "route": "NoCoords",
+            "county": 'Somewhere <a href="/MiDrive/map?zoom=15&id=8888">Go to</a>',
+            "location": "",
+            "image": '<img src="https://micamerasimages.net/thumbs/y.jpg">',
+        },
+        # Missing image → drop.
+        {
+            "route": "NoImage",
+            "county": 'Kent <a href="/MiDrive/map?lat=42.8841&lon=-85.6646&zoom=15&id=2113">Go to</a>',
+            "location": " @ Division",
+            "image": "",
+        },
+        # RTSP image src → drop.
+        {
+            "route": "Rtsp",
+            "county": 'Wayne <a href="/MiDrive/map?lat=42.4&lon=-83.1&zoom=15&id=1234">Go to</a>',
+            "location": "",
+            "image": '<img src="rtsp://10.0.0.1/stream">',
+        },
+    ]
+    cams = parse_mdot_json(json.dumps(rows), "mdotjboss.state.mi.us")
+    assert len(cams) == 1
+    cam = cams[0]
+    assert cam["discovery_source"] == "mdot"
+    assert cam["location_lat"] == 42.491304
+    assert cam["location_lon"] == -83.04479
+    assert cam["snapshot_url"] == "https://micamerasimages.net/thumbs/semtoc_cam_253.flv.jpg?item=1"
+    assert cam["source_url"] == "https://mdotjboss.state.mi.us/MiDrive/camera/1129"
+    assert cam["device_type"] == "http"
+    assert cam["vendor"] == "MDOT"
+    assert "11 Mile @ Mound NB" in cam["location_name"]
+    assert "Wayne County" in cam["location_name"]
+
+
+def test_parse_mdot_handles_malformed_payload():
+    assert parse_mdot_json("not json", "mdot") == []
+    assert parse_mdot_json('{"not": "a list"}', "mdot") == []
+    assert parse_mdot_json("[]", "mdot") == []
+
+
 def test_quantize_bbox_stable_under_jitter():
     a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102"))
     b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))
@@ -734,8 +917,9 @@ def test_fetch_sentinel1_vv_signed_tile_url(monkeypatch):
     post_url, post_json = calls[0][1], calls[0][2]
     assert post_url.endswith("/api/stac/v1/search")
     assert post_json["collections"] == ["sentinel-1-grd"]
-    assert post_json["limit"] == 1
+    assert post_json["limit"] >= 1
     assert post_json["sortby"][0]["direction"] == "desc"
+    assert "bbox" in out
 
 
 def test_fetch_sentinel1_uses_hh_when_vv_missing(monkeypatch):
@@ -783,3 +967,22 @@ def test_fetch_sentinel1_none_when_no_vv_or_hh(monkeypatch):
     _cache.clear()
 
     assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None
+
+
+def test_pick_sentinel_feature_prefers_scene_covering_center():
+    features = [
+        {"id": "far", "bbox": [10.0, 10.0, 12.0, 12.0]},
+        {"id": "cover", "bbox": [-80.5, 34.5, -78.5, 36.5]},
+        {"id": "also-far", "bbox": [-10.0, 0.0, -8.0, 2.0]},
+    ]
+    picked = pick_sentinel_feature(features, -79.5, 35.5)
+    assert picked["id"] == "cover"
+
+
+def test_pick_sentinel_feature_falls_back_to_first_when_none_cover():
+    features = [
+        {"id": "a", "bbox": [10.0, 10.0, 12.0, 12.0]},
+        {"id": "b", "bbox": [20.0, 20.0, 22.0, 22.0]},
+    ]
+    assert pick_sentinel_feature(features, -79.5, 35.5)["id"] == "a"
+    assert pick_sentinel_feature([], -79.5, 35.5) is None

tests/test_phase1_guardrails.py

@@ -31,3 +31,35 @@ def test_no_redis_kafka_celery():
         assert "kafka" not in blob
         assert "celery" not in blob
     assert "cachetools" in req
+
+
+def test_titiler_image_pinned_by_digest():
+    text = (ROOT / "docker-compose.yml").read_text()
+    assert (
+        "ghcr.io/developmentseed/titiler:latest@sha256:"
+        "1809958d063543e3ec858259536002b2de78e9f8f09a22a8d9591bdc2b550b14"
+        in text
+    )
+    # Unpinned :latest would drift on every pull.
+    for line in text.splitlines():
+        if "titiler" in line.lower() and "image:" in line:
+            assert "@sha256:" in line
+
+
+def test_uvicorn_single_worker_guard():
+    text = (ROOT / "app" / "main.py").read_text()
+    main_block = text.split('if __name__ == "__main__":', 1)[1]
+    assert "workers=1" in main_block
+
+
+def test_bbox_gist_migration_keeps_btree_and_adds_gist():
+    text = (ROOT / "alembic" / "versions" / "010_bbox_gist.py").read_text()
+    assert "down_revision" in text and "009_vessels" in text
+    assert "ix_events_geom_gist" in text
+    assert "ix_fires_geom_gist" in text
+    assert "ST_MakePoint(location_lon, location_lat)" in text
+    assert "ST_MakePoint(longitude, latitude)" in text
+    assert "USING gist" in text
+    models = (ROOT / "app" / "models.py").read_text()
+    assert 'Index("ix_events_location"' in models
+    assert 'Index("ix_fires_bbox"' in models

tests/test_phase2_api.py

@@ -17,6 +17,9 @@ async def _req(method: str, path: str, **kw) -> httpx.Response:
         return await client.request(method, path, **kw)
 
 
+FENCE_ID = "11111111-1111-1111-1111-111111111111"
+
+
 def test_geofence_post_rejects_point():
     resp = asyncio.run(_req(
         "POST", "/api/geofences",
@@ -25,6 +28,111 @@ def test_geofence_post_rejects_point():
     assert resp.status_code == 422
 
 
+def test_delete_geofence_404_when_missing(monkeypatch):
+    async def missing(_gid: str) -> bool:
+        return False
+
+    monkeypatch.setattr("geofence.delete_geofence", missing)
+    resp = asyncio.run(_req("DELETE", f"/api/geofences/{FENCE_ID}"))
+    assert resp.status_code == 404
+
+
+def test_geofence_alerts_passes_filters(monkeypatch):
+    seen = {}
+
+    async def fake_list(**kwargs):
+        seen.update(kwargs)
+        return [{"id": "a", "geofence_id": FENCE_ID, "source_kind": "ais"}]
+
+    monkeypatch.setattr("geofence.list_alerts", fake_list)
+    resp = asyncio.run(_req(
+        "GET", "/api/geofence-alerts",
+        params={
+            "geofence_id": FENCE_ID,
+            "since": "2026-08-28T00:00:00Z",
+            "until": "2026-08-29T00:00:00Z",
+            "source_kind": "ais",
+            "limit": 10,
+        },
+    ))
+    assert resp.status_code == 200
+    assert resp.json()[0]["source_kind"] == "ais"
+    assert seen["geofence_id"] == FENCE_ID
+    assert seen["source_kind"] == "ais"
+    assert seen["limit"] == 10
+    assert seen["since"] is not None
+    assert seen["until"] is not None
+
+
+def test_geofence_alerts_rejects_bad_source_kind():
+    resp = asyncio.run(_req(
+        "GET", "/api/geofence-alerts", params={"source_kind": "camera"},
+    ))
+    assert resp.status_code == 422
+
+
+def test_geofence_at_404_when_missing(monkeypatch):
+    async def no_snap(gid: str, ts):
+        return None
+
+    monkeypatch.setattr("geofence.snapshot_at", no_snap)
+    resp = asyncio.run(_req(
+        "GET", f"/api/geofences/{FENCE_ID}/at",
+        params={"timestamp": "2026-08-28T12:04:00Z"},
+    ))
+    assert resp.status_code == 404
+
+
+def test_geofence_at_empty_lists_when_db_down(monkeypatch):
+    async def empty_snap(gid: str, ts):
+        return {
+            "geofence_id": gid,
+            "timestamp": ts.isoformat(),
+            "aircraft": [],
+            "vessels": [],
+            "fires": [],
+        }
+
+    monkeypatch.setattr("geofence.snapshot_at", empty_snap)
+    resp = asyncio.run(_req(
+        "GET", f"/api/geofences/{FENCE_ID}/at",
+        params={"timestamp": "2026-08-28T12:04:00Z"},
+    ))
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["geofence_id"] == FENCE_ID
+    assert body["aircraft"] == []
+    assert body["vessels"] == []
+    assert body["fires"] == []
+    assert "timestamp" in body
+
+
+def test_geofence_at_does_not_notify(monkeypatch):
+    called = {"notify": 0}
+
+    async def empty_snap(gid: str, ts):
+        return {
+            "geofence_id": gid,
+            "timestamp": ts.isoformat(),
+            "aircraft": [],
+            "vessels": [],
+            "fires": [],
+        }
+
+    async def boom(**_kw):
+        called["notify"] += 1
+        raise AssertionError("GET /at must not record_and_notify")
+
+    monkeypatch.setattr("geofence.snapshot_at", empty_snap)
+    monkeypatch.setattr("geofence.record_and_notify", boom)
+    resp = asyncio.run(_req(
+        "GET", f"/api/geofences/{FENCE_ID}/at",
+        params={"timestamp": "2026-08-28T12:04:00Z"},
+    ))
+    assert resp.status_code == 200
+    assert called["notify"] == 0
+
+
 def test_geofences_list_does_not_collide_with_alerts():
     resp = asyncio.run(_req("GET", "/api/geofences"))
     assert resp.status_code == 200

tests/test_place_dossier_frontend.py

@@ -0,0 +1,59 @@
+"""Right-click place dossier HUD contract."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+HTML = (ROOT / "app/static/index.html").read_text()
+
+
+def test_place_dossier_panel_markup():
+    assert 'id="place-dossier"' in HTML
+    assert "What’s here?" in HTML or "What's here?" in HTML
+    assert 'id="pd-nearby"' in HTML
+    assert 'id="pd-close"' in HTML
+    assert 'role="dialog"' in HTML
+
+
+def test_place_dossier_uses_backend_nominatim_proxy():
+    js = HTML.split("async function openPlaceDossier", 1)[1].split(
+        "/* ═══════════════ INITIAL LOAD", 1
+    )[0]
+    assert "/api/place?lat=" in js
+    assert "nominatim.openstreetmap.org" not in js
+    assert "/api/aircraft" not in js
+    assert "/api/vessels" not in js
+    assert "/api/cameras" not in js
+    assert "/api/fires" not in js
+    assert "/api/weather-alerts" not in js
+    assert "/api/infrastructure" not in js
+
+
+def test_place_dossier_scans_loaded_overlays_5km():
+    assert "const PLACE_PAD_KM = 5" in HTML
+    assert "function collectNearby" in HTML
+    assert "lastCams" in HTML
+    assert "lastAircraft" in HTML
+    assert "lastVessels" in HTML
+    assert "lastFires" in HTML
+    assert "lastAlerts" in HTML
+    assert "function haversineKm" in HTML
+
+
+def test_place_dossier_right_click_and_long_press():
+    assert "map.on('contextmenu'" in HTML
+    assert "function bindPlaceLongPress" in HTML
+    assert "function closePlaceDossier" in HTML
+    assert "Escape" in HTML.split("function initMap", 1)[1][:8000] or "Escape" in HTML.split(
+        "bindPlaceLongPress(map)", 1
+    )[0][-500:]
+
+
+def test_place_dossier_mobile_is_bottom_sheet():
+    mobile = HTML.split("@media (max-width: 820px)")[1].split(
+        "@media (prefers-reduced-motion"
+    )[0]
+    assert "#place-dossier" in mobile
+    assert "bottom: 56px" in mobile
+    assert "max-height: 36vh" in mobile

tests/test_satellites.py

@@ -0,0 +1,200 @@
+"""CelesTrak satellites overlay: GP JSON parser, 2h cache, groups, bbox."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import httpx
+
+import satellites
+from live_layers import _cache, overlay_catalog
+from main import app
+
+BASE = "http://test"
+
+# Two real CelesTrak GP JSON records (trimmed to the OMM fields sgp4 needs).
+ISS = {
+    "OBJECT_NAME": "ISS (ZARYA)", "OBJECT_ID": "1998-067A",
+    "EPOCH": "2026-08-31T11:11:23.184384", "MEAN_MOTION": 15.4894954,
+    "ECCENTRICITY": 0.00050456, "INCLINATION": 51.6314,
+    "RA_OF_ASC_NODE": 287.5025, "ARG_OF_PERICENTER": 92.8598,
+    "MEAN_ANOMALY": 267.2968, "EPHEMERIS_TYPE": 0,
+    "CLASSIFICATION_TYPE": "U", "NORAD_CAT_ID": 25544,
+    "ELEMENT_SET_NO": 999, "REV_AT_EPOCH": 58342,
+    "BSTAR": 9.9862358e-5, "MEAN_MOTION_DOT": 5.046e-5,
+    "MEAN_MOTION_DDOT": 0,
+}
+HST = {
+    "OBJECT_NAME": "HST", "OBJECT_ID": "1990-037B",
+    "EPOCH": "2026-08-31T11:11:23.184384", "MEAN_MOTION": 15.0865888,
+    "ECCENTRICITY": 0.0002426, "INCLINATION": 28.4697,
+    "RA_OF_ASC_NODE": 102.1854, "ARG_OF_PERICENTER": 152.8462,
+    "MEAN_ANOMALY": 207.2795, "EPHEMERIS_TYPE": 0,
+    "CLASSIFICATION_TYPE": "U", "NORAD_CAT_ID": 20580,
+    "ELEMENT_SET_NO": 999, "REV_AT_EPOCH": 12345,
+    "BSTAR": 2.9e-5, "MEAN_MOTION_DOT": 0.0,
+    "MEAN_MOTION_DDOT": 0,
+}
+FIXTURE = [ISS, HST]
+
+NOW = datetime(2026, 8, 31, 12, 0, 0, tzinfo=timezone.utc)
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+def test_propagate_gp_shape():
+    rows = satellites.propagate_gp(FIXTURE, "stations", NOW)
+    assert len(rows) == 2
+    by_id = {r["id"]: r for r in rows}
+    assert set(by_id) == {"25544", "20580"}
+    iss = by_id["25544"]
+    assert iss["name"] == "ISS (ZARYA)"
+    assert iss["group"] == "stations"
+    # ISS is in LEO: ~400 km, |lat| <= inclination 51.63, lon in range.
+    assert 300 < iss["alt_km"] < 500
+    assert -51.7 <= iss["lat"] <= 51.7
+    assert -180 <= iss["lon"] <= 180
+    for key in ("id", "name", "lat", "lon", "alt_km", "group"):
+        assert key in iss
+
+
+def test_propagate_gp_skips_malformed():
+    bad = [{"OBJECT_NAME": "x"}, None, 42, {"NORAD_CAT_ID": 1}]
+    assert satellites.propagate_gp(bad, "stations", NOW) == []
+
+
+def test_parse_groups_defaults_and_validation():
+    assert satellites.parse_groups("stations,weather") == ["stations", "weather"]
+    assert satellites.parse_groups("weather,gps-ops") == ["weather", "gps-ops"]
+    # starlink is allowed only when explicitly requested
+    assert satellites.parse_groups("starlink") == ["starlink"]
+    assert satellites.parse_groups("stations,stations") == ["stations"]
+    for bad in ("", None, "debris", "stations,active", "stations, weather, active"):
+        try:
+            satellites.parse_groups(bad)
+        except ValueError:
+            pass
+        else:
+            raise AssertionError(f"expected ValueError for {bad!r}")
+
+
+def test_overlay_catalog_has_satellites_stub():
+    entry = overlay_catalog()["satellites"]
+    assert entry["kind"] == "points"
+    assert entry["endpoint"] == "/api/satellites"
+    assert "CelesTrak" in entry["attribution"]
+
+
+def test_unknown_group_400():
+    resp = asyncio.run(_get("/api/satellites?groups=debris"))
+    assert resp.status_code == 400
+
+
+def test_default_groups_ok_with_fake_fetch(monkeypatch):
+    async def fake(groups, bbox=None, limit=2000):
+        return {"satellites": [], "source": "celestrak",
+                "tle_epoch": None, "timestamp": "t"}
+
+    monkeypatch.setattr("main.fetch_satellites", fake)
+    resp = asyncio.run(_get("/api/satellites"))
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["source"] == "celestrak"
+    assert "max-age" in (resp.headers.get("cache-control") or "").lower()
+
+
+def test_2h_cache_does_not_refetch(monkeypatch):
+    satellites._last_good.clear()
+    _cache.clear()
+    hits = {"n": 0}
+
+    class FakeResp:
+        def __init__(self, data):
+            self._data = data
+
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return self._data
+
+    class FakeClient:
+        def __init__(self):
+            pass
+
+        async def get(self, url, params=None, headers=None):
+            hits["n"] += 1
+            assert "celestrak.org/NORAD/elements/gp.php" in url
+            return FakeResp(FIXTURE)
+
+    monkeypatch.setattr("live_layers._http", FakeClient())
+
+    async def run():
+        p1 = await satellites.fetch_satellites(["stations"])
+        p2 = await satellites.fetch_satellites(["stations"])
+        return p1, p2
+
+    p1, p2 = asyncio.run(run())
+    assert len(p1["satellites"]) == 2
+    assert p1["tle_epoch"] == "2026-08-31T11:11:23.184384"
+    # Same element blob served from cache (no refetch), same ids/epochs.
+    assert [s["id"] for s in p2["satellites"]] == [s["id"] for s in p1["satellites"]]
+    assert p2["tle_epoch"] == p1["tle_epoch"]
+    assert hits["n"] == 1
+    _cache.clear()
+    satellites._last_good.clear()
+
+
+def test_bbox_culls_satellites(monkeypatch):
+    """bbox filtering in fetch_satellites, deterministic via fake propagation."""
+
+    async def fake_elements(group):
+        return [{"x": 1}], "2026-08-31T11:11:23.184384"
+
+    monkeypatch.setattr("satellites._group_elements", fake_elements)
+
+    def fake_propagate(elements, group, now):
+        return [
+            {"id": "a", "name": "A", "lat": 10.0, "lon": 20.0, "alt_km": 400.0, "group": group},
+            {"id": "b", "name": "B", "lat": 45.0, "lon": -70.0, "alt_km": 400.0, "group": group},
+            {"id": "c", "name": "C", "lat": -10.0, "lon": 30.0, "alt_km": 400.0, "group": group},
+        ]
+
+    monkeypatch.setattr("satellites.propagate_gp", fake_propagate)
+
+    payload = asyncio.run(
+        satellites.fetch_satellites(["stations"], bbox="-80,0,-60,50")
+    )
+    ids = [s["id"] for s in payload["satellites"]]
+    assert ids == ["b"]  # only (45, -70) falls inside the box
+
+
+def test_bbox_culls_nothing_when_empty():
+    from satellites import fetch_satellites
+
+    # No bbox: all rows returned up to limit.
+    # (skip network; just sanity-check the arg is accepted by signature)
+    assert callable(fetch_satellites)
+
+
+def test_satnogs_fallback_parser():
+    payload = [{
+        "tle0": "0 ISS (ZARYA)",
+        "tle1": "1 25544U 98067A   26243.85334329  .00004554  00000-0  90917-4 0  9992",
+        "tle2": "2 25544  51.6312 285.5873 0005057  94.2999 265.8567 15.48953200583481",
+        "norad_cat_id": 25544,
+        "updated": "2026-09-01T01:19:54.327653Z",
+    }]
+    rows, epoch = satellites.propagate_satnogs_tle(payload, "stations", NOW)
+    assert len(rows) == 1
+    row = rows[0]
+    assert row["id"] == "25544"
+    assert row["name"] == "ISS (ZARYA)"
+    assert row["group"] == "stations"
+    assert epoch == "2026-09-01T01:19:54.327653Z"
+    assert 300 < row["alt_km"] < 500

tests/test_sentinel1_frontend.py

@@ -39,3 +39,11 @@ def test_sentinel1_not_fetched_on_init_unless_on():
     assert "loadSentinel1()" not in init
     refresh = HTML.split("function refreshLiveOverlays", 1)[1].split("function addExtraAttrib", 1)[0]
     assert "if (sentinelOn) loadSentinel1();" in refresh
+
+
+def test_sentinel1_reuses_covering_scene_and_clips_tiles():
+    js = HTML.split("async function loadSentinel1", 1)[1].split("function loadThermal", 1)[0]
+    assert "sentinelStillCovers" in HTML
+    assert "itemId" in js
+    assert "L.latLngBounds" in js
+    assert "sentinelBounds" in HTML

tests/test_vesselapi.py

@@ -234,6 +234,8 @@ def _patch_side_effects(monkeypatch):
 
     monkeypatch.setattr("tracks.record_position", _noop)
     monkeypatch.setattr("geofence.record_and_notify", _noop)
+    monkeypatch.setattr(vesselapi, "persist_vessel_snapshot", _noop)
+    monkeypatch.setattr(vesselapi, "purge_old_vessels", _noop)
 
 
 def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch):

tests/test_vessels_snapshot.py

@@ -0,0 +1,36 @@
+"""VesselAPI daily snapshot store — as-of DVR + UTC-day purge (no DB)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from vesselapi import pick_poll_at, snapshot_as_of, utc_day_start
+
+
+def test_utc_day_start_floors_to_midnight_utc():
+    now = datetime(2026, 8, 29, 15, 30, 12, tzinfo=timezone.utc)
+    assert utc_day_start(now) == datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+
+
+def test_pick_poll_at_returns_latest_snapshot_at_or_before_as_of():
+    t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+    t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc)
+    t3 = datetime(2026, 8, 29, 9, 36, tzinfo=timezone.utc)
+    as_of = datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc)
+    assert pick_poll_at([t1, t2, t3], as_of) == t2
+    assert pick_poll_at([t1, t2, t3], t1) == t1
+    assert pick_poll_at([t1, t2, t3], datetime(2026, 8, 28, 23, tzinfo=timezone.utc)) is None
+
+
+def test_snapshot_as_of_returns_the_matching_poll_only():
+    t1 = datetime(2026, 8, 29, 0, 0, tzinfo=timezone.utc)
+    t2 = datetime(2026, 8, 29, 4, 48, tzinfo=timezone.utc)
+    rows = [
+        {"id": "1", "poll_at": t1, "lat": 26.5, "lon": 56.0},
+        {"id": "2", "poll_at": t1, "lat": 26.6, "lon": 56.1},
+        {"id": "1", "poll_at": t2, "lat": 26.7, "lon": 56.2},
+    ]
+    out = snapshot_as_of(rows, datetime(2026, 8, 29, 6, 0, tzinfo=timezone.utc))
+    assert {r["id"] for r in out} == {"1"}
+    assert out[0]["lat"] == 26.7
+    assert all(r["poll_at"] == t2 for r in out)