feat: lean-Pi WS fan-out, geofences, DVR playback, fire/aircraft correlation

Phase 1: in-memory ConnectionManager viewport fan-out, 500ms map debounce,
cachetools TTLCache, background masscan/ffmpeg, compose memory caps.

Phase 2: PostGIS geofences + ST_Intersects alerts, Timescale 1-min CAGGs
and timestamp playback, FIRMS/WFIGS x firefighting ADS-B within 20 miles.

No Redis/Kafka/Celery.

alembic/versions/005_phase2.py

@@ -0,0 +1,196 @@
+"""phase 2: geofences, 1-min track CAGGs, fire/aircraft hits
+
+Revision ID: 005_phase2
+Revises: 004_camera_enum
+Create Date: 2026-08-28
+"""
+
+from alembic import op
+
+revision = "005_phase2"
+down_revision = "004_camera_enum"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute("CREATE EXTENSION IF NOT EXISTS postgis")
+    op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb")
+
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS geofences (
+            id UUID PRIMARY KEY,
+            name TEXT NOT NULL,
+            geojson JSONB NOT NULL,
+            geom geometry(Polygon, 4326),
+            active INTEGER NOT NULL DEFAULT 1,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS ix_geofences_geom
+        ON geofences USING gist (geom)
+    """)
+
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS geofence_alerts (
+            id UUID PRIMARY KEY,
+            geofence_id UUID NOT NULL,
+            source_kind TEXT NOT NULL,
+            entity_id TEXT NOT NULL,
+            lat DOUBLE PRECISION,
+            lon DOUBLE PRECISION,
+            payload JSONB,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS ix_geofence_alerts_created
+        ON geofence_alerts (created_at DESC)
+    """)
+
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS vessel_positions (
+            mmsi TEXT NOT NULL,
+            ts 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, ts)
+        )
+    """)
+    op.execute("""
+        SELECT create_hypertable(
+            'vessel_positions', 'ts', if_not_exists => TRUE
+        )
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS ix_vessel_positions_bbox
+        ON vessel_positions (lon, lat)
+    """)
+
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS aircraft_positions (
+            hex TEXT NOT NULL,
+            ts 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 (hex, ts)
+        )
+    """)
+    op.execute("""
+        SELECT create_hypertable(
+            'aircraft_positions', 'ts', if_not_exists => TRUE
+        )
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS ix_aircraft_positions_bbox
+        ON aircraft_positions (lon, lat)
+    """)
+
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS fire_aircraft_hits (
+            id UUID NOT NULL,
+            fire_id TEXT NOT NULL,
+            fire_lat DOUBLE PRECISION NOT NULL,
+            fire_lon DOUBLE PRECISION NOT NULL,
+            aircraft_hex TEXT NOT NULL,
+            aircraft_type TEXT,
+            aircraft_lat DOUBLE PRECISION NOT NULL,
+            aircraft_lon DOUBLE PRECISION NOT NULL,
+            distance_mi DOUBLE PRECISION NOT NULL,
+            seen_at TIMESTAMPTZ NOT NULL,
+            PRIMARY KEY (fire_id, aircraft_hex, seen_at)
+        )
+    """)
+    op.execute("""
+        CREATE INDEX IF NOT EXISTS ix_fire_aircraft_hits_seen
+        ON fire_aircraft_hits (seen_at DESC)
+    """)
+
+    # 1-minute continuous aggregates (Timescale). last() keeps the newest
+    # sample in each bucket — the DVR slider reads these, not the raw table.
+    op.execute("""
+        DO $$
+        BEGIN
+            IF NOT EXISTS (
+                SELECT 1 FROM timescaledb_information.continuous_aggregates
+                WHERE view_name = 'vessel_tracks_1min'
+            ) THEN
+                EXECUTE $v$
+                    CREATE MATERIALIZED VIEW vessel_tracks_1min
+                    WITH (timescaledb.continuous) AS
+                    SELECT time_bucket('1 minute', ts) AS bucket,
+                           mmsi,
+                           last(lat, ts) AS lat,
+                           last(lon, ts) AS lon,
+                           last(heading, ts) AS heading,
+                           last(speed, ts) AS speed,
+                           last(label, ts) AS label
+                    FROM vessel_positions
+                    GROUP BY bucket, mmsi
+                    WITH NO DATA
+                $v$;
+            END IF;
+            IF NOT EXISTS (
+                SELECT 1 FROM timescaledb_information.continuous_aggregates
+                WHERE view_name = 'aircraft_tracks_1min'
+            ) THEN
+                EXECUTE $a$
+                    CREATE MATERIALIZED VIEW aircraft_tracks_1min
+                    WITH (timescaledb.continuous) AS
+                    SELECT time_bucket('1 minute', ts) AS bucket,
+                           hex,
+                           last(lat, ts) AS lat,
+                           last(lon, ts) AS lon,
+                           last(heading, ts) AS heading,
+                           last(speed, ts) AS speed,
+                           last(label, ts) AS label
+                    FROM aircraft_positions
+                    GROUP BY bucket, hex
+                    WITH NO DATA
+                $a$;
+            END IF;
+        END
+        $$;
+    """)
+    op.execute("""
+        DO $$
+        BEGIN
+            PERFORM add_continuous_aggregate_policy(
+                'vessel_tracks_1min',
+                start_offset => INTERVAL '3 hours',
+                end_offset => INTERVAL '1 minute',
+                schedule_interval => INTERVAL '1 minute',
+                if_not_exists => TRUE
+            );
+            PERFORM add_continuous_aggregate_policy(
+                'aircraft_tracks_1min',
+                start_offset => INTERVAL '3 hours',
+                end_offset => INTERVAL '1 minute',
+                schedule_interval => INTERVAL '1 minute',
+                if_not_exists => TRUE
+            );
+        EXCEPTION WHEN OTHERS THEN
+            NULL;
+        END
+        $$;
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP MATERIALIZED VIEW IF EXISTS aircraft_tracks_1min CASCADE")
+    op.execute("DROP MATERIALIZED VIEW IF EXISTS vessel_tracks_1min CASCADE")
+    op.execute("DROP TABLE IF EXISTS fire_aircraft_hits")
+    op.execute("DROP TABLE IF EXISTS aircraft_positions")
+    op.execute("DROP TABLE IF EXISTS vessel_positions")
+    op.execute("DROP TABLE IF EXISTS geofence_alerts")
+    op.execute("DROP TABLE IF EXISTS geofences")

app/bg_jobs.py

@@ -0,0 +1,103 @@
+"""Background masscan / ffmpeg — never block a FastAPI request on a scan.
+
+masscan is capped at 200 pps (home uplink saturates at 1k+). ffmpeg frame
+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)
+
+
+def schedule_ffmpeg_snapshot(url: str, timeout: float = 8.0) -> asyncio.Task:
+    """Start (or reuse) an ffmpeg JPEG grab. Caller may await the task."""
+    existing = _ffmpeg_tasks.get(url)
+    if existing is not None and not existing.done():
+        return existing
+    task = asyncio.create_task(_ffmpeg_grab_and_cache(url, timeout))
+    _ffmpeg_tasks[url] = task
+    return task
+
+
+async def _ffmpeg_grab(url: str, timeout: float = 8.0) -> bytes | None:
+    """Grab one JPEG frame. Isolated so tests can stub it."""
+    if not _FFMPEG or not url.lower().startswith("rtsp://"):
+        return None
+    cmd = [
+        _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
+        "-rtsp_transport", "tcp",
+        "-timeout", "4000000",
+        "-i", url,
+        "-frames:v", "1",
+        "-f", "image2pipe", "-vcodec", "mjpeg",
+        "pipe:1",
+    ]
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+    except FileNotFoundError:
+        return None
+    try:
+        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        proc.kill()
+        try:
+            await proc.wait()
+        except Exception:  # noqa: BLE001
+            pass
+        return None
+    if proc.returncode not in (0, None) or not stdout or len(stdout) < 64:
+        return None
+    if stdout[:2] != b"\xff\xd8":
+        return None
+    return stdout
+
+
+async def _ffmpeg_grab_and_cache(url: str, timeout: float) -> bytes | None:
+    data = await _ffmpeg_grab(url, timeout)
+    if data:
+        _ffmpeg_cache[url] = data
+    return data

app/camera_preview.py

@@ -137,38 +137,17 @@ async def probe_public_feed(host: str) -> str | 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."""
+    """Grab a single JPEG frame from an RTSP URL. None if ffmpeg missing/fails.
-    if not _FFMPEG or not url.lower().startswith("rtsp://"):
+
-        return None
+    The subprocess is scheduled via asyncio.create_task (shared per URL) so
-    cmd = [
+    concurrent popup clicks do not stack ffmpeg processes on the request path.
-        _FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
+    """
-        "-rtsp_transport", "tcp",
+    from bg_jobs import cached_ffmpeg_jpeg, schedule_ffmpeg_snapshot
-        "-timeout", "4000000",          # 4s socket timeout, microseconds
+
-        "-i", url,
+    hit = cached_ffmpeg_jpeg(url)
-        "-frames:v", "1",
+    if hit:
-        "-f", "image2pipe", "-vcodec", "mjpeg",
+        return hit
-        "pipe:1",
+    return await schedule_ffmpeg_snapshot(url, timeout)
-    ]
-    try:
-        proc = await asyncio.create_subprocess_exec(
-            *cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.DEVNULL,
-        )
-    except FileNotFoundError:
-        return None
-    try:
-        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout)
-    except asyncio.TimeoutError:
-        proc.kill()
-        try:
-            await proc.wait()
-        except Exception:  # noqa: BLE001
-            pass
-        return None
-    if proc.returncode not in (0, None) or not _looks_like_jpeg(stdout or b""):
-        return None
-    return stdout
 
 
 async def ffmpeg_mjpeg_stream(url: str):

app/fire_aircraft.py

