feat: AIS stream follows the map viewport

The Vessels layer now retunes the server-side AISStream subscription to the
client viewport instead of a static AISSTREAM_BBOX. The frontend POSTs its
quantized viewport box to /api/vessels/subscribe on moveend; the ais_stream
worker coalesces and applies it at the service's 1 subscription/s cap, then
last-known positions for the new area arrive within a couple of seconds (the
frontend does one follow-up fetch after retuning). Bounds the in-memory
vessel store across regions. Key stays server-side.

app/ais_stream.py

@@ -14,6 +14,7 @@ import json
 import logging
 import os
 import random
+import time
 
 from config import AISSTREAM_API_KEY, AISSTREAM_BBOX
 from keystore import get_api_key
@@ -29,6 +30,36 @@ FILTER_TYPES = [
     "ShipStaticData",
 ]
 
+# ── Viewport-following ─────────────────────────────────────────────────────
+# The frontend POSTs its current viewport box to /api/vessels/subscribe; the
+# worker retunes the AISStream subscription to it (throttled to 1/s, the
+# service's subscription-update cap). ``None`` keeps the env AISSTREAM_BBOX
+# default. Last writer wins; the key never reaches the browser.
+_desired_boxes: list[list[list[float]]] | None = None
+_bbox_guard = asyncio.Lock()
+
+
+async def request_viewport_bbox(
+    minlon: float, minlat: float, maxlon: float, maxlat: float
+) -> None:
+    """Retune the live subscription to a viewport box (lon/lat input order)."""
+    global _desired_boxes
+    box = [[minlat, minlon], [maxlat, maxlon]]  # AISStream wants [lat, lon] corners
+    async with _bbox_guard:
+        _desired_boxes = [box]
+
+
+async def reset_viewport_bbox() -> None:
+    """Fall back to the env AISSTREAM_BBOX default."""
+    global _desired_boxes
+    async with _bbox_guard:
+        _desired_boxes = None
+
+
+async def _take_desired_boxes() -> list[list[list[float]]] | None:
+    async with _bbox_guard:
+        return _desired_boxes
+
 
 def _parse_boxes(raw: str) -> list[list[list[float]]]:
     """Env format: minlat,minlon,maxlat,maxlon[; ...]. AIS wants [[lat,lon],[lat,lon]]."""
@@ -65,7 +96,7 @@ async def run_ais_worker() -> None:
             )
             await asyncio.sleep(60)
             continue
-        boxes = _parse_boxes(AISSTREAM_BBOX)
+        boxes = await _take_desired_boxes() or _parse_boxes(AISSTREAM_BBOX)
         try:
             async with websockets.connect(
                 WS_URL,
@@ -82,7 +113,28 @@ async def run_ais_worker() -> None:
                 await ws.send(json.dumps(sub))
                 logger.info("AISStream subscribed (%d bbox(es))", len(boxes))
                 backoff = 2.0
-                async for raw in ws:
+                last_submit = time.monotonic()
+                while True:
+                    # Follow the client viewport: coalesce to the latest request
+                    # and honor AISStream's 1 subscription-update/s cap.
+                    desired = await _take_desired_boxes()
+                    if (
+                        desired is not None
+                        and desired != boxes
+                        and time.monotonic() - last_submit >= 1.0
+                    ):
+                        boxes = desired
+                        sub["BoundingBoxes"] = boxes
+                        await ws.send(json.dumps(sub))
+                        last_submit = time.monotonic()
+                        logger.info(
+                            "AISStream re-subscribed to viewport (%d bbox)",
+                            len(boxes),
+                        )
+                    try:
+                        raw = await asyncio.wait_for(ws.recv(), timeout=0.25)
+                    except asyncio.TimeoutError:
+                        continue
                     if isinstance(raw, bytes):
                         raw = raw.decode("utf-8", errors="replace")
                     try:

app/live_layers.py

@@ -66,6 +66,9 @@ _QUANT = 0.25  # degrees — pan jitter inside a cell reuses the TTL entry
 # Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker.
 vessel_last_known: dict[str, dict] = {}
 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
 
 
 def overlay_catalog() -> dict:
@@ -620,6 +623,14 @@ async def upsert_vessel(marker: dict) -> None:
             ),
             "seen_at": datetime.now(timezone.utc).isoformat(),
         }
