Merge pull request 'feat(vessels): AISStream (US) + VesselAPI (Hormuz) as two active providers' (#19) from feat/vesselapi-ais-poller into master

Reviewed-on: #19

.env.example

@@ -59,6 +59,24 @@ FIRMS_INTERVAL=900
 # Set to 0 to disable the fire loop entirely.
 INGEST_FIRES=1
 
+# ── VesselAPI (commercial REST AIS — Hormuz, 5×/day, 150 calls/mo cap) ─────
+# Independent of AISStream (open/shared live US-coast WebSocket). Both stay
+# on when their keys are set; missing one never disables the other.
+# Prefer pasting VESSELAPI_API_KEY on the dashboard Keys page.
+# The poller idles when the key is unset. Never called from map pans
+# (GET /api/vessels serves the shared last-known cache only).
+VESSELAPI_API_KEY=
+# Bounding box(es) as minlat,minlon,maxlat,maxlon (lat/lon order). Semicolon-
+# separated for multiple boxes. Default = Strait of Hormuz (span 3.6 ≤ 4° cap).
+VESSELAPI_BBOX=25.5,55.4,27.3,57.2
+# Poll cadence in seconds (17280 = 4.8h → 5 polls/day = 150/mo).
+VESSELAPI_INTERVAL=17280
+# Local hard cap on successful 2xx calls per UTC day (persisted in Postgres).
+VESSELAPI_MAX_CALLS_PER_DAY=5
+# 1 = run the poller inside the dashboard process (default); ingester off.
+VESSELAPI_IN_APP=1
+VESSELAPI_IN_INGEST=0
+
 # ── API keys (managed from the dashboard UI) ──────────────────────────────
 # Keys such as NOUS_API_KEY and TELEGRAM_TOKEN are stored in the Postgres
 # `api_keys` table and managed from the dashboard's "Keys" tab

app/config.py

@@ -81,3 +81,19 @@ AISSTREAM_BBOX = os.getenv("AISSTREAM_BBOX", "24,-125,50,-66")
 # without the ingest profile). Set 0 if the ingester owns the only connection.
 AISSTREAM_IN_APP = os.getenv("AISSTREAM_IN_APP", "1").lower() in ("1", "true", "yes")
 AISSTREAM_IN_INGEST = os.getenv("AISSTREAM_IN_INGEST", "0").lower() in ("1", "true", "yes")
+
+# VesselAPI (quota-capped REST AIS poller — free tier 150 calls/mo).
+# AISStream keeps US coasts; VesselAPI fills the Middle East blind spot. The
+# poller idles when VESSELAPI_API_KEY is unset (never from GET /api/vessels).
+VESSELAPI_API_KEY = os.getenv("VESSELAPI_API_KEY", "")
+# Bounding box(es) as minlat,minlon,maxlat,maxlon — note lat/lon order (same as
+# AISSTREAM_BBOX). Semicolon-separated for multiple boxes. Default: Strait of
+# Hormuz (|dLat|+|dLon| = 3.6 ≤ 4° span cap). VesselAPI 400s any box over 4°.
+VESSELAPI_BBOX = os.getenv("VESSELAPI_BBOX", "25.5,55.4,27.3,57.2")
+# Poll cadence in seconds. 17280 = 4.8h → 5 polls/day (150/mo free tier).
+VESSELAPI_INTERVAL = int(os.getenv("VESSELAPI_INTERVAL", "17280"))
+# Local hard cap on successful 2xx calls per UTC day (persisted in Postgres).
+VESSELAPI_MAX_CALLS_PER_DAY = int(os.getenv("VESSELAPI_MAX_CALLS_PER_DAY", "5"))
+# Run the VesselAPI poller inside the dashboard process (default on, like AIS).
+VESSELAPI_IN_APP = os.getenv("VESSELAPI_IN_APP", "1").lower() in ("1", "true", "yes")
+VESSELAPI_IN_INGEST = os.getenv("VESSELAPI_IN_INGEST", "0").lower() in ("1", "true", "yes")

app/keystore.py

@@ -68,10 +68,15 @@ KEY_REGISTRY: dict[str, dict] = {
         "example": "123456789:AA… (bot token from @BotFather)",
     },
     "AISSTREAM_API_KEY": {
-        "description": "AISStream WebSocket key — live vessel positions (server-side only).",
+        "description": "AISStream (open/shared) — live US-coast AIS. Server-side WebSocket only.",
         "pattern": r"^.{8,}$",
         "example": "key from https://aisstream.io/account (GitHub login)",
     },
+    "VESSELAPI_API_KEY": {
+        "description": "VesselAPI (commercial) — Strait of Hormuz AIS, 5×/day cache. Paste the Bearer token from dashboard.vesselapi.com. Not a US-coast feed.",
+        "pattern": r"^.{8,}$",
+        "example": "Bearer token from https://dashboard.vesselapi.com/",
+    },
     "OPENSKY_CLIENT_ID": {
         "description": "OpenSky OAuth client id — optional ADS-B fallback (unused until enabled).",
         "example": "client id from opensky-network.org account",

app/main.py

@@ -71,18 +71,26 @@ async def _lifespan(app: FastAPI):
         await refresh_cache()
     except Exception:
         pass
-    from config import AISSTREAM_IN_APP
+    from config import AISSTREAM_IN_APP, VESSELAPI_IN_APP
     ais_task = None
+    vesselapi_task = None
     adsb_task = None
     if AISSTREAM_IN_APP:
         from ais_stream import run_ais_worker
         ais_task = asyncio.create_task(run_ais_worker())
+    if VESSELAPI_IN_APP:
+        from vesselapi import run_vesselapi_worker
+        vesselapi_task = asyncio.create_task(run_vesselapi_worker())
     adsb_task = asyncio.create_task(_adsb_refresh_loop())
     yield
     if ais_task is not None:
         ais_task.cancel()
+    if vesselapi_task is not None:
+        vesselapi_task.cancel()
     if adsb_task is not None:
         adsb_task.cancel()
+    from vesselapi import close_client
+    await close_client()
     await close_http()
 
 
@@ -1464,7 +1472,12 @@ async def list_vessels(
     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."""
+    """AIS last-known — union of two independent providers.
+
+    AISStream (extra.src="aisstream", live US-coast WebSocket) and VesselAPI
+    (extra.src="vesselapi", Strait of Hormuz 5×/day poll) both upsert into the
+    same store. Empty without either key / until the first successful poll.
+    """
     if bbox:
         _parse_bbox_query(bbox)
     try:

app/run_ingester.py

@@ -24,7 +24,7 @@ import sys
 
 sys.path.insert(0, sys_path)
 
-from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST  # noqa: E402
+from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST, VESSELAPI_IN_INGEST  # noqa: E402
 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_eonet, ingest_cisa_kev  # noqa: E402
 from fire_sources import ingest_fires  # noqa: E402
 from ingestor import ingest_event, start_nats_consumer  # noqa: E402
@@ -134,6 +134,9 @@ async def main() -> None:
     if AISSTREAM_IN_INGEST:
         from ais_stream import run_ais_worker  # noqa: E402
         tasks.append(asyncio.create_task(run_ais_worker()))
+    if VESSELAPI_IN_INGEST:
+        from vesselapi import run_vesselapi_worker  # noqa: E402
+        tasks.append(asyncio.create_task(run_vesselapi_worker()))
     await asyncio.gather(producer_loop(), consumer_loop(), *tasks)
 
 

app/static/index.html

@@ -805,7 +805,7 @@
                                 <label class="lp-name"><input type="checkbox" id="lp-vessels-on" onchange="toggleVessels()"> <span class="lp-dot vessels"></span> Vessels · AIS</label>
                                 <span class="lp-count" id="lp-vessels-count">0</span>
                             </div>
-                            <div class="lp-note">Needs AISSTREAM_API_KEY. Magenta = military · gold = law/gov · teal = underway. Click for type / dest / IMO.</div>
+                            <div class="lp-note">Two feeds: AISStream (US live) and VesselAPI (Hormuz, 5×/day). Paste both keys on the Keys page. Magenta = military · gold = law/gov · teal = underway. Click for type / dest / IMO.</div>
                         </div>
                         <div class="lp-layer">
                             <div class="lp-row">

app/vesselapi.py

@@ -0,0 +1,445 @@
+"""VesselAPI REST poller — quota-capped AIS for the Middle East (free tier 150 calls/mo).
+
+VesselAPI and AISStream are two independent, first-class vessel providers —
+not a primary/fallback pair. AISStream (WebSocket) owns live US-coast AIS;
+VesselAPI (REST) covers the Strait of Hormuz (default box) where AISStream
+has no coverage. Missing one key never disables the other. This worker polls
+the REST ``GET /v1/location/vessels/bounding-box`` endpoint at most
+``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day
+and upserts the results into the shared ``vessel_last_known`` store.
+
+Idle (no crash) when VESSELAPI_API_KEY is unset. Never called from the GET
+/api/vessels path — map pans must not hit upstream. One request per poll,
+``pagination.limit=50``, never follow ``nextToken``, never send ``filter.sat``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import calendar
+import logging
+import os
+from datetime import date, datetime, timezone
+
+import httpx
+from sqlalchemy import Column, Date, DateTime, Integer, Table, func, select, text
+
+from config import (
+    OSINT_USER_AGENT,
+    VESSELAPI_API_KEY,
+    VESSELAPI_BBOX,
+    VESSELAPI_INTERVAL,
+    VESSELAPI_MAX_CALLS_PER_DAY,
+)
+from database import async_session, engine, metadata
+from live_layers import to_marker, upsert_vessel
+
+logger = logging.getLogger("osint.vesselapi")
+
+BASE_URL = "https://api.vesselapi.com/v1"
+ENDPOINT = f"{BASE_URL}/location/vessels/bounding-box"
+MAX_SPAN_DEG = 4.0  # |dLat| + |dLon| — VesselAPI 400s above this.
+PAGE_LIMIT = 50  # pagination.limit; never follow nextToken on the free tier.
+
+_client: httpx.AsyncClient | None = None
+_client_lock = asyncio.Lock()
+
+
+# ── Box parsing / span validation ─────────────────────────────────────────
+
+class BboxError(ValueError):
+    """A configured VesselAPI box violates the 4° span rule or is malformed."""
+
+
+def validate_bbox_span(
+    minlat: float, minlon: float, maxlat: float, maxlon: float,
+) -> None:
+    """Reject boxes VesselAPI would 400 on (span > 4°, bad order, bad range)."""
+    if not (-90 <= minlat <= 90 and -90 <= maxlat <= 90
+            and -180 <= minlon <= 180 and -180 <= maxlon <= 180):
+        raise BboxError("coordinates out of range")
+    if minlat >= maxlat or minlon >= maxlon:
+        raise BboxError("bbox must have min < max on both axes")
+    dlat = abs(maxlat - minlat)
+    dlon = abs(maxlon - minlon)
+    if dlat + dlon > MAX_SPAN_DEG:
+        raise BboxError(
+            f"span |dLat|+|dLon| = {dlat + dlon:.2f}° exceeds {MAX_SPAN_DEG}° cap"
+        )
+
+
+def parse_boxes(raw: str) -> list[tuple[float, float, float, float]]:
+    """Env format: ``minlat,minlon,maxlat,maxlon[; ...]`` (lat/lon order)."""
+    out: list[tuple[float, float, float, float]] = []
+    for chunk in (raw or "").split(";"):
+        parts = [p.strip() for p in chunk.split(",") if p.strip()]
+        if len(parts) != 4:
+            continue
+        try:
+            minlat = float(parts[0])
+            minlon = float(parts[1])
+            maxlat = float(parts[2])
+            maxlon = float(parts[3])
+        except ValueError:
+            continue
+        out.append((minlat, minlon, maxlat, maxlon))
+    return out
+
+
+def parse_boxes_validated(raw: str) -> list[tuple[float, float, float, float]]:
+    """Parse boxes, log + skip any that violate the span/order/range rules."""
+    valid: list[tuple[float, float, float, float]] = []
+    for box in parse_boxes(raw):
+        try:
+            validate_bbox_span(*box)
+            valid.append(box)
+        except BboxError as exc:
+            logger.warning("VesselAPI bbox %r skipped: %s", box, exc)
+    return valid
+
+
+# ── Position → marker transform ───────────────────────────────────────────
+
+def _f(value: object) -> float | None:
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _s(value: object) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def transform_vesselapi_position(obj: dict | None) -> dict | None:
+    """Map one VesselAPI position object to the shared marker contract.
+
+    Returns None for glitch rows, missing MMSI, or missing coordinates.
+    """
+    if not obj or not isinstance(obj, dict):
+        return None
+    if obj.get("suspected_glitch") is True:
+        return None
+    mmsi = obj.get("mmsi")
+    if mmsi is None:
+        return None
+    lat = _f(obj.get("latitude"))
+    lon = _f(obj.get("longitude"))
+    if lat is None or lon is None:
+        return None
+    mmsi_s = str(mmsi)
+    name = _s(obj.get("vessel_name") or obj.get("name"))
+    heading = _f(obj.get("heading"))
+    if heading is None:
+        heading = _f(obj.get("cog"))
+    sog = _f(obj.get("sog"))
+    extra: dict = {
+        "src": "vesselapi",
+        "mmsi": mmsi_s,
+        "cog": obj.get("cog"),
+        "sog": obj.get("sog"),
+        "navstat": obj.get("nav_status"),
+    }
+    imo = obj.get("imo")
+    if imo:
+        extra["imo"] = imo
+    dest = _s(obj.get("dest") or obj.get("destination"))
+    if dest:
+        extra["dest"] = dest
+    ts = obj.get("timestamp") or obj.get("processed_timestamp")
+    if ts:
+        extra["timestamp"] = ts
+    return to_marker(
+        mmsi_s, lat, lon,
+        heading=heading,
+        speed=sog,
+        label=name or mmsi_s,
+        extra=extra,
+    )
+
+
+def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
+    """Flatten a bounding-box response ``{vessels: [...]}`` to markers."""
+    if not payload or not isinstance(payload, dict):
+        return []
+    rows = payload.get("vessels") or []
+    out = []
+    for row in rows:
+        marker = transform_vesselapi_position(row)
+        if marker:
+            out.append(marker)
+    return out
+
+
+# ── Durable daily quota (Postgres, survives restarts) ─────────────────────
+# Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.
+
+vesselapi_quota = Table(
+    "vesselapi_quota",
+    metadata,
+    Column("day", Date, primary_key=True),
+    Column("calls", Integer, nullable=False, server_default="0"),
+    Column("remaining", Integer, nullable=True),
+    Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+)
+
+_CREATE_QUOTA_SQL = text(
+    """
+    CREATE TABLE IF NOT EXISTS vesselapi_quota (
+        day        DATE PRIMARY KEY,
+        calls      INTEGER NOT NULL DEFAULT 0,
+        remaining  INTEGER,
+        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+    """
+)
+
+_quota_lock = asyncio.Lock()
+_quota_ensured = False
+
+
+async def ensure_quota_table() -> None:
+    global _quota_ensured
+    if _quota_ensured:
+        return
+    async with _quota_lock:
+        if _quota_ensured:
+            return
+        async with engine.begin() as conn:
+            await conn.execute(_CREATE_QUOTA_SQL)
+        _quota_ensured = True
+
+
+class PgQuotaStore:
+    """Postgres-backed daily call counter. Injected for tests."""
+
+    async def calls_today(self, day: date) -> int:
+        await ensure_quota_table()
+        async with async_session() as session:
+            row = (await session.execute(
+                select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
+            )).scalar()
+        return int(row) if row else 0
+
+    async def remaining_today(self, day: date) -> int | None:
+        await ensure_quota_table()
+        async with async_session() as session:
+            row = (await session.execute(
+                select(vesselapi_quota.c.remaining).where(vesselapi_quota.c.day == day)
+            )).scalar()
+        return int(row) if row is not None else None
+
+    async def bump(self, day: date, remaining: int | None) -> int:
+        await ensure_quota_table()
+        now = datetime.now(timezone.utc)
+        async with async_session() as session:
+            existing = (await session.execute(
+                select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
+            )).scalar()
+            if existing is None:
+                await session.execute(
+                    vesselapi_quota.insert().values(
+                        day=day, calls=1, remaining=remaining, updated_at=now,
+                    )
+                )
+            else:
+                await session.execute(
+                    vesselapi_quota.update()
+                    .where(vesselapi_quota.c.day == day)
+                    .values(
+                        calls=vesselapi_quota.c.calls + 1,
+                        remaining=remaining,
+                        updated_at=now,
+                    )
+                )
+            await session.commit()
+        return (int(existing) if existing else 0) + 1
+
+
+# ── Budget / scheduling (pure, unit-testable) ─────────────────────────────
+
+def days_left_in_month(now: datetime) -> int:
+    """UTC days remaining in the current month, inclusive of today."""
+    _, last = calendar.monthrange(now.year, now.month)
+    return last - now.day + 1
+
+
+def budget_allows(
+    calls_today: int,
+    remaining: int | None,
+    days_left: int,
+    max_per_day: int,
+) -> bool:
+    """True if another poll is permitted today.
+
+    Local hard cap: fewer than ``max_per_day`` successful calls today.
+    Monthly floor: if ``X-RateLimit-Remaining`` is known, keep at least
+    ``max_per_day * days_left`` in reserve for the rest of the month.
+    """
+    if calls_today >= max_per_day:
+        return False
+    if remaining is not None and remaining <= max_per_day * days_left:
+        return False
+    return True
+
+
+def choose_box(
+    boxes: list[tuple[float, float, float, float]],
+    calls_today: int,
+    max_per_day: int,
+) -> int:
+    """Index into ``boxes`` for the next poll.
+
+    Prefer refreshing the first (primary) box rather than spraying one call
+    across every region — round-robin only when the remaining daily budget is
+    enough to cover all boxes.
+    """
+    if len(boxes) <= 1:
+        return 0
+    budget_left = max_per_day - calls_today
+    if budget_left >= len(boxes):
+        return calls_today % len(boxes)
+    return 0
+
+
+# ── HTTP / poll ───────────────────────────────────────────────────────────
+
+async def _resolve_key() -> str:
+    from keystore import get_api_key
+    return (
+        os.getenv("VESSELAPI_API_KEY")
+        or VESSELAPI_API_KEY
+        or (await get_api_key("VESSELAPI_API_KEY"))
+        or ""
+    ).strip()
+
+
+async def _get_client() -> httpx.AsyncClient:
+    global _client
+    if _client is None:
+        async with _client_lock:
+            if _client is None:
+                _client = httpx.AsyncClient(
+                    timeout=httpx.Timeout(15.0, connect=5.0),
+                    follow_redirects=True,
+                    headers={"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"},
+                    limits=httpx.Limits(max_connections=1, max_keepalive_connections=1),
+                )
+    return _client
+
+
+async def close_client() -> None:
+    global _client
+    if _client is not None:
+        await _client.aclose()
+        _client = None
+
+
+def _int_header(value: str | None) -> int | None:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: str) -> bool:
+    """One quota-checked poll. Returns True if a successful 2xx was made.
+
+    Only successful 2xx responses count against the monthly quota; 4xx/5xx/429
+    are skipped without retry-storming (Retry-After respected by simply
+    sleeping the interval).
+    """
+    now = datetime.now(timezone.utc)
+    today = now.date()
+    calls = await store.calls_today(today)
+    remaining = await store.remaining_today(today)
+    days_left = days_left_in_month(now)
+    if not budget_allows(calls, remaining, days_left, VESSELAPI_MAX_CALLS_PER_DAY):
+        logger.info(
+            "VesselAPI quota reached (calls_today=%d, remaining=%s, days_left=%d) — skip poll",
+            calls, remaining, days_left,
+        )
+        return False
+
+    idx = choose_box(boxes, calls, VESSELAPI_MAX_CALLS_PER_DAY)
+    minlat, minlon, maxlat, maxlon = boxes[idx]
+    client = await _get_client()
+    params = {
+        "filter.latBottom": str(minlat),
+        "filter.latTop": str(maxlat),
+        "filter.lonLeft": str(minlon),
+        "filter.lonRight": str(maxlon),
+        "pagination.limit": str(PAGE_LIMIT),
+    }
+    headers = {"Authorization": f"Bearer {key}"}
+    try:
+        resp = await client.get(ENDPOINT, params=params, headers=headers)
+    except httpx.HTTPError as exc:
+        logger.warning("VesselAPI request failed: %s", exc)
+        return False
+
+    if resp.status_code == 429:
+        logger.warning(
+            "VesselAPI rate-limited (Retry-After=%s) — skip poll",
+            resp.headers.get("Retry-After"),
+        )
+        return False
+    if resp.status_code >= 400:
+        logger.warning("VesselAPI HTTP %d — not counted against quota", resp.status_code)
+        return False
+
+    # 2xx success — counts against the monthly quota.
+    remaining = _int_header(resp.headers.get("X-RateLimit-Remaining"))
+    calls = await store.bump(today, remaining)
+    try:
+        data = resp.json()
+    except ValueError:
+        logger.warning("VesselAPI 2xx with non-JSON body — counted but ignored")
+        return True
+    markers = transform_vesselapi_payload(data)
+    for m in markers:
+        await upsert_vessel(m)
+    logger.info(
+        "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
+        len(markers), remaining, calls,
+    )
+    return True
+
+
+# ── Worker loop ───────────────────────────────────────────────────────────
+
+async def run_vesselapi_worker(store: PgQuotaStore | None = None) -> None:
+    """Long-lived poll loop. Idle when the key is unset; never crashes the app."""
+    if store is None:
+        store = PgQuotaStore()
+    boxes = parse_boxes_validated(VESSELAPI_BBOX)
+    if not boxes:
+        logger.warning(
+            "VESSELAPI_BBOX has no valid boxes (span ≤ %.1f°) — poller idle", MAX_SPAN_DEG,
+        )
+    while True:
+        try:
+            if not boxes:
+                await asyncio.sleep(VESSELAPI_INTERVAL)
+                continue
+            key = await _resolve_key()
+            if not key:
+                logger.warning(
+                    "VESSELAPI_API_KEY not set — VesselAPI poller idle. "
+                    "Create a free key at https://dashboard.vesselapi.com/"
+                )
+                await asyncio.sleep(VESSELAPI_INTERVAL)
+                continue
+            await poll_once(store, boxes, key)
+        except asyncio.CancelledError:
+            raise
+        except Exception:  # noqa: BLE001 — keep the loop alive across transient failures
+            logger.exception("VesselAPI poll error")
+        await asyncio.sleep(VESSELAPI_INTERVAL)

docker-compose.yml

@@ -97,6 +97,11 @@ services:
       AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-}
       AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66}
       AISSTREAM_IN_INGEST: ${AISSTREAM_IN_INGEST:-0}