@@ -0,0 +1,186 @@
+"""Flag firefighting ADS-B aircraft within 20 miles of an active fire."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+from sqlalchemy import text
+
+from database import async_session
+from live_layers import _haversine_km
+
+
+RADIUS_MILES = 20.0
+RADIUS_KM = RADIUS_MILES * 1.609344
+_recent: dict[tuple[str, str], datetime] = {}
+_COOLDOWN = timedelta(minutes=5)
+
+# ICAO type designators commonly used on wildfire air tankers, scoopers,
+# helitack, and air-attack platforms. Uppercase; compared case-insensitively.
+FIREFIGHTER_ICAO = frozenset({
+    "AT802", "AT8T", "AT8P", "AT8B",
+    "C130", "C30J", "C130J",
+    "DC10", "MD10", "MD11", "MD87",
+    "B737", "B38M",
+    "CL2T", "CL215", "CL415", "CL5T",
+    "S64", "SK64",
+    "UH1", "UH1Y", "UH60", "H60", "S70",
+    "B412", "B212", "B205",
+    "AS50", "AS350", "A119", "A109",
+    "B350", "BE20",
+    "OV10",
+    "RJ85", "RJ1H", "B461", "B462", "B463",
+    "PC12",
+    "TBM7", "TBM8", "TBM9",
+    "C208",
+    "DH8D", "Q400",
+})
+
+
+def _icao(marker: dict) -> str:
+    extra = marker.get("extra") or {}
+    return str(extra.get("type") or extra.get("t") or "").strip().upper()
+
+
+def is_firefighter(marker: dict) -> bool:
+    return _icao(marker) in FIREFIGHTER_ICAO
+
+
+def _haversine_mi(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    return _haversine_km(lat1, lon1, lat2, lon2) / 1.609344
+
+
+def correlate_aircraft_to_fires(
+    fires: list[dict],
+    aircraft: list[dict],
+    radius_mi: float = RADIUS_MILES,
+) -> list[dict]:
+    hits: list[dict] = []
+    for fire in fires:
+        flat, flon = fire.get("lat"), fire.get("lon")
+        if flat is None or flon is None:
+            continue
+        fid = str(fire.get("id") or fire.get("label") or "fire")
+        for ac in aircraft:
+            if not is_firefighter(ac):
+                continue
+            alat, alon = ac.get("lat"), ac.get("lon")
+            if alat is None or alon is None:
+                continue
+            dist = _haversine_mi(float(flat), float(flon), float(alat), float(alon))
+            if dist > radius_mi:
+                continue
+            hits.append({
+                "fire_id": fid,
+                "fire_lat": float(flat),
+                "fire_lon": float(flon),
+                "aircraft_hex": str(ac.get("id")),
+                "aircraft_type": _icao(ac),
+                "aircraft_lat": float(alat),
+                "aircraft_lon": float(alon),
+                "distance_mi": round(dist, 2),
+                "label": ac.get("label") or ac.get("id"),
+            })
+    return hits
+
+
+async def persist_hits(hits: list[dict], seen_at: datetime | None = None) -> int:
+    if not hits:
+        return 0
+    ts = seen_at or datetime.now(timezone.utc)
+    n = 0
+    async with async_session() as session:
+        for h in hits:
+            try:
+                await session.execute(
+                    text(
+                        """
+                        INSERT INTO fire_aircraft_hits
+                            (id, fire_id, fire_lat, fire_lon,
+                             aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
+                             distance_mi, seen_at)
+                        VALUES (
+                            CAST(:id AS uuid), :fire_id, :fire_lat, :fire_lon,
+                            :aircraft_hex, :aircraft_type, :aircraft_lat, :aircraft_lon,
+                            :distance_mi, :seen_at
+                        )
+                        ON CONFLICT (fire_id, aircraft_hex, seen_at) DO NOTHING
+                        """
+                    ),
+                    {
+                        "id": str(uuid4()),
+                        "fire_id": h["fire_id"],
+                        "fire_lat": h["fire_lat"],
+                        "fire_lon": h["fire_lon"],
+                        "aircraft_hex": h["aircraft_hex"],
+                        "aircraft_type": h["aircraft_type"],
+                        "aircraft_lat": h["aircraft_lat"],
+                        "aircraft_lon": h["aircraft_lon"],
+                        "distance_mi": h["distance_mi"],
+                        "seen_at": ts.replace(microsecond=0),
+                    },
+                )
+                n += 1
+            except Exception:
+                continue
+        try:
+            await session.commit()
+        except Exception:
+            return 0
+    return n
+
+
+async def recent_hits(limit: int = 200) -> list[dict]:
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                text(
+                    """
+                    SELECT fire_id, fire_lat, fire_lon,
+                           aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
+                           distance_mi, seen_at
+                    FROM fire_aircraft_hits
+                    ORDER BY seen_at DESC
+                    LIMIT :limit
+                    """
+                ),
+                {"limit": limit},
+            )).mappings().all()
+        out = []
+        for r in rows:
+            item = dict(r)
+            if item.get("seen_at") is not None:
+                item["seen_at"] = item["seen_at"].isoformat()
+            out.append(item)
+        return out
+    except Exception:
+        return []
+
+
+async def correlate_and_notify(fires: list[dict], aircraft: list[dict]) -> list[dict]:
+    hits = correlate_aircraft_to_fires(fires, aircraft)
+    if not hits:
+        return []
+    now = datetime.now(timezone.utc)
+    fresh = []
+    for h in hits:
+        key = (h["fire_id"], h["aircraft_hex"])
+        prev = _recent.get(key)
+        if prev is not None and now - prev < _COOLDOWN:
+            continue
+        _recent[key] = now
+        fresh.append(h)
+    if not fresh:
+        return []
+    await persist_hits(fresh, seen_at=now)
+    from ws_manager import manager
+    for h in fresh:
+        body = {**h, "seen_at": now.isoformat()}
+        await manager.publish_point(
+            "fire_aircraft", body, lat=h["aircraft_lat"], lon=h["aircraft_lon"],
+        )
+        await manager.publish_point(
+            "fire_aircraft", body, lat=h["fire_lat"], lon=h["fire_lon"],
+        )
+    return fresh

app/fire_sources.py

@@ -40,6 +40,7 @@ from config import (
     NATS_URL,
 )
 from keystore import get_api_key
+from upstream_cache import firms_cache
 
 logger = logging.getLogger("osint.firms")
 
@@ -185,12 +186,16 @@ async def ingest_fires(bbox: str | None = None) -> int:
     total_published = 0
     async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client:
         for dataset in datasets:
+            cache_key = (dataset, area, FIRMS_DAYS)
+            text = firms_cache.get(cache_key)
+            if text is None:
                 url = FIRMS_AREA_CSV.format(
                     key=map_key, dataset=dataset, bbox=area, days=FIRMS_DAYS
                 )
                 resp = await client.get(url)
                 resp.raise_for_status()
                 text = resp.text
+                firms_cache[cache_key] = text
             # FIRMS returns HTTP 200 with a plain-text error for some failure modes
             # (bad key, invalid bbox); surface the first line for debuggability.
             if "latitude" not in text.lower()[:4096]:

app/geofence.py

@@ -0,0 +1,321 @@
+"""Geofences: GeoJSON polygons, ST_Intersects on ingest, WS alerts.
+
+``/api/alerts`` is the dashboard entity/keyword table — geofence hits live
+in ``geofence_alerts`` and fan out as WS type ``geofence_alert``.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import text
+
+from database import async_session
+
+
+def _rings_from_geojson(geojson: dict) -> list[list[list[float]]]:
+    if not isinstance(geojson, dict):
+        raise ValueError("geojson must be an object")
+    gj = geojson
+    if gj.get("type") == "Feature":
+        gj = gj.get("geometry") or {}
+    if gj.get("type") == "FeatureCollection":
+        raise ValueError("FeatureCollection is not a single polygon")
+    if gj.get("type") != "Polygon":
+        raise ValueError("geojson must be a Polygon")
+    coords = gj.get("coordinates")
+    if not isinstance(coords, list) or not coords:
+        raise ValueError("polygon has no rings")
+    rings: list[list[list[float]]] = []
+    for ring in coords:
+        if not isinstance(ring, list) or len(ring) < 4:
+            raise ValueError("polygon ring needs ≥4 positions (closed)")
+        pts = []
+        for pt in ring:
+            if not isinstance(pt, (list, tuple)) or len(pt) < 2:
+                raise ValueError("position must be [lon, lat]")
+            pts.append([float(pt[0]), float(pt[1])])
+        rings.append(pts)
+    return rings
+
+
+def validate_polygon_geojson(geojson: dict) -> dict:
+    """Return a canonical Polygon GeoJSON or raise ValueError."""
+    rings = _rings_from_geojson(geojson)
+    return {"type": "Polygon", "coordinates": rings}
+
+
+def _ring_contains(lon: float, lat: float, ring: list[list[float]]) -> bool:
+    """Ray-cast even-odd rule. Ring is [lon, lat] positions."""
+    inside = False
+    n = len(ring)
+    if n < 4:
+        return False
+    j = n - 1
+    for i in range(n):
+        xi, yi = ring[i][0], ring[i][1]
+        xj, yj = ring[j][0], ring[j][1]
+        intersects = ((yi > lat) != (yj > lat)) and (
+            lon < (xj - xi) * (lat - yi) / ((yj - yi) or 1e-16) + xi
+        )
+        if intersects:
+            inside = not inside
+        j = i
+    return inside
+
+
+def point_in_geojson(lon: float, lat: float, geojson: dict) -> bool:
+    """True if (lon, lat) is inside the outer ring and outside holes."""
+    try:
+        rings = _rings_from_geojson(geojson)
+    except (ValueError, TypeError, KeyError):
+        return False
+    if not _ring_contains(lon, lat, rings[0]):
+        return False
+    for hole in rings[1:]:
+        if _ring_contains(lon, lat, hole):
+            return False
+    return True
+
+
+def matching_geofences(lon: float, lat: float, fences: list[dict]) -> list[dict]:
+    hits = []
+    for fence in fences:
+        if not fence.get("active", True):
+            continue
+        gj = fence.get("geojson") or {}
+        if point_in_geojson(lon, lat, gj):
+            hits.append(fence)
+    return hits
+
+
+# In-process copy of active fences so ingest does not round-trip Postgres
+# on every AIS frame. CRUD endpoints refresh this list.
+_cache: list[dict] = []
+_recent_hits: dict[tuple[str, str], datetime] = {}
+_HIT_COOLDOWN = timedelta(minutes=5)
+
+
+async def refresh_cache() -> list[dict]:
+    global _cache
+    async with async_session() as session:
+        rows = (await session.execute(text(
+            "SELECT id::text, name, geojson, active FROM geofences"
+        ))).mappings().all()
+    _cache = [
+        {
+            "id": r["id"],
+            "name": r["name"],
+            "geojson": r["geojson"] if isinstance(r["geojson"], dict)
+            else json.loads(r["geojson"] or "{}"),
+            "active": bool(r["active"]),
+        }
+        for r in rows
+    ]
+    return _cache
+
+
+def cached_fences() -> list[dict]:
+    return list(_cache)
+
+
+async def list_geofences() -> list[dict]:
+    if not _cache:
+        try:
+            await refresh_cache()
+        except Exception:
+            return []
+    return cached_fences()
+
+
+async def create_geofence(name: str, geojson: dict, active: bool = True) -> dict:
+    polygon = validate_polygon_geojson(geojson)
+    gid = str(uuid4())
+    gj = json.dumps(polygon)
+    async with async_session() as session:
+        await session.execute(
+            text(
+                """
+                INSERT INTO geofences (id, name, geojson, geom, active)
+                VALUES (
+                    :id, :name, CAST(:geojson AS jsonb),
+                    ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
+                    :active
+                )
+                """
+            ),
+            {"id": gid, "name": name, "geojson": gj, "active": 1 if active else 0},
+        )
+        await session.commit()
+    row = {"id": gid, "name": name, "geojson": polygon, "active": active}
+    _cache.append(row)
+    return row
+
+
+async def update_geofence(gid: str, *, name: str | None = None,
+                          geojson: dict | None = None,
+                          active: bool | None = None) -> dict | None:
+    current = next((f for f in _cache if f["id"] == gid), None)
+    if current is None:
+        await refresh_cache()
+        current = next((f for f in _cache if f["id"] == gid), None)
+    if current is None:
+        return None
+    if name is not None:
+        current["name"] = name
+    if geojson is not None:
+        current["geojson"] = validate_polygon_geojson(geojson)
+    if active is not None:
+        current["active"] = active
+    gj = json.dumps(current["geojson"])
+    async with async_session() as session:
+        await session.execute(
+            text(
+                """
+                UPDATE geofences SET
+                    name = :name,
+                    geojson = CAST(:geojson AS jsonb),
+                    geom = ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
+                    active = :active,
+                    updated_at = now()
+                WHERE id = CAST(:id AS uuid)
+                """
+            ),
+            {
+                "id": gid,
+                "name": current["name"],
+                "geojson": gj,
+                "active": 1 if current["active"] else 0,
+            },
+        )
+        await session.commit()
+    return current
+
+
+async def delete_geofence(gid: str) -> bool:
+    async with async_session() as session:
+        result = await session.execute(
+            text("DELETE FROM geofences WHERE id = CAST(:id AS uuid)"),
+            {"id": gid},
+        )
+        await session.commit()
+    _cache[:] = [f for f in _cache if f["id"] != gid]
+    return bool(result.rowcount)
+
+
+async def st_intersects(lon: float, lat: float) -> list[dict]:
+    """PostGIS ST_Intersects against active geofences.
+
+    Falls back to the in-memory GeoJSON test if the DB is unreachable so
+    ingest never dies because a fence check failed.
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                text(
+                    """
+                    SELECT id::text, name, geojson, active
+                    FROM geofences
+                    WHERE active = 1
+                      AND ST_Intersects(
+                            geom,
+                            ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)
+                          )
+                    """
+                ),
+                {"lon": lon, "lat": lat},
+            )).mappings().all()
+        return [
+            {
+                "id": r["id"],
+                "name": r["name"],
+                "geojson": r["geojson"] if isinstance(r["geojson"], dict)
+                else json.loads(r["geojson"] or "{}"),
+                "active": True,
+            }
+            for r in rows
+        ]
+    except Exception:
+        return matching_geofences(lon, lat, cached_fences())
+
+
+async def record_and_notify(
+    *,
+    source_kind: str,
+    entity_id: str,
+    lat: float,
+    lon: float,
+    payload: dict[str, Any] | None = None,
+) -> int:
+    """Insert a geofence_alerts row per hit and WS-push to viewport clients.
+
+    PostGIS ST_Intersects is the source of truth. The in-process GeoJSON
+    cache is not a reject filter — the FIRMS ingester never fills it.
+    """
+    hits = await st_intersects(lon, lat)
+    if not hits:
+        return 0
+    from ws_manager import manager
+
+    sent = 0
+    now = datetime.now(timezone.utc)
+    fresh = []
+    for fence in hits:
+        key = (str(fence["id"]), str(entity_id))
+        prev = _recent_hits.get(key)
+        if prev is not None and now - prev < _HIT_COOLDOWN:
+            continue
+        _recent_hits[key] = now
+        fresh.append(fence)
+    if not fresh:
+        return 0
+    hits = fresh
+    async with async_session() as session:
+        for fence in hits:
+            aid = str(uuid4())
+            body = {
+                "id": aid,
+                "geofence_id": fence["id"],
+                "geofence_name": fence.get("name"),
+                "source_kind": source_kind,
+                "entity_id": str(entity_id),
+                "lat": lat,
+                "lon": lon,
+                "payload": payload or {},
+                "created_at": now.isoformat(),
+            }
+            try:
+                await session.execute(
+                    text(
+                        """
+                        INSERT INTO geofence_alerts
+                            (id, geofence_id, source_kind, entity_id, lat, lon, payload)
+                        VALUES (
+                            CAST(:id AS uuid), CAST(:geofence_id AS uuid),
+                            :source_kind, :entity_id, :lat, :lon, CAST(:payload AS jsonb)
+                        )
+                        """
+                    ),
+                    {
+                        "id": aid,
+                        "geofence_id": fence["id"],
+                        "source_kind": source_kind,
+                        "entity_id": str(entity_id),
+                        "lat": lat,
+                        "lon": lon,
+                        "payload": json.dumps(payload or {}),
+                    },
+                )
+            except Exception:
+                pass
+            sent += await manager.publish_point(
+                "geofence_alert", body, lat=lat, lon=lon,
+            )
+        try:
+            await session.commit()
+        except Exception:
+            pass
+    return sent

app/ingestor.py

@@ -95,6 +95,23 @@ async def ingest_fire_row(msg: dict) -> bool:
             "ingested fire %.5f,%.5f %s satellite=%s",
             row["latitude"], row["longitude"], row["acq_time"], row["satellite"],
         )
+        lat, lon = row["latitude"], row["longitude"]
+        from geofence import record_and_notify
+        await record_and_notify(
+            source_kind="firms",
+            entity_id=f"{lat},{lon},{row['satellite']}",
+            lat=lat, lon=lon, payload={"satellite": row["satellite"]},
+        )
+        from live_layers import aircraft_last_known
+        from fire_aircraft import correlate_and_notify
+        from tracks import recent_markers
+        acs = list(aircraft_last_known.values()) or await recent_markers("aircraft")
+        if acs:
+            fire = {
+                "id": f"firms:{lat:.4f},{lon:.4f}",
+                "lat": lat, "lon": lon, "label": "FIRMS",
+            }
+            await correlate_and_notify([fire], acs)
     return inserted
 
 

app/live_layers.py

@@ -69,6 +69,9 @@ vessel_lock = asyncio.Lock()
 # Viewport-following accumulates vessels across every region visited in a
 # session — keep the in-memory store bounded (oldest entries evicted).
 _MAX_VESSELS = 6000
+# Last ADS-B snapshot + WFIGS points for fire↔tanker correlation.
+aircraft_last_known: dict[str, dict] = {}
+fire_last_known: list[dict] = []
 
 
 def overlay_catalog() -> dict:
@@ -576,6 +579,25 @@ async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]:
         return transform_adsb_lol(await _get_json(url))
 
     rows = await _ttl_get(cache_key, 8.0, _load)
+    from ws_manager import manager
+    from tracks import record_position
+    from geofence import record_and_notify
+    aircraft_last_known.clear()
+    for m in rows:
+        aircraft_last_known[str(m.get("id"))] = m
+        mlat, mlon = m.get("lat"), m.get("lon")
+        if mlat is None or mlon is None:
+            continue
+        await record_position("aircraft", m)
+        if manager.has_clients():
+            await manager.publish_point("adsb", m, lat=mlat, lon=mlon)
+        await record_and_notify(
+            source_kind="adsb", entity_id=str(m.get("id")),
+            lat=mlat, lon=mlon, payload=m,
+        )
+    if fire_last_known:
+        from fire_aircraft import correlate_and_notify
+        await correlate_and_notify(fire_last_known, rows)
     return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
 
 
@@ -613,7 +635,7 @@ async def upsert_vessel(marker: dict) -> None:
         label = marker.get("label")
         if not label or label == vid:
             label = prev.get("label") or vid
-        vessel_last_known[vid] = {
+        stored = {
             **to_marker(
                 vid, lat, lon,
                 heading=marker.get("heading") if marker.get("heading") is not None else prev.get("heading"),
@@ -623,6 +645,7 @@ async def upsert_vessel(marker: dict) -> None:
             ),
             "seen_at": datetime.now(timezone.utc).isoformat(),
         }
+        vessel_last_known[vid] = stored
         if len(vessel_last_known) > _MAX_VESSELS:
             excess = len(vessel_last_known) - int(_MAX_VESSELS * 0.9)
             oldest = sorted(
@@ -631,6 +654,15 @@ async def upsert_vessel(marker: dict) -> None:
             )[:excess]
             for k in oldest:
                 vessel_last_known.pop(k, None)
+    if lat is not None and lon is not None:
+        from ws_manager import manager
+        from tracks import record_position
+        from geofence import record_and_notify
+        await manager.publish_point("ais", stored, lat=lat, lon=lon)
+        await record_position("vessel", stored)
+        await record_and_notify(
+            source_kind="ais", entity_id=vid, lat=lat, lon=lon, payload=stored,
+        )
 
 
 def _wfigs_params(bbox: str | None, *, offset_m: float = 250.0) -> dict:
@@ -662,6 +694,10 @@ async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) ->
         return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params))
 
     rows = await _ttl_get(f"wfigs:inc:{bbox_cell_key(bbox)}", 600.0, _load)
+    fire_last_known[:] = list(rows)
+    if rows and aircraft_last_known:
+        from fire_aircraft import correlate_and_notify
+        await correlate_and_notify(rows, list(aircraft_last_known.values()))
     return rows[:limit]
 
 

app/main.py

@@ -21,7 +21,7 @@ from pathlib import Path
 from uuid import UUID
 
 import structlog
-from fastapi import FastAPI, HTTPException, Query
+from fastapi import BackgroundTasks, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
@@ -41,6 +41,7 @@ from schemas import (
     KeyOut, KeyValueIn,
     SearchResult, SentimentSummary, SourceType,
     SearchQuery, TimelinePoint, VesselBboxUpdate,
+    GeofenceCreate, GeofenceUpdate,
 )
 from ingestor import ingest_event, fetch_and_process
 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
@@ -61,6 +62,11 @@ async def _lifespan(app: FastAPI):
     await init_extensions()
     from live_layers import close_http, init_http
     await init_http()
+    try:
+        from geofence import refresh_cache
+        await refresh_cache()
+    except Exception:
+        pass
     from config import AISSTREAM_IN_APP
     ais_task = None
     if AISSTREAM_IN_APP:
@@ -664,6 +670,56 @@ 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}."""
+    from ws_manager import manager
+
+    client_id = str(id(ws))
+    await ws.accept()
+    queue = manager.register(client_id)
+
+    async def _pump() -> None:
+        try:
+            while True:
+                msg = await queue.get()
+                await ws.send_json(msg)
+        except Exception:  # noqa: BLE001
+            return
+
+    pump = asyncio.create_task(_pump())
+    try:
+        while True:
+            data = await ws.receive_json()
+            if not isinstance(data, dict):
+                continue
+            if data.get("type") == "viewport" and data.get("bbox"):
+                try:
+                    manager.set_viewport(client_id, parse_bbox(str(data["bbox"])))
+                except ValueError:
+                    continue
+    except WebSocketDisconnect:
+        pass
+    finally:
+        pump.cancel()
+        manager.unregister(client_id)
+
+
 @app.post("/api/ingest/process")
 async def trigger_nats_processing(batch_size: int = 100):
     """Process pending NATS JetStream messages."""