+        if len(vessel_last_known) > _MAX_VESSELS:
+            excess = len(vessel_last_known) - int(_MAX_VESSELS * 0.9)
+            oldest = sorted(
+                vessel_last_known,
+                key=lambda k: vessel_last_known[k].get("seen_at", ""),
+            )[:excess]
+            for k in oldest:
+                vessel_last_known.pop(k, None)
 
 
 def _wfigs_params(bbox: str | None, *, offset_m: float = 250.0) -> dict:

app/main.py

@@ -40,7 +40,7 @@ from schemas import (
     FeedSourceCreate, FeedSourceOut,
     KeyOut, KeyValueIn,
     SearchResult, SentimentSummary, SourceType,
-    SearchQuery, TimelinePoint,
+    SearchQuery, TimelinePoint, VesselBboxUpdate,
 )
 from ingestor import ingest_event, fetch_and_process
 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
@@ -1176,6 +1176,34 @@ async def list_vessels(
         raise HTTPException(422, str(exc)) from exc
 
 
+@app.post("/api/vessels/subscribe")
+async def vessels_subscribe(payload: VesselBboxUpdate):
+    """Retune the server-side AISStream subscription to the client viewport.
+
+    The stream follows the map: after a moveend the frontend posts its bbox,
+    the worker re-subscribes (≤1/s upstream), and last-known vessels for the
+    new area start arriving within a second or two. Empty/null bbox resets to
+    the env AISSTREAM_BBOX default. The API key never reaches the browser.
+    """
+    raw = (payload.bbox or "").strip()
+    if not raw:
+        from ais_stream import reset_viewport_bbox
+        await reset_viewport_bbox()
+        return {"ok": True, "bbox": None}
+    try:
+        minlon, minlat, maxlon, maxlat = parse_bbox(raw)
+    except ValueError as exc:
+        raise HTTPException(422, str(exc)) from exc
+    if not (-180 <= minlon <= 180 and -180 <= maxlon <= 180
+            and -90 <= minlat <= 90 and -90 <= maxlat <= 90):
+        raise HTTPException(422, "bbox coordinates out of range")
+    if minlon >= maxlon or minlat >= maxlat:
+        raise HTTPException(422, "bbox must have min < max")
+    from ais_stream import request_viewport_bbox
+    await request_viewport_bbox(minlon, minlat, maxlon, maxlat)
+    return {"ok": True, "bbox": raw}
+
+
 @app.get("/api/fire-incidents")
 async def list_fire_incidents(
     bbox: str | None = Query(None),

app/schemas.py

@@ -300,3 +300,13 @@ class DashboardSummary(BaseModel):
     sentiment: SentimentSummary
     top_entities: list[EntityOut]
 
+
+class VesselBboxUpdate(BaseModel):
+    """Retune the server-side AISStream subscription to a client viewport box.
+
+    ``bbox`` is "minlon,minlat,maxlon,maxlat" (Leaflet order). ``None``/empty
+    resets to the env AISSTREAM_BBOX default.
+    """
+
+    bbox: str | None = None
+

app/static/index.html

@@ -1652,6 +1652,8 @@ let overlayReq = {ac:0, trains:0, vessels:0, alerts:0, perim:0, incidents:0, sto
 let moveDebounce = null;
 let overlayAbort = null;
 let lastCell = '';
+let lastVesselSubBox = '';      // last viewport box sent to the AIS stream
+let vesselRefollowTimer = null; // one follow-up fetch after a retune
 function bboxCell() {
     if (!map) return '';
     return currentBBox().split(',').map(n => Number(n).toFixed(2)).join(',') + '@' + map.getZoom();
@@ -1917,6 +1919,17 @@ function currentBBox() {
     }
     return `${west.toFixed(4)},${south.toFixed(4)},${east.toFixed(4)},${north.toFixed(4)}`;
 }
+function quantizeBBox(bbox, decimals = 1) {
+    // Round outward (west/south floor, east/north ceil) so the box always
+    // covers the viewport; a coarse box keeps AIS stream retunes sparse.
+    const [w, s, e, n] = bbox.split(',').map(Number);
+    const p = Math.pow(10, decimals);
+    const wq = Math.floor(w * p) / p;
+    const sq = Math.floor(s * p) / p;
+    const eq = Math.ceil(e * p) / p;
+    const nq = Math.ceil(n * p) / p;
+    return [wq, sq, eq, nq].map(v => v.toFixed(decimals)).join(',');
+}
 function toggleLayerPanel() {
     const panel = document.getElementById('layer-panel');
     panel.classList.toggle('collapsed');
@@ -2576,8 +2589,13 @@ async function loadTrains() {
 }
 async function toggleVessels() {
     vesselsOn = document.getElementById('lp-vessels-on').checked;
-    if (vesselsOn) await loadVessels();
-    else vesselsGroup = dropLayer(vesselsGroup);
+    if (vesselsOn) {
+        lastVesselSubBox = '';
+        await loadVessels();
+    } else {
+        vesselsGroup = dropLayer(vesselsGroup);
+        clearTimeout(vesselRefollowTimer);
+    }
 }
 async function loadVessels() {
     if (!map) return;
@@ -2586,8 +2604,25 @@ async function loadVessels() {
         return;
     }
     const req = ++overlayReq.vessels;
+    const bb = currentBBox();
+    // Retune the server-side AIS stream to this viewport. Skip whole-world
+    // boxes (zoom out / dateline wrap) so the free tier is not flooded; the
+    // worker coalesces retunes and caps them at 1/s upstream.
+    const [bw, bs, be, bn] = bb.split(',').map(Number);
+    const isWorld = (be - bw) >= 300 || bb.startsWith('-180.0000,-85.0000,180.0000,85.0000');
+    const subKey = isWorld ? '' : quantizeBBox(bb);
+    let retuned = false;
+    if (subKey && subKey !== lastVesselSubBox) {
+        lastVesselSubBox = subKey;
+        retuned = true;
+        fetch(`${API}/api/vessels/subscribe`, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ bbox: subKey })
+        }).catch(() => {});
+    }
     try {
-        const r = await overlayFetch(`${API}/api/vessels?bbox=${currentBBox()}`);
+        const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}`);
         const pts = await r.json();
         if (req !== overlayReq.vessels) return;
         vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => {
@@ -2596,6 +2631,11 @@ async function loadVessels() {
         }, true, 'vessel');
         document.getElementById('lp-vessels-count').textContent = (pts.length || 0).toLocaleString();
         addExtraAttrib('AISStream');
+        // First positions for a new zone arrive ~1-3s after the retune.
+        if (retuned) {
+            clearTimeout(vesselRefollowTimer);
+            vesselRefollowTimer = setTimeout(() => { if (vesselsOn) loadVessels(); }, 2200);
+        }
     } catch (e) {
         if (isAbort(e)) return;
         console.error('Vessels load failed', e);

tests/test_api_live_layers.py

@@ -16,6 +16,12 @@ async def _get(path: str) -> httpx.Response:
         return await client.get(path)
 
 
+async def _post(path: str, payload: dict | None) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.post(path, json=payload)
+
+
 def test_map_layers_includes_overlays():
     body = asyncio.run(_get("/api/map/layers")).json()
     assert "layers" in body
@@ -37,3 +43,35 @@ def test_vessels_empty_without_ais_key():
     assert resp.status_code == 200
     assert resp.json() == []
     assert "max-age" in (resp.headers.get("cache-control") or "").lower()
+
+
+def _desired_boxes():
+    import asyncio as _a
+    from ais_stream import _take_desired_boxes
+    return _a.run(_take_desired_boxes())
+
+
+def test_vessels_subscribe_sets_viewport_box():
+    assert _desired_boxes() is None
+    resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": "-70,40,-60,45"}))
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["ok"] is True and body["bbox"] == "-70,40,-60,45"
+    # AISStream corner order: [[lat, lon], [lat, lon]] (southwest, northeast).
+    assert _desired_boxes() == [[[40.0, -70.0], [45.0, -60.0]]]
+
+
+def test_vessels_subscribe_empty_resets():
+    assert asyncio.run(_post("/api/vessels/subscribe", {"bbox": "-70,40,-60,45"})).status_code == 200
+    resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": ""}))
+    assert resp.status_code == 200
+    assert resp.json()["bbox"] is None
+    assert _desired_boxes() is None
+
+
+def test_vessels_subscribe_rejects_bad_bbox():
+    for bad in ("1,2,3", "a,b,c,d", "20,30,10,40", "0,0,0,200"):
+        resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": bad}))
+        assert resp.status_code == 422, bad
+    # Explicit null bbox is the "reset to env default" path (still 200).
+    assert asyncio.run(_post("/api/vessels/subscribe", {"bbox": None})).status_code == 200