+      VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-}
+      VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2}
+      VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280}
+      VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5}
+      VESSELAPI_IN_INGEST: ${VESSELAPI_IN_INGEST:-0}
     command: ["python", "app/run_ingester.py"]
     entrypoint: ["python", "app/run_ingester.py"]
 
@@ -131,6 +136,11 @@ services:
       AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-}
       AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66}
       AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1}
+      VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-}
+      VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2}
+      VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280}
+      VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5}
+      VESSELAPI_IN_APP: ${VESSELAPI_IN_APP:-1}
     ports:
       - "127.0.0.1:8000:8000"
     deploy:

tests/test_vesselapi.py

@@ -0,0 +1,351 @@
+"""Unit tests for the VesselAPI poller (no network, no DB).
+
+Covers box span validation, position → marker mapping, glitch skipping,
+and the daily-quota gate (6th 2xx attempt skipped). ``poll_once`` is driven
+with an in-memory fake quota store + fake HTTP client; ``upsert_vessel``'s
+DB/WS side effects are monkeypatched to no-ops so markers can be asserted in
+``vessel_last_known``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+import vesselapi
+from vesselapi import (
+    BboxError,
+    budget_allows,
+    choose_box,
+    days_left_in_month,
+    parse_boxes,
+    parse_boxes_validated,
+    poll_once,
+    transform_vesselapi_payload,
+    transform_vesselapi_position,
+    validate_bbox_span,
+)
+from live_layers import vessel_last_known
+
+
+# ── Fake quota store (in-memory, injected) ────────────────────────────────
+
+class FakeStore:
+    def __init__(self, calls: int = 0, remaining: int | None = None):
+        self.calls = calls
+        self.remaining = remaining
+        self.bumps = 0
+
+    async def calls_today(self, day):
+        return self.calls
+
+    async def remaining_today(self, day):
+        return self.remaining
+
+    async def bump(self, day, remaining):
+        self.calls += 1
+        self.remaining = remaining
+        self.bumps += 1
+        return self.calls
+
+
+# ── Fake HTTP client ───────────────────────────────────────────────────────
+
+class FakeResp:
+    def __init__(self, status: int = 200, headers: dict | None = None, data: dict | None = None):
+        self.status_code = status
+        self.headers = headers or {}
+        self._data = data or {}
+
+    def json(self):
+        return self._data
+
+
+class FakeClient:
+    def __init__(self, *responses: FakeResp):
+        self.responses = list(responses)
+        self.calls: list[tuple] = []
+
+    async def get(self, url, params=None, headers=None):
+        self.calls.append((url, params, headers))
+        return self.responses.pop(0)
+
+    async def aclose(self):
+        pass
+
+
+def _sample_payload() -> dict:
+    return {
+        "vessels": [
+            {
+                "mmsi": 422050100,
+                "imo": 9321483,
+                "vessel_name": "HORMUZ STAR",
+                "latitude": 26.5,
+                "longitude": 56.3,
+                "cog": 88.0,
+                "sog": 12.4,
+                "heading": 90,
+                "nav_status": 0,
+                "timestamp": "2026-08-29T12:00:00Z",
+                "suspected_glitch": False,
+            },
+            {
+                "mmsi": 422050101,
+                "vessel_name": "GLITCHY",
+                "latitude": 26.6,
+                "longitude": 56.4,
+                "cog": 45.0,
+                "sog": 5.0,
+                "heading": 45,
+                "nav_status": 5,
+                "suspected_glitch": True,
+            },
+            {"mmsi": 422050102, "vessel_name": "NOFIX"},  # no coords → drop
+        ],
+        "nextToken": "deadbeef",
+    }
+
+
+# ── Box span / parsing ─────────────────────────────────────────────────────
+
+def test_validate_bbox_span_accepts_hormuz():
+    validate_bbox_span(25.5, 55.4, 27.3, 57.2)  # span 3.6 — no raise
+
+
+def test_validate_bbox_span_rejects_conus():
+    with pytest.raises(BboxError):
+        validate_bbox_span(24.0, -125.0, 50.0, -66.0)  # span 85
+
+
+def test_validate_bbox_span_rejects_marine_regions_gazetteer_box():
+    # 25.2732–27.3713 N, 55.1647–57.3419 E → span 4.28 > 4.0.
+    with pytest.raises(BboxError):
+        validate_bbox_span(25.2732, 55.1647, 27.3713, 57.3419)
+
+
+def test_validate_bbox_span_rejects_inverted_axes():
+    with pytest.raises(BboxError):
+        validate_bbox_span(27.0, 55.0, 25.0, 57.0)
+
+
+def test_parse_boxes_semicolon_and_skip_malformed():
+    boxes = parse_boxes("25.5,55.4,27.3,57.2; 10,20,11,21; garbage")
+    assert boxes == [(25.5, 55.4, 27.3, 57.2), (10.0, 20.0, 11.0, 21.0)]
+
+
+def test_parse_boxes_validated_skips_over_span():
+    # Second box is CONUS-sized → dropped, first kept.
+    valid = parse_boxes_validated("25.5,55.4,27.3,57.2;24,-125,50,-66")
+    assert valid == [(25.5, 55.4, 27.3, 57.2)]
+
+
+# ── Position → marker mapping ─────────────────────────────────────────────
+
+def test_transform_position_maps_shared_marker_contract():
+    m = transform_vesselapi_position({
+        "mmsi": 422050100, "imo": 9321483, "vessel_name": "HORMUZ STAR",
+        "latitude": 26.5, "longitude": 56.3, "heading": 90, "cog": 88.0,
+        "sog": 12.4, "nav_status": 0, "timestamp": "2026-08-29T12:00:00Z",
+        "suspected_glitch": False,
+    })
+    assert m is not None
+    assert m["id"] == "422050100"
+    assert m["lat"] == 26.5
+    assert m["lon"] == 56.3
+    assert m["label"] == "HORMUZ STAR"
+    assert m["heading"] == 90
+    assert m["speed"] == 12.4
+    assert m["extra"]["src"] == "vesselapi"
+    assert m["extra"]["mmsi"] == "422050100"
+    assert m["extra"]["imo"] == 9321483
+    assert m["extra"]["navstat"] == 0
+    assert m["extra"]["cog"] == 88.0
+    assert m["extra"]["sog"] == 12.4
+    assert m["extra"]["timestamp"] == "2026-08-29T12:00:00Z"
+
+
+def test_transform_position_heading_falls_back_to_cog():
+    m = transform_vesselapi_position({
+        "mmsi": 123456789, "vessel_name": "X", "latitude": 1.0, "longitude": 2.0,
+        "heading": None, "cog": 123.4, "sog": 5.0,
+    })
+    assert m["heading"] == 123.4
+
+
+def test_transform_position_skips_glitch():
+    assert transform_vesselapi_position({
+        "mmsi": 123456789, "latitude": 1.0, "longitude": 2.0,
+        "suspected_glitch": True,
+    }) is None
+
+
+def test_transform_position_skips_missing_coords():
+    assert transform_vesselapi_position({"mmsi": 123456789, "vessel_name": "NOFIX"}) is None
+
+
+def test_transform_payload_skips_glitch_and_nofix_rows():
+    rows = transform_vesselapi_payload(_sample_payload())
+    assert [r["id"] for r in rows] == ["422050100"]
+
+
+# ── Quota budget / scheduling ─────────────────────────────────────────────
+
+def test_days_left_in_month():
+    assert days_left_in_month(datetime(2026, 8, 29, tzinfo=timezone.utc)) == 3
+
+
+def test_budget_allows_local_daily_cap():
+    # 5 calls already made → 6th is blocked regardless of remaining.
+    assert budget_allows(5, remaining=1000, days_left=3, max_per_day=5) is False
+
+
+def test_budget_allows_monthly_floor():
+    # remaining 14 ≤ 5*3=15 → skip; 16 > 15 → allow.
+    assert budget_allows(2, remaining=14, days_left=3, max_per_day=5) is False
+    assert budget_allows(2, remaining=16, days_left=3, max_per_day=5) is True
+
+
+def test_budget_allows_unknown_remaining():
+    assert budget_allows(2, remaining=None, days_left=3, max_per_day=5) is True
+
+
+def test_choose_box_prefers_primary_when_budget_tight():
+    boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)]
+    # 4 calls made, 1 left → always box 0.
+    assert choose_box(boxes, 4, max_per_day=5) == 0
+
+
+def test_choose_box_round_robins_when_budget_covers_all():
+    boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)]
+    # 0 calls made, 5 left ≥ 3 boxes → round-robin.
+    assert choose_box(boxes, 0, max_per_day=5) == 0
+    assert choose_box(boxes, 1, max_per_day=5) == 1
+    assert choose_box(boxes, 2, max_per_day=5) == 2
+
+
+# ── poll_once integration (fake store + fake client) ─────────────────────
+
+def _patch_side_effects(monkeypatch):
+    async def _noop(*a, **k):
+        return None
+
+    monkeypatch.setattr("tracks.record_position", _noop)
+    monkeypatch.setattr("geofence.record_and_notify", _noop)
+
+
+def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch):
+    _patch_side_effects(monkeypatch)
+    vessel_last_known.clear()
+    client = FakeClient(
+        FakeResp(200, {"X-RateLimit-Remaining": "140"}, _sample_payload()),
+    )
+    monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+    store = FakeStore()
+
+    ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+    assert ok is True
+    assert store.calls == 1
+    assert "422050100" in vessel_last_known
+    assert vessel_last_known["422050100"]["extra"]["src"] == "vesselapi"
+    assert "422050101" not in vessel_last_known  # glitch skipped
+    # One HTTP call, bounding-box params, no sat / no nextToken follow.
+    assert len(client.calls) == 1
+    _url, params, headers = client.calls[0]
+    assert params["filter.latBottom"] == "25.5"
+    assert params["filter.latTop"] == "27.3"
+    assert params["filter.lonLeft"] == "55.4"
+    assert params["filter.lonRight"] == "57.2"
+    assert params["pagination.limit"] == "50"
+    assert "sat" not in params
+    assert headers["Authorization"] == "Bearer test-key"
+
+
+def test_poll_once_sixth_2xx_is_skipped_with_zero_http(monkeypatch):
+    client = FakeClient()
+    monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+    # 5 successful calls already today → 6th poll makes no HTTP request.
+    store = FakeStore(calls=5, remaining=1000)
+
+    ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+    assert ok is False
+    assert store.bumps == 0
+    assert client.calls == []
+
+
+def test_poll_once_4xx_not_counted_and_no_upsert(monkeypatch):
+    _patch_side_effects(monkeypatch)
+    vessel_last_known.clear()
+    client = FakeClient(FakeResp(400, {}, {"error": {}}))
+    monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client))
+    store = FakeStore()
+
+    ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key"))
+
+    assert ok is False
+    assert store.bumps == 0  # 4xx does not count against quota
+    assert vessel_last_known == {}
+
+
+def _make_get_client(client):
+    async def _get_client():
+        return client
+
+    return _get_client
+
+
+# ── Independent-provider idle behaviour ───────────────────────────────────
+# VesselAPI must never be gated on AISStream (or vice versa): a missing key on
+# one provider leaves the other running. ``_StopLoop`` is a BaseException so
+# the worker's ``except Exception`` handler can't swallow it — the first
+# ``asyncio.sleep`` aborts the loop after exactly one decision.
+
+class _StopLoop(BaseException):
+    pass
+
+
+async def _stop_sleep(*_a, **_k):
+    raise _StopLoop()
+
+
+async def _resolve(value: str):
+    return value
+
+
+def test_worker_idles_without_vesselapi_key_even_if_aisstream_set(monkeypatch):
+    # AISStream key present, VesselAPI key absent -> no poll, still idles.
+    monkeypatch.setenv("AISSTREAM_API_KEY", "unused-aisstream-key")
+    monkeypatch.setenv("VESSELAPI_API_KEY", "")
+    monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve(""))
+    monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+    poll_calls: list = []
+    async def _spy_poll(store, boxes, key):
+        poll_calls.append(key)
+        return True
+    monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+    with pytest.raises(_StopLoop):
+        asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+    assert poll_calls == []
+
+
+def test_worker_polls_with_vesselapi_key_even_if_aisstream_unset(monkeypatch):
+    # AISStream key absent, VesselAPI key present -> still polls exactly once.
+    monkeypatch.setenv("AISSTREAM_API_KEY", "")
+    monkeypatch.setattr(vesselapi, "_resolve_key", lambda: _resolve("vesselapi-key"))
+    monkeypatch.setattr(vesselapi.asyncio, "sleep", _stop_sleep)
+    poll_calls: list = []
+    async def _spy_poll(store, boxes, key):
+        poll_calls.append(key)
+        return True
+    monkeypatch.setattr(vesselapi, "poll_once", _spy_poll)
+
+    with pytest.raises(_StopLoop):
+        asyncio.run(vesselapi.run_vesselapi_worker(FakeStore()))
+
+    assert poll_calls == ["vesselapi-key"]