@@ -1135,10 +1191,15 @@ async def map_radar():
 async def list_aircraft(
     bbox: str = Query(..., description="minlon,minlat,maxlon,maxlat"),
     limit: int = Query(2000, ge=1, le=5000),
+    timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
 ):
     """Viewport ADS-B last-known (ADSB.lol). Requires bbox; radius clamped ≤ 150 nm."""
     _parse_bbox_query(bbox)
     try:
+        from tracks import fetch_positions_at, parse_timestamp
+        ts = parse_timestamp(timestamp)
+        if ts is not None:
+            return overlay_json(await fetch_positions_at("aircraft", ts, bbox, limit), 5)
         return overlay_json(await fetch_aircraft(bbox, limit), 5)
     except ValueError as exc:
         raise HTTPException(422, str(exc)) from exc
@@ -1166,11 +1227,16 @@ async def list_trains(
 async def list_vessels(
     bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
     limit: int = Query(2000, ge=1, le=5000),
+    timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
 ):
     """AIS last-known from the server-side AISStream worker. Empty without a key."""
     if bbox:
         _parse_bbox_query(bbox)
     try:
+        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)
         return overlay_json(await fetch_vessels(bbox, limit), 5)
     except ValueError as exc:
         raise HTTPException(422, str(exc)) from exc
@@ -1204,6 +1270,90 @@ async def vessels_subscribe(payload: VesselBboxUpdate):
     return {"ok": True, "bbox": raw}
 
 
+@app.get("/api/tracks/range")
+async def tracks_range():
+    """Earliest/latest 1-minute track buckets for the DVR slider."""
+    from tracks import track_range
+    return await track_range()
+
+
+@app.get("/api/geofences")
+async def api_list_geofences():
+    """Drawn GeoJSON polygons. Not /api/alerts (entity/keyword)."""
+    from geofence import list_geofences
+    return await list_geofences()
+
+
+@app.post("/api/geofences", status_code=201)
+async def api_create_geofence(payload: GeofenceCreate):
+    from geofence import create_geofence, validate_polygon_geojson
+    try:
+        validate_polygon_geojson(payload.geojson)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    try:
+        return await create_geofence(payload.name, payload.geojson, payload.active)
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(503, f"geofence persist failed: {exc}") from exc
+
+
+@app.patch("/api/geofences/{gid}")
+async def api_update_geofence(gid: str, payload: GeofenceUpdate):
+    from geofence import update_geofence, validate_polygon_geojson
+    if payload.geojson is not None:
+        try:
+            validate_polygon_geojson(payload.geojson)
+        except ValueError as exc:
+            raise HTTPException(422, str(exc)) from exc
+    row = await update_geofence(
+        gid, name=payload.name, geojson=payload.geojson, active=payload.active,
+    )
+    if row is None:
+        raise HTTPException(404, "geofence not found")
+    return row
+
+
+@app.delete("/api/geofences/{gid}", status_code=204)
+async def api_delete_geofence(gid: str):
+    from geofence import delete_geofence
+    await delete_geofence(gid)
+    return None
+
+
+@app.get("/api/geofence-alerts")
+async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)):
+    from sqlalchemy import text as sql_text
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(sql_text(
+                """
+                SELECT id::text, geofence_id::text, source_kind, entity_id,
+                       lat, lon, payload, created_at
+                FROM geofence_alerts
+                ORDER BY created_at DESC
+                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 []
+
+
+@app.get("/api/fire-aircraft")
+async def api_fire_aircraft(limit: int = Query(200, ge=1, le=1000)):
+    """Persisted firefighting ADS-B × wildfire correlations (20 mi)."""
+    from fire_aircraft import recent_hits
+    return await recent_hits(limit)
+
+
 @app.get("/api/fire-incidents")
 async def list_fire_incidents(
     bbox: str | None = Query(None),

app/requirements.txt

@@ -11,3 +11,4 @@ feedparser>=6.0
 python-dateutil>=2.9
 structlog>=24.4
 websockets>=14
+cachetools>=5.5

app/run_ingester.py

@@ -107,6 +107,11 @@ async def main() -> None:
         "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, fires=%s, interval=%ss)",
         len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, ENABLE_FIRES, INTERVAL,
     )
+    try:
+        from geofence import refresh_cache
+        await refresh_cache()
+    except Exception:
+        logger.exception("geofence cache refresh failed (ST_Intersects still runs on ingest)")
     tasks: list[asyncio.Task] = []
     if ENABLE_FIRES:
         # Fire ingest only starts once FIRMS_MAP_KEY is set (ingest_fires logs

app/schemas.py

@@ -310,3 +310,15 @@ class VesselBboxUpdate(BaseModel):
 
     bbox: str | None = None
 
+
+class GeofenceCreate(BaseModel):
+    name: str
+    geojson: dict
+    active: bool = True
+
+
+class GeofenceUpdate(BaseModel):
+    name: Optional[str] = None
+    geojson: Optional[dict] = None
+    active: Optional[bool] = None
+

app/sources.py

@@ -12,6 +12,7 @@ import feedparser
 import nats
 
 from config import NATS_URL
+from upstream_cache import rss_cache
 
 logger = logging.getLogger("osint.sources")
 
@@ -40,12 +41,16 @@ async def publish_event(subject: str, event: dict):
 
 # ─── RSS Feed Ingestor ──────────────────────────────────────────────────
 
-async def ingest_rss_feed(feed_url: str):
+async def ingest_rss_feed(feed_url: str, source_id: str | None = None):
     """Fetch and parse an RSS feed, publish items to NATS."""
+    text = rss_cache.get(feed_url)
+    if text is None:
         async with httpx.AsyncClient(timeout=30) as client:
             resp = await client.get(feed_url)
             resp.raise_for_status()
-        feed = feedparser.parse(resp.text)
+            text = resp.text
+        rss_cache[feed_url] = text
+    feed = feedparser.parse(text)
 
     count = 0
     for entry in feed.entries[:100]:  # max 100 per run

app/static/index.html

@@ -265,6 +265,26 @@
         .hud-chip .c.cams { background: #4ade80; box-shadow: 0 0 6px #4ade80; }
         .hud-chip .c.blips { background: var(--magenta); box-shadow: 0 0 6px var(--magenta); }
         .hud-chip.off { opacity: 0.4; }
+        #dvr-bar {
+            position: absolute; left: 50%; bottom: 12px; transform: translateX(-50%);
+            z-index: 520; display: flex; align-items: center; gap: 0.55rem;
+            background: rgba(6,11,20,0.86); border: 1px solid var(--line-hi);
+            border-radius: 4px; padding: 0.28rem 0.7rem;
+            font-family: 'Share Tech Mono', monospace; font-size: 0.66rem; color: var(--muted);
+            min-width: 280px; max-width: 52%;
+        }
+        #dvr-bar input[type=range] { flex: 1; accent-color: var(--cyan); }
+        #dvr-bar.live #dvr-live { color: var(--green); text-shadow: 0 0 8px rgba(83,240,165,0.5); }
+        #dvr-toast {
+            position: absolute; top: 12px; right: 12px; z-index: 700;
+            max-width: 280px; pointer-events: none;
+        }
+        #dvr-toast .gf-alert {
+            background: rgba(255,46,151,0.16); border: 1px solid var(--magenta);
+            color: var(--text); font-family: 'Share Tech Mono', monospace;
+            font-size: 0.68rem; padding: 0.4rem 0.6rem; margin-bottom: 0.35rem;
+            border-radius: 4px;
+        }
 
         /* ── Layer control panel (dark HUD) ── */
         #layer-panel {
@@ -754,6 +774,12 @@
                             </div>
                             <div class="lp-note">Needs AISSTREAM_API_KEY in Keys.</div>
                         </div>
+                        <div class="lp-layer">
+                            <div class="lp-row">
+                                <button class="btn" id="gf-draw" type="button" onclick="toggleGeofenceDraw()">Draw geofence</button>
+                            </div>
+                            <div class="lp-note">No Leaflet.Draw — click vertices, double-click to close. Saved to POST /api/geofences.</div>
+                        </div>
                         <div class="lp-layer">
                             <div class="lp-row">
                                 <label class="lp-name"><input type="checkbox" id="lp-storms-on" checked onchange="toggleStorms()"> <span class="lp-dot storms"></span> NHC Storms</label>
@@ -776,6 +802,12 @@
                     <div class="hud-chip" id="hud-cams"><span class="c cams"></span>CAMS <b id="hud-cams-n">—</b></div>
                     <div class="hud-chip" id="hud-blips"><span class="c blips"></span>BLIPS <b id="hud-blips-n">—</b></div>
                 </div>
+                <div id="dvr-bar" class="live">
+                    <button class="btn" id="dvr-live" type="button" onclick="dvrGoLive()">LIVE</button>
+                    <input type="range" id="dvr-slider" min="0" max="360" value="360" oninput="dvrScrub(this.value)" aria-label="DVR time">
+                    <span id="dvr-label">live</span>
+                </div>
+                <div id="dvr-toast"></div>
             </div>
         </section>
 
@@ -1658,6 +1690,76 @@ function bboxCell() {
     if (!map) return '';
     return currentBBox().split(',').map(n => Number(n).toFixed(2)).join(',') + '@' + map.getZoom();
 }
+let liveWs = null;
+const FF_ICAO = new Set(['AT802','AT8T','AT8P','C130','C30J','C130J','DC10','MD10','MD87','S64','UH60','S70','CL415','CL215','B350','OV10','C208']);
+let firefighterHex = new Set();
+let dvrTs = null;
+let dvrMinMs = 0, dvrMaxMs = 0;
+let gfDrawOn = false, gfVerts = [], gfLayer = null, gfSaved = null;
+function liveWsUrl() {
+    const proto = location.protocol === 'https:' ? 'wss:' : 'ws:';
+    return proto + '//' + location.host + '/ws/live';
+}
+function sendLiveViewport() {
+    if (!liveWs || liveWs.readyState !== 1 || !map) return;
+    liveWs.send(JSON.stringify({ type: 'viewport', bbox: currentBBox() }));
+}
+function upsertLivePoint(group, p, colorFn, feed) {
+    if (!map || !p || p.id == null || p.lat == null || p.lon == null) return group;
+    if (!group || !map.hasLayer(group) || !group._osintById) {
+        return renderPoints(group, [p], colorFn, true, feed);
+    }
+    const id = String(p.id);
+    const col = sanitizeColor(colorFn(p), '#35e0ff');
+    let m = group._osintById.get(id);
+    if (m) {
+        m.setLatLng([p.lat, p.lon]);
+        if (feed) m.setIcon(feedIcon(feed, col, p.heading));
+        else if (m.setStyle) m.setStyle({ color: col, fillColor: col });
+    } else {
+        m = makePointMarker(p, colorFn, feed, pointCanvas());
+        if (m) { group.addLayer(m); group._osintById.set(id, m); }
+    }
+    return group;
+}
+function applyLiveMarker(kind, p) {
+    if (!p) return;
+    if (kind === 'ais' && vesselsOn && map && map.getZoom() > 3 && !dvrTs) {
+        vesselsGroup = upsertLivePoint(vesselsGroup, p, q => {
+            const sog = Number(q.speed || 0);
+            return sog > 0.5 ? '#2dd4bf' : '#64748b';
+        }, 'vessel');
+    } else if (kind === 'adsb' && acOn && map && map.getZoom() > 3 && !dvrTs) {
+        acGroup = upsertLivePoint(acGroup, p, q => acColor(q), 'ac');
+    } else if (kind === 'geofence_alert') {
+        showGeofenceToast(p);
+    } else if (kind === 'fire_aircraft') {
+        firefighterHex.add(String(p.aircraft_hex || ''));
+        if (acOn && p.aircraft_lat != null) {
+            acGroup = upsertLivePoint(acGroup, {
+                id: p.aircraft_hex, lat: p.aircraft_lat, lon: p.aircraft_lon,
+                label: p.label || p.aircraft_hex, extra: { type: p.aircraft_type, firefighter: true },
+            }, q => acColor(q), 'ac');
+        }
+    }
+}
+function acColor(p) {
+    const t = String(((p.extra || {}).type || '')).toUpperCase();
+    if ((p.extra || {}).firefighter || firefighterHex.has(String(p.id)) || FF_ICAO.has(t)) return '#fb923c';
+    return altColor((p.extra || {}).alt_baro);
+}
+function connectLiveWs() {
+    if (liveWs && (liveWs.readyState === 0 || liveWs.readyState === 1)) return;
+    try { liveWs = new WebSocket(liveWsUrl()); } catch (e) { return; }
+    liveWs.onopen = () => sendLiveViewport();
+    liveWs.onmessage = (ev) => {
+        try {
+            const msg = JSON.parse(ev.data);
+            applyLiveMarker(msg.type, msg.payload);
+        } catch (e) { /* ignore malformed */ }
+    };
+    liveWs.onclose = () => { liveWs = null; setTimeout(connectLiveWs, 4000); };
+}
 function overlayFetch(url) {
     return fetch(url, overlayAbort ? { signal: overlayAbort.signal } : {});
 }
@@ -1754,7 +1856,8 @@ async function initMap() {
                 if (camsOn) loadCams();
                 if (blipsOn) loadBlips();
                 refreshLiveOverlays();
-            }, 300);
+                sendLiveViewport();
+            }, 500);
         });
         // Static Blue Marble has no time-domain fetch — don't block overlays on GIBS.
         mapLayerChanged();
@@ -1778,6 +1881,7 @@ async function initMap() {
         if (camsOn) loadCams();
         if (blipsOn) loadBlips();
         refreshLiveOverlays();
+        connectLiveWs();
     } catch(e) {
         hint.textContent = `Failed to load map layers: ${e.message || e}`;
         console.error('Map init failed', e);
@@ -2226,6 +2330,8 @@ function refreshLiveOverlays() {
     if (trainsOn) loadTrains();
     if (vesselsOn) loadVessels();
     if (stormsOn) loadStorms();
+    loadGeofences();
+    loadFireAircraftHits();
 }
 function addExtraAttrib(html) {
     if (!map || !html || extraAttribs.has(html)) return;
@@ -2554,10 +2660,10 @@ async function loadAircraft() {
     }
     const req = ++overlayReq.ac;
     try {
-        const r = await overlayFetch(`${API}/api/aircraft?bbox=${currentBBox()}`);
+        const r = await overlayFetch(`${API}/api/aircraft?bbox=${currentBBox()}${dvrQs()}`);
         const pts = await r.json();
         if (req !== overlayReq.ac) return;
-        acGroup = renderPoints(acGroup, Array.isArray(pts) ? pts : [], p => altColor((p.extra || {}).alt_baro), true, 'ac');
+        acGroup = renderPoints(acGroup, Array.isArray(pts) ? pts : [], p => acColor(p), true, 'ac');
         document.getElementById('lp-ac-count').textContent = (pts.length || 0).toLocaleString();
         addExtraAttrib('<a href="https://www.adsb.lol/docs/open-data/api">ADSB.lol</a> ODbL');
     } catch (e) {
@@ -2622,7 +2728,7 @@ async function loadVessels() {
         }).catch(() => {});
     }
     try {
-        const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}`);
+        const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}${dvrQs()}`);
         const pts = await r.json();
         if (req !== overlayReq.vessels) return;
         vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => {
@@ -2674,6 +2780,122 @@ initMap();
 // contend with the first bbox burst. Payload is slim (no article bodies).
 setTimeout(() => loadNews(true), 400);
 setInterval(checkHealth, 30000);
+
+/* Phase 2: DVR slider + geofence draw (no Leaflet.Draw) + fire/aircraft hits */
+function dvrQs() { return dvrTs ? `&timestamp=${encodeURIComponent(dvrTs)}` : ''; }
+function dvrGoLive() {
+    dvrTs = null;
+    const bar = document.getElementById('dvr-bar');
+    if (bar) bar.classList.add('live');
+    const sl = document.getElementById('dvr-slider');
+    if (sl) sl.value = sl.max;
+    const lab = document.getElementById('dvr-label');
+    if (lab) lab.textContent = 'live';
+    if (acOn) loadAircraft();
+    if (vesselsOn) loadVessels();
+}
+function dvrScrub(v) {
+    if (!dvrMaxMs) { dvrGoLive(); return; }
+    const ms = dvrMinMs + (Number(v) / 360) * (dvrMaxMs - dvrMinMs);
+    dvrTs = new Date(ms).toISOString();
+    const bar = document.getElementById('dvr-bar');
+    if (bar) bar.classList.toggle('live', Number(v) >= 359);
+    if (Number(v) >= 359) { dvrGoLive(); return; }
+    const lab = document.getElementById('dvr-label');
+    if (lab) lab.textContent = dvrTs.slice(11, 16) + 'Z';
+    if (acOn) loadAircraft();
+    if (vesselsOn) loadVessels();
+}
+async function initDvrRange() {
+    try {
+        const r = await fetch(`${API}/api/tracks/range`);
+        const b = await r.json();
+        dvrMinMs = Date.parse(b.min);
+        dvrMaxMs = Date.parse(b.max);
+        if (!dvrMinMs || !dvrMaxMs) {
+            dvrMaxMs = Date.now();
+            dvrMinMs = dvrMaxMs - 6 * 3600 * 1000;
+        }
+    } catch (e) {
+        dvrMaxMs = Date.now();
+        dvrMinMs = dvrMaxMs - 6 * 3600 * 1000;
+    }
+}
+function showGeofenceToast(p) {
+    const box = document.getElementById('dvr-toast');
+    if (!box) return;
+    const el = document.createElement('div');
+    el.className = 'gf-alert';
+    el.textContent = `GEOFENCE ${p.geofence_name || ''} · ${p.source_kind} ${p.entity_id || ''}`;
+    box.prepend(el);
+    setTimeout(() => el.remove(), 8000);
+}
+function toggleGeofenceDraw() {
+    gfDrawOn = !gfDrawOn;
+    gfVerts = [];
+    if (gfLayer && map) { map.removeLayer(gfLayer); gfLayer = null; }
+    const btn = document.getElementById('gf-draw');
+    if (btn) btn.textContent = gfDrawOn ? 'Click map… (dbl-click close)' : 'Draw geofence';
+    if (!map) return;
+    if (gfDrawOn) {
+        map.getContainer().style.cursor = 'crosshair';
+        map.on('click', onGfClick);
+        map.on('dblclick', onGfClose);
+    } else {
+        map.getContainer().style.cursor = '';
+        map.off('click', onGfClick);
+        map.off('dblclick', onGfClose);
+    }
+}
+function onGfClick(e) {
+    if (!gfDrawOn) return;
+    L.DomEvent.stop(e);
+    gfVerts.push([e.latlng.lng, e.latlng.lat]);
+    if (gfLayer && map) map.removeLayer(gfLayer);
+    if (gfVerts.length >= 2) {
+        gfLayer = L.polyline(gfVerts.map(v => [v[1], v[0]]), { color: '#ff2e97', weight: 2 }).addTo(map);
+    }
+}
+async function onGfClose(e) {
+    if (!gfDrawOn) return;
+    L.DomEvent.stop(e);
+    if (gfVerts.length < 3) return;
+    const ring = gfVerts.concat([gfVerts[0]]);
+    const geojson = { type: 'Polygon', coordinates: [ring] };
+    try {
+        await fetch(`${API}/api/geofences`, {
+            method: 'POST', headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ name: 'Fence ' + new Date().toISOString().slice(11, 19), geojson }),
+        });
+    } catch (err) { console.error('geofence save failed', err); }
+    toggleGeofenceDraw();
+    loadGeofences();
+}
+async function loadGeofences() {
+    if (!map) return;
+    try {
+        const r = await overlayFetch(`${API}/api/geofences`);
+        const rows = await r.json();
+        if (gfSaved && map.hasLayer(gfSaved)) map.removeLayer(gfSaved);
+        const feats = (Array.isArray(rows) ? rows : []).map(f => ({
+            type: 'Feature', properties: { name: f.name, id: f.id }, geometry: f.geojson,
+        }));
+        gfSaved = L.geoJSON({ type: 'FeatureCollection', features: feats }, {
+            style: { color: '#ff2e97', weight: 2, fillOpacity: 0.08 },
+        }).addTo(map);
+    } catch (e) { if (!isAbort(e)) console.error('geofences load failed', e); }
+}
+async function loadFireAircraftHits() {
+    try {
+        const r = await fetch(`${API}/api/fire-aircraft`);
+        const rows = await r.json();
+        (Array.isArray(rows) ? rows : []).forEach(h => {
+            if (h.aircraft_hex) firefighterHex.add(String(h.aircraft_hex));
+        });
+    } catch (e) { /* empty without table */ }
+}
+initDvrRange();
+
 </script>
 </body>
 </html>

app/tracks.py

@@ -0,0 +1,245 @@
+"""Timescale 1-minute track rollups for DVR playback.
+
+Live overlays stay in memory. Historical `?timestamp=` reads the 1-minute
+continuous aggregates (or an in-process downsample when the DB is down).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from typing import Any, Literal
+
+from sqlalchemy import text
+
+from database import async_session
+from live_layers import parse_bbox, to_marker
+
+
+TRACK_BUCKET = "1 minute"
+Kind = Literal["vessel", "aircraft"]
+
+_RAW_TABLE = {
+    "vessel": "vessel_positions",
+    "aircraft": "aircraft_positions",
+}
+_CAGG = {
+    "vessel": "vessel_tracks_1min",
+    "aircraft": "aircraft_tracks_1min",
+}
+_ID_COL = {
+    "vessel": "mmsi",
+    "aircraft": "hex",
+}
+
+# Last persist time per entity so AIS/ADS-B does not write every frame.
+_last_write: dict[tuple[str, str], datetime] = {}
+_MIN_WRITE_GAP = timedelta(seconds=20)
+
+
+def minute_bucket(ts: datetime) -> datetime:
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return ts.replace(second=0, microsecond=0)
+
+
+def downsample_tracks(rows: list[dict]) -> list[dict]:
+    """Last sample per id per 1-minute bucket (mirrors the CAGG)."""
+    last: dict[tuple[str, datetime], dict] = {}
+    for row in rows:
+        rid = str(row.get("id") or "")
+        ts = row.get("ts")
+        if not rid or not isinstance(ts, datetime):
+            continue
+        bucket = minute_bucket(ts)
+        key = (rid, bucket)
+        prev = last.get(key)
+        if prev is None or ts >= prev["ts"]:
+            last[key] = {**row, "id": rid, "bucket": bucket, "ts": ts}
+    out = []
+    for (_id, bucket), row in last.items():
+        out.append({
+            "id": row["id"],
+            "bucket": bucket,
+            "lat": row.get("lat"),
+            "lon": row.get("lon"),
+            "heading": row.get("heading"),
+            "speed": row.get("speed"),
+            "label": row.get("label"),
+        })
+    return out
+
+
+def positions_at_timestamp(rows: list[dict], ts: datetime) -> list[dict]:
+    """Positions whose 1-minute bucket equals floor(ts)."""
+    want = minute_bucket(ts)
+    picked = [r for r in downsample_tracks(rows) if r["bucket"] == want]
+    return [
+        to_marker(
+            r["id"], r.get("lat"), r.get("lon"),
+            heading=r.get("heading"), speed=r.get("speed"),
+            label=r.get("label") or r["id"],
+        )
+        for r in picked
+        if r.get("lat") is not None and r.get("lon") is not None
+    ]
+
+
+def parse_timestamp(value: str | datetime | None) -> datetime | None:
+    if value is None or value == "":
+        return None
+    if isinstance(value, datetime):
+        ts = value
+    else:
+        raw = str(value).strip().replace("Z", "+00:00")
+        ts = datetime.fromisoformat(raw)
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return ts
+
+
+async def record_position(kind: Kind, marker: dict, ts: datetime | None = None) -> bool:
+    """Insert one sample into the raw hypertable (rate-limited)."""
+    vid = str(marker.get("id") or "")
+    lat, lon = marker.get("lat"), marker.get("lon")
+    if not vid or lat is None or lon is None:
+        return False
+    now = ts or datetime.now(timezone.utc)
+    key = (kind, vid)
+    prev = _last_write.get(key)
+    if prev is not None and now - prev < _MIN_WRITE_GAP:
+        return False
+    _last_write[key] = now
+    table = _RAW_TABLE[kind]
+    id_col = _ID_COL[kind]
+    extra = marker.get("extra") or {}
+    try:
+        async with async_session() as session:
+            await session.execute(
+                text(
+                    f"""
+                    INSERT INTO {table} ({id_col}, ts, lat, lon, heading, speed, label, extra)
+                    VALUES (:id, :ts, :lat, :lon, :heading, :speed, :label, CAST(:extra AS jsonb))
+                    ON CONFLICT ({id_col}, ts) DO NOTHING
+                    """
+                ),
+                {
+                    "id": vid,
+                    "ts": now,
+                    "lat": float(lat),
+                    "lon": float(lon),
+                    "heading": marker.get("heading"),
+                    "speed": marker.get("speed"),
+                    "label": marker.get("label") or vid,
+                    "extra": json.dumps(extra),
+                },
+            )
+            await session.commit()
+        return True
+    except Exception:
+        return False
+
+
+async def fetch_positions_at(
+    kind: Kind,
+    ts: datetime,
+    bbox: str | None = None,
+    limit: int = 2000,
+) -> list[dict]:
+    """Read the 1-minute CAGG for the bucket containing ``ts``."""
+    bucket = minute_bucket(ts)
+    table = _CAGG[kind]
+    id_col = _ID_COL[kind]
+    where = "bucket = :bucket"
+    params: dict[str, Any] = {"bucket": bucket, "limit": limit}
+    if bbox:
+        minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
+        where += " AND lon BETWEEN :minlon AND :maxlon AND lat BETWEEN :minlat AND :maxlat"
+        params.update(minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat)
+    sql = f"""
+        SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket
+        FROM {table}
+        WHERE {where}
+        LIMIT :limit
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(text(sql), params)).mappings().all()
+        points = [
+            to_marker(
+                r["id"], r["lat"], r["lon"],
+                heading=r["heading"], speed=r["speed"],
+                label=r["label"] or r["id"],
+                extra={"bucket": r["bucket"].isoformat() if r["bucket"] else None, "dvr": True},
+            )
+            for r in rows
+            if r["lat"] is not None and r["lon"] is not None
+        ]
+        return points
+    except Exception:
+        return []
+
+
+async def track_range() -> dict:
+    """Earliest/latest buckets across both CAGGs — slider bounds."""
+    try:
+        async with async_session() as session:
+            row = (await session.execute(text(
+                """
+                SELECT
+                  LEAST(
+                    (SELECT min(bucket) FROM vessel_tracks_1min),
+                    (SELECT min(bucket) FROM aircraft_tracks_1min)
+                  ) AS tmin,
+                  GREATEST(
+                    (SELECT max(bucket) FROM vessel_tracks_1min),
+                    (SELECT max(bucket) FROM aircraft_tracks_1min)
+                  ) AS tmax
+                """
+            ))).mappings().first()
+        if not row or row["tmin"] is None:
+            now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
+            return {"min": (now - timedelta(hours=6)).isoformat(), "max": now.isoformat()}
+        return {
+            "min": row["tmin"].isoformat(),
+            "max": row["tmax"].isoformat(),
+        }
+    except Exception:
+        now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
+        return {"min": (now - timedelta(hours=6)).isoformat(), "max": now.isoformat()}
+
+
+async def recent_markers(kind: Kind, limit: int = 2000) -> list[dict]:
+    """Latest raw sample per id — used when in-process last-known is empty."""
+    table = _RAW_TABLE[kind]
+    id_col = _ID_COL[kind]
+    sql = f"""
+        SELECT DISTINCT ON ({id_col})
+               {id_col} AS id, lat, lon, heading, speed, label, extra
+        FROM {table}
+        WHERE ts > now() - interval '15 minutes'
+        ORDER BY {id_col}, ts DESC
+        LIMIT :limit
+    """
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(text(sql), {"limit": limit})).mappings().all()
+        out = []
+        for r in rows:
+            extra = r.get("extra") or {}
+            if isinstance(extra, str):
+                try:
+                    extra = json.loads(extra)
+                except (TypeError, ValueError):
+                    extra = {}
+            m = to_marker(
+                r["id"], r["lat"], r["lon"],
+                heading=r["heading"], speed=r["speed"],
+                label=r["label"] or r["id"],
+                extra=extra if isinstance(extra, dict) else {},
+            )
+            if m.get("lat") is not None and m.get("lon") is not None:
+                out.append(m)
+        return out
+    except Exception:
+        return []

app/upstream_cache.py

@@ -0,0 +1,11 @@
+"""In-process TTL caches for chatty upstreams (FIRMS, RSS). No Redis."""
+
+from __future__ import annotations
+
+from cachetools import TTLCache
+
+# FIRMS NRT updates every ~5–10 min; 5 min / 100 keys is enough for bbox×dataset.
+firms_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
+
+# News RSS: 1 minute is enough to absorb dashboard double-clicks / retries.
+rss_cache: TTLCache = TTLCache(maxsize=100, ttl=60)

app/ws_manager.py

@@ -0,0 +1,80 @@
+"""In-memory WebSocket pub/sub with viewport filtering.
+
+Zero extra deps. Ingest workers publish AIS/ADS-B points; only clients whose
+current map bbox contains the point receive the payload. No Redis/Kafka.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+BBox = tuple[float, float, float, float]  # minlon, minlat, maxlon, maxlat
+
+
+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:
+        return False
+    minlon, minlat, maxlon, maxlat = bbox
+    return minlon <= lon <= maxlon and minlat <= lat <= maxlat
+
+
+class ConnectionManager:
+    """Maps Tailscale/browser clients → viewport bbox + per-client queue."""
+
+    def __init__(self) -> None:
+        self._queues: dict[str, asyncio.Queue] = {}
+        self._viewports: dict[str, BBox] = {}
+
+    def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue:
+        q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
+        self._queues[client_id] = q
+        return q
+
+    def unregister(self, client_id: str) -> None:
+        self._queues.pop(client_id, None)
+        self._viewports.pop(client_id, None)
+
+    def set_viewport(self, client_id: str, bbox: BBox) -> None:
+        if client_id in self._queues:
+            self._viewports[client_id] = bbox
+
+    def viewport_of(self, client_id: str) -> BBox | None:
+        return self._viewports.get(client_id)
+
+    def has_clients(self) -> bool:
+        return bool(self._queues)
+
+    async def publish_point(
+        self,
+        kind: str,
+        payload: dict[str, Any],
+        *,
+        lat: float,
+        lon: float,
+    ) -> 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
+        tab cannot stall ingest. Returns the number of clients that got a copy.
+        """
+        msg = {"type": kind, "payload": payload}
+        sent = 0
+        for client_id, queue in list(self._queues.items()):
+            if not point_in_bbox(lon, lat, self._viewports.get(client_id)):
+                continue
+            if queue.full():
+                try:
+                    queue.get_nowait()
+                except asyncio.QueueEmpty:
+                    pass
+            try:
+                queue.put_nowait(msg)
+            except asyncio.QueueFull:
+                continue
+            sent += 1
+        return sent
+
+
+manager = ConnectionManager()

docker-compose.yml

@@ -30,7 +30,16 @@ services:
       POSTGRES_DB: ${DB_NAME:-osint_data}
     # Ensure TimescaleDB is preloaded (conf.d drop-in may be ignored by the
     # official image's runtime-generated postgresql.conf, so pass it explicitly).
-    command: ["-c", "shared_preload_libraries=timescaledb"]
+    # shared_buffers capped at 2GB for Pi 5 8GB / 4 cores.
+    command:
+      [
+        "-c", "shared_preload_libraries=timescaledb",
+        "-c", "shared_buffers=2GB",
+      ]
+    deploy:
+      resources:
+        limits:
+          memory: 3G
     ports:
       - "127.0.0.1:5432:5432"
     volumes:
@@ -124,6 +133,10 @@ services:
       AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1}
     ports:
       - "127.0.0.1:8000:8000"
+    deploy:
+      resources:
+        limits:
+          memory: 2G
     healthcheck:
       test: ["CMD-SHELL", "python -c \"import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/api/health').status==200 else 1)\""]
       interval: 30s

tests/test_bg_jobs.py

@@ -0,0 +1,64 @@
+"""masscan/ffmpeg stay off the request path (asyncio.create_task)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import bg_jobs
+
+
+def test_schedule_masscan_pass_returns_without_awaiting_scan(monkeypatch):
+    started = {"n": 0}
+
+    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():
+    assert bg_jobs.MASSCAN_PPS_CAP == 200
+
+
+def test_schedule_ffmpeg_snapshot_is_a_task_not_inline(monkeypatch):
+    calls = {"n": 0}
+
+    async def fake_grab(url, timeout=8.0):
+        calls["n"] += 1
+        await asyncio.sleep(5)
+        return b"\xff\xd8fakejpeg"
+
+    monkeypatch.setattr(bg_jobs, "_ffmpeg_grab", fake_grab)
+    bg_jobs._ffmpeg_tasks.clear()
+    bg_jobs._ffmpeg_cache.clear()
+
+    async def run():
+        task = bg_jobs.schedule_ffmpeg_snapshot("rtsp://10.0.0.1/")
+        assert isinstance(task, asyncio.Task)
+        assert not task.done()
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, Exception):
+            pass
+
+    asyncio.run(run())

tests/test_fire_aircraft.py

@@ -0,0 +1,66 @@
+"""WFIGS/FIRMS × firefighting ADS-B correlation within 20 miles."""
+
+from __future__ import annotations
+
+from fire_aircraft import (
+    FIREFIGHTER_ICAO,
+    RADIUS_MILES,
+    correlate_aircraft_to_fires,
+    is_firefighter,
+)
+
+
+def _ac(hex_id, lat, lon, icao, **extra):
+    return {
+        "id": hex_id,
+        "lat": lat,
+        "lon": lon,
+        "label": hex_id,
+        "heading": 0,
+        "speed": 120,
+        "extra": {"type": icao, "hex": hex_id, **extra},
+    }
+
+
+def _fire(name, lat, lon, **extra):
+    return {
+        "id": name,
+        "lat": lat,
+        "lon": lon,
+        "label": name,
+        "extra": extra,
+    }
+
+
+def test_air_tractor_is_firefighter_airliner_is_not():
+    assert is_firefighter(_ac("aaa", 0, 0, "AT802")) is True
+    assert is_firefighter(_ac("bbb", 0, 0, "C130")) is True
+    assert is_firefighter(_ac("ccc", 0, 0, "B738")) is False
+    assert "AT802" in FIREFIGHTER_ICAO
+
+
+def test_flags_tanker_within_20_miles_of_fire():
+    # ~10 miles north of a Piedmont fire
+    fire = _fire("Jones Gap", 35.00, -82.00)
+    tanker = _ac("acf001", 35.145, -82.00, "AT802")
+    airliner = _ac("a0b738", 35.145, -82.00, "B738")
+    far = _ac("acfar", 35.50, -82.00, "C130")  # ~34 miles
+    hits = correlate_aircraft_to_fires([fire], [tanker, airliner, far])
+    assert RADIUS_MILES == 20.0
+    assert len(hits) == 1
+    h = hits[0]
+    assert h["aircraft_hex"] == "acf001"
+    assert h["fire_id"] == "Jones Gap"
+    assert h["aircraft_type"] == "AT802"
+    assert 0 < h["distance_mi"] <= 20.0
+
+
+def test_persist_shape_has_reload_fields():
+    fire = _fire("Jones Gap", 35.00, -82.00, src="wfigs")
+    tanker = _ac("acf001", 35.10, -82.00, "S64")
+    hits = correlate_aircraft_to_fires([fire], [tanker])
+    assert set(hits[0]).issuperset({
+        "fire_id", "fire_lat", "fire_lon",
+        "aircraft_hex", "aircraft_type", "aircraft_lat", "aircraft_lon",
+        "distance_mi",
+    })

tests/test_fire_ingest.py

@@ -85,3 +85,87 @@ def test_ingest_fire_row_drops_malformed(clean_fires):
         assert await ingest_fire_row(make_fire_msg(acq_time="garbage")) is False
 
     asyncio.run(run())
+
+
+class _InsertSession:
+    """async_session stand-in: fire insert succeeds (rowcount=1)."""
+
+    def __init__(self):
+        self.rowcount = 1
+
+    async def execute(self, *a, **k):
+        return self
+
+    async def commit(self):
+        return None
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        return False
+
+
+def test_ingest_fire_row_geofence_when_cache_empty(monkeypatch):
+    """Ingester process has empty geofence cache; still notify on FIRMS insert."""
+    import geofence
+    import ingestor
+    import live_layers
+
+    geofence._cache.clear()
+    live_layers.aircraft_last_known.clear()
+    notified = []
+
+    async def fake_record(**kw):
+        notified.append(kw)
+        return 1
+
+    async def no_markers(*a, **k):
+        return []
+
+    monkeypatch.setattr(ingestor, "async_session", _InsertSession)
+    monkeypatch.setattr(geofence, "record_and_notify", fake_record)
+    monkeypatch.setattr("tracks.recent_markers", no_markers)
+
+    assert asyncio.run(ingest_fire_row(make_fire_msg())) is True
+    assert len(notified) == 1
+    assert notified[0]["source_kind"] == "firms"
+    assert notified[0]["lat"] == 39.45678
+    assert notified[0]["lon"] == -121.12345
+
+
+def test_ingest_fire_row_correlates_from_hypertable_when_last_known_empty(monkeypatch):
+    """FIRMS ingester has no ADS-B last-known; still correlate from aircraft_positions."""
+    import geofence
+    import ingestor
+    import live_layers
+
+    geofence._cache.clear()
+    live_layers.aircraft_last_known.clear()
+    correlated = []
+
+    async def fake_record(**kw):
+        return 0
+
+    async def fake_recent(kind, limit=2000):
+        assert kind == "aircraft"
+        return [{
+            "id": "acf001",
+            "lat": 39.45,
+            "lon": -121.12,
+            "extra": {"type": "AT802"},
+        }]
+
+    async def fake_corr(fires, aircraft):
+        correlated.append((fires, aircraft))
+        return aircraft
+
+    monkeypatch.setattr(ingestor, "async_session", _InsertSession)
+    monkeypatch.setattr(geofence, "record_and_notify", fake_record)
+    monkeypatch.setattr("tracks.recent_markers", fake_recent)
+    monkeypatch.setattr("fire_aircraft.correlate_and_notify", fake_corr)
+
+    assert asyncio.run(ingest_fire_row(make_fire_msg())) is True
+    assert len(correlated) == 1
+    assert correlated[0][1][0]["id"] == "acf001"
+    assert correlated[0][0][0]["lat"] == 39.45678

tests/test_fire_sources.py

@@ -83,6 +83,8 @@ def test_ingest_fires_idles_without_map_key(monkeypatch):
 
 def test_ingest_fires_uses_keystore_key(monkeypatch):
     # Key saved via the dashboard Keys page (Postgres) is picked up.
+    from upstream_cache import firms_cache
+    firms_cache.clear()
     monkeypatch.setenv("FIRMS_MAP_KEY", "")
     monkeypatch.setattr(
         "fire_sources.get_api_key",

tests/test_geofence.py

@@ -0,0 +1,136 @@
+"""Geofence hit detection and WS alert routing (no Redis)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import geofence
+from geofence import (
+    matching_geofences,
+    point_in_geojson,
+    validate_polygon_geojson,
+)
+from ws_manager import ConnectionManager
+
+
+NC_BOX = {
+    "type": "Polygon",
+    "coordinates": [[
+        [-80.0, 35.0],
+        [-78.0, 35.0],
+        [-78.0, 36.0],
+        [-80.0, 36.0],
+        [-80.0, 35.0],
+    ]],
+}
+
+
+def test_point_inside_polygon_hits():
+    assert point_in_geojson(-79.0, 35.5, NC_BOX) is True
+
+
+def test_point_outside_polygon_misses():
+    assert point_in_geojson(-122.4, 37.7, NC_BOX) is False
+
+
+def test_validate_rejects_non_polygon():
+    try:
+        validate_polygon_geojson({"type": "Point", "coordinates": [-79.0, 35.5]})
+        assert False, "expected ValueError"
+    except ValueError:
+        pass
+
+
+def test_matching_geofences_only_active_hits():
+    fences = [
+        {"id": "a", "name": "NC", "active": True, "geojson": NC_BOX},
+        {"id": "b", "name": "off", "active": False, "geojson": NC_BOX},
+    ]
+    hits = matching_geofences(-79.0, 35.5, fences)
+    assert [h["id"] for h in hits] == ["a"]
+    assert matching_geofences(-122.4, 37.7, fences) == []
+
+
+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("sf", (-123.0, 37.0, -121.0, 38.0))
+
+    async def run():
+        payload = {
+            "geofence_id": "a",
+            "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"
+        assert msg["payload"]["entity_id"] == "366123456"
+        assert q_sf.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
+
+    geofence._cache.clear()
+    geofence._recent_hits.clear()
+    st_called = []
+
+    async def fake_st(lon, lat):
+        st_called.append((lon, lat))
+        return [{
+            "id": "11111111-1111-1111-1111-111111111111",
+            "name": "NC",
+            "geojson": NC_BOX,
+            "active": True,
+        }]
+
+    monkeypatch.setattr(geofence, "st_intersects", fake_st)
+
+    executed: list = []
+
+    class FakeSession:
+        async def execute(self, stmt, params=None):
+            executed.append(params or {})
+            return None
+
+        async def commit(self):
+            executed.append("commit")
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+    monkeypatch.setattr(geofence, "async_session", FakeSession)
+
+    async def run():
+        from ws_manager import manager
+        q = manager.register("nc")
+        manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
+        n = await geofence.record_and_notify(
+            source_kind="firms", entity_id="35.5,-79.0,N",
+            lat=35.5, lon=-79.0, payload={"satellite": "N"},
+        )
+        msg = None if q.empty() else q.get_nowait()
+        manager.unregister("nc")
+        return n, msg
+
+    n, msg = asyncio.run(run())
+    assert st_called == [(-79.0, 35.5)]
+    assert n == 1
+    assert msg["type"] == "geofence_alert"
+    assert msg["payload"]["source_kind"] == "firms"
+    inserts = [p for p in executed if isinstance(p, dict)]
+    assert inserts and inserts[0]["source_kind"] == "firms"
+    assert "commit" in executed

tests/test_phase1_guardrails.py

@@ -0,0 +1,33 @@
+"""Phase 1 guardrails: compose limits, 500ms debounce, no extra brokers."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def test_compose_memory_limits_and_shared_buffers():
+    text = (ROOT / "docker-compose.yml").read_text()
+    assert "shared_buffers=2GB" in text
+    assert "shared_preload_libraries=timescaledb" in text
+    assert "memory: 3G" in text or "memory: 3GB" in text
+    assert "memory: 2G" in text or "memory: 2GB" in text
+
+
+def test_map_moveend_debounced_500ms():
+    html = (ROOT / "app" / "static" / "index.html").read_text()
+    assert "map.on('moveend'" in html
+    # Existing 300ms debounce must be 500ms so pans don't spam bbox POSTs/WS.
+    assert "}, 500);" in html
+    assert "}, 300);" not in html.split("map.on('moveend'")[1][:800]
+
+
+def test_no_redis_kafka_celery():
+    req = (ROOT / "app" / "requirements.txt").read_text().lower()
+    compose = (ROOT / "docker-compose.yml").read_text().lower()
+    for blob in (req, compose):
+        assert "redis" not in blob
+        assert "kafka" not in blob
+        assert "celery" not in blob
+    assert "cachetools" in req

tests/test_phase2_api.py

@@ -0,0 +1,51 @@
+"""Phase 2 API contracts (no Redis; geofences ≠ /api/alerts)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+
+from main import app
+
+BASE = "http://test"
+
+
+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_geofence_post_rejects_point():
+    resp = asyncio.run(_req(
+        "POST", "/api/geofences",
+        json={"name": "nope", "geojson": {"type": "Point", "coordinates": [-79, 35]}},
+    ))
+    assert resp.status_code == 422
+
+
+def test_geofences_list_does_not_collide_with_alerts():
+    resp = asyncio.run(_req("GET", "/api/geofences"))
+    assert resp.status_code == 200
+    assert isinstance(resp.json(), list)
+
+
+def test_aircraft_timestamp_returns_list_not_live_error():
+    resp = asyncio.run(_req(
+        "GET", "/api/aircraft",
+        params={"bbox": "-80,35,-78,36", "timestamp": "2026-08-28T12:04:00Z"},
+    ))
+    assert resp.status_code == 200
+    assert isinstance(resp.json(), list)
+
+
+def test_tracks_range_has_bounds():
+    body = asyncio.run(_req("GET", "/api/tracks/range")).json()
+    assert "min" in body and "max" in body
+
+
+def test_fire_aircraft_reload_endpoint():
+    resp = asyncio.run(_req("GET", "/api/fire-aircraft"))
+    assert resp.status_code == 200
+    assert isinstance(resp.json(), list)

tests/test_tracks.py

@@ -0,0 +1,50 @@
+"""1-minute downsampled tracks for DVR playback."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from tracks import (
+    TRACK_BUCKET,
+    downsample_tracks,
+    minute_bucket,
+    positions_at_timestamp,
+)
+
+
+def _ts(h, m, s=0):
+    return datetime(2026, 8, 28, h, m, s, tzinfo=timezone.utc)
+
+
+def test_minute_bucket_floors_seconds():
+    assert minute_bucket(_ts(12, 4, 47)) == _ts(12, 4, 0)
+    assert TRACK_BUCKET == "1 minute"
+
+
+def test_downsample_keeps_last_sample_per_id_per_minute():
+    rows = [
+        {"id": "a1", "ts": _ts(12, 4, 10), "lat": 35.0, "lon": -79.0},
+        {"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1},
+        {"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2},
+        {"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0},
+    ]
+    out = downsample_tracks(rows)
+    by = {(r["id"], r["bucket"]): r for r in out}
+    assert by[("a1", _ts(12, 4))]["lat"] == 35.1
+    assert by[("a1", _ts(12, 5))]["lat"] == 35.2
+    assert by[("b2", _ts(12, 4))]["lat"] == 36.0
+    assert len(out) == 3
+
+
+def test_positions_at_timestamp_uses_that_minute_window():
+    rows = [
+        {"id": "a1", "ts": _ts(12, 4, 50), "lat": 35.1, "lon": -79.1, "heading": 90, "speed": 10, "label": "A1"},
+        {"id": "a1", "ts": _ts(12, 5, 5), "lat": 35.2, "lon": -79.2, "heading": 91, "speed": 11, "label": "A1"},
+        {"id": "b2", "ts": _ts(12, 4, 20), "lat": 36.0, "lon": -80.0, "heading": 0, "speed": 0, "label": "B2"},
+    ]
+    at = positions_at_timestamp(rows, _ts(12, 4, 59))
+    ids = {p["id"]: p for p in at}
+    assert ids["a1"]["lat"] == 35.1
+    assert ids["b2"]["lat"] == 36.0
+    later = positions_at_timestamp(rows, _ts(12, 5, 30))
+    assert {p["id"]: p["lat"] for p in later} == {"a1": 35.2}

tests/test_upstream_cache.py

@@ -0,0 +1,104 @@
+"""FIRMS / RSS in-process TTLCache — no Redis."""
+
+from __future__ import annotations
+
+import asyncio
+
+from cachetools import TTLCache
+
+from fire_sources import ingest_fires
+from sources import ingest_rss_feed
+from test_fire_sources import SAMPLE_CSV, _async_return
+from upstream_cache import firms_cache, rss_cache
+
+
+def test_firms_and_rss_caches_are_ttlcache():
+    assert isinstance(firms_cache, TTLCache)
+    assert firms_cache.maxsize == 100
+    assert firms_cache.ttl == 300
+    assert isinstance(rss_cache, TTLCache)
+    assert rss_cache.maxsize == 100
+    assert 60 <= rss_cache.ttl <= 300
+
+
+def test_ingest_fires_hits_http_once_within_ttl(monkeypatch):
+    firms_cache.clear()
+    monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32)
+    monkeypatch.setenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT")
+    # fire_sources already imported FIRMS_DATASETS — patch the module attr
+    monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"])
+
+    hits = {"n": 0}
+
+    class FakeResp:
+        text = SAMPLE_CSV
+
+        def raise_for_status(self):
+            pass
+
+    class FakeClient:
+        def __init__(self, **kw):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def get(self, url):
+            hits["n"] += 1
+            return FakeResp()
+
+    monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient)
+
+    async def fake_publish(points):
+        return len(points)
+
+    monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish)
+
+    assert asyncio.run(ingest_fires()) == 5
+    assert asyncio.run(ingest_fires()) == 5
+    assert hits["n"] == 1
+
+
+def test_ingest_rss_hits_http_once_within_ttl(monkeypatch):
+    rss_cache.clear()
+    hits = {"n": 0}
+
+    class FakeResp:
+        text = """<?xml version="1.0"?>
+        <rss version="2.0"><channel><title>t</title>
+        <item><title>hello</title><link>http://x.example/1</link>
+        <description>body</description></item></channel></rss>"""
+
+        def raise_for_status(self):
+            pass
+
+    class FakeClient:
+        def __init__(self, **kw):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def get(self, url):
+            hits["n"] += 1
+            return FakeResp()
+
+    monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient)
+
+    published = []
+
+    async def fake_publish(subject, event):
+        published.append((subject, event))
+
+    monkeypatch.setattr("sources.publish_event", fake_publish)
+
+    assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1
+    assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1
+    assert hits["n"] == 1
+    assert published[0][0] == "events.rss"

tests/test_ws_manager.py

@@ -0,0 +1,91 @@
+"""Viewport-filtered in-memory pub/sub — no Redis."""
+
+from __future__ import annotations
+
+import asyncio
+
+from ws_manager import ConnectionManager, point_in_bbox
+
+
+def test_point_in_bbox_inclusive():
+    box = (-80.0, 35.0, -78.0, 36.0)
+    assert point_in_bbox(-79.0, 35.5, box) is True
+    assert point_in_bbox(-80.0, 35.0, box) is True
+    assert point_in_bbox(-77.0, 35.5, box) is False
+    assert point_in_bbox(-79.0, 34.0, box) is False
+
+
+def test_missing_viewport_does_not_get_firehose():
+    mgr = ConnectionManager()
+    q = mgr.register("tailnet-a")
+
+    async def run():
+        n = await mgr.publish_point("ais", {"id": "mmsi-1"}, lat=35.5, lon=-79.0)
+        assert n == 0
+        assert q.empty()
+
+    asyncio.run(run())
+
+
+def test_fanout_only_to_intersecting_viewports():
+    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("sf", (-123.0, 37.0, -121.0, 38.0))
+
+    async def run():
+        n = await mgr.publish_point(
+            "ais", {"id": "mmsi-1", "lat": 35.5, "lon": -79.0},
+            lat=35.5, lon=-79.0,
+        )
+        assert n == 1
+        msg = q_nc.get_nowait()
+        assert msg["type"] == "ais"
+        assert msg["payload"]["id"] == "mmsi-1"
+        assert q_sf.empty()
+
+        n2 = await mgr.publish_point(
+            "adsb", {"id": "a1b2c3"}, lat=37.7, lon=-122.4,
+        )
+        assert n2 == 1
+        msg2 = q_sf.get_nowait()
+        assert msg2["type"] == "adsb"
+        assert q_nc.empty()
+
+    asyncio.run(run())
+
+
+def test_disconnect_stops_fanout():
+    mgr = ConnectionManager()
+    q = mgr.register("gone")
+    mgr.set_viewport("gone", (-180.0, -90.0, 180.0, 90.0))
+    mgr.unregister("gone")
+
+    async def run():
+        n = await mgr.publish_point("ais", {"id": "x"}, lat=0.0, lon=0.0)
+        assert n == 0
+        assert q.empty()
+
+    asyncio.run(run())
+
+
+def test_upsert_vessel_fans_out_to_intersecting_client(monkeypatch):
+    from live_layers import upsert_vessel, vessel_last_known
+    from ws_manager import manager
+
+    vessel_last_known.clear()
+    manager._queues.clear()
+    manager._viewports.clear()
+    q = manager.register("nc")
+    manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
+
+    async def run():
+        await upsert_vessel({"id": "366123456", "lat": 35.2, "lon": -79.1, "label": "TEST"})
+        msg = q.get_nowait()
+        assert msg["type"] == "ais"
+        assert msg["payload"]["id"] == "366123456"
+
+    asyncio.run(run())
+    manager.unregister("nc")
+    vessel_last_known.clear()