Wire AIS end to end: telemetry ingestion + live map overlay

node-26#9, same shape as the ADS-B commit. Adds POST /telemetry/ais (same node-key auth, same org_id-stamped upsert-by-key pattern, this time by mmsi into a new `vessels` collection) and its docInMyOrg() firestore rule. Frontend gets useVessels() (mirrors useAircraft(), longer staleness window since AIS position reports are minutes apart, not seconds) and an opt-in "Vessels" map overlay. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-20 16:12:56 -04:00
parent 5537b095df
commit 3d2b722c64
7 changed files with 213 additions and 0 deletions
@@ -85,6 +85,20 @@ class AircraftTrack(BaseModel):
    last_seen: datetime


+class VesselTrack(BaseModel):
+    """Live AIS position, one doc per mmsi. Same live-snapshot shape as
+    AircraftTrack — overwritten on every sighting (see node-26#9)."""
+    mmsi: str
+    org_id: Optional[str] = None
+    node_id: str
+    name: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    speed_kt: Optional[float] = None
+    heading_deg: Optional[float] = None
+    last_seen: datetime
+
+
 class CommandPayload(BaseModel):
    action: str                      # discord_join / discord_leave / op25_restart
    guild_id: Optional[str] = None
@@ -69,3 +69,57 @@ async def upload_adsb(
            logger.warning(f"Failed to upsert {collection}/{doc_id} from node {node_id}: {e}")

    return {"ok": True, "count": len(writes)}
+
+
+class VesselReport(BaseModel):
+    mmsi: str
+    name: Optional[str] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    speed_kt: Optional[float] = None
+    heading_deg: Optional[float] = None
+
+
+class AisUploadBody(BaseModel):
+    vessels: List[VesselReport]
+
+
+@router.post("/ais")
+async def upload_ais(
+    body: AisUploadBody,
+    decoded: dict = Depends(require_node_service_or_firebase_token),
+):
+    """Same shape as /telemetry/adsb, one doc per mmsi in `vessels`."""
+    node_id = decoded.get("node_id")
+    if not node_id:
+        raise HTTPException(400, "This endpoint requires node identity, not a service/admin token.")
+
+    node = await fstore.doc_get_cached("nodes", node_id)
+    org_id = node.get("org_id") if node else None
+    now = datetime.now(timezone.utc).isoformat()
+
+    writes = []
+    for v in body.vessels:
+        if not v.mmsi:
+            continue
+        doc = {
+            "mmsi": v.mmsi,
+            "node_id": node_id,
+            "name": v.name,
+            "lat": v.lat,
+            "lon": v.lon,
+            "speed_kt": v.speed_kt,
+            "heading_deg": v.heading_deg,
+            "last_seen": now,
+        }
+        if org_id:
+            doc["org_id"] = org_id
+        writes.append(("vessels", v.mmsi, doc))
+
+    for collection, doc_id, doc in writes:
+        try:
+            await fstore.doc_set(collection, doc_id, doc, merge=True)
+        except Exception as e:
+            logger.warning(f"Failed to upsert {collection}/{doc_id} from node {node_id}: {e}")
+
+    return {"ok": True, "count": len(writes)}
@@ -58,3 +58,37 @@ def test_node_upload_skips_entries_missing_icao():
    assert resp.status_code == 200
    assert resp.json() == {"ok": True, "count": 0}
    mock_set.assert_not_awaited()
+
+
+def test_ais_service_token_without_node_id_is_rejected():
+    _override({"service": True})
+    resp = client.post("/telemetry/ais", json={"vessels": []})
+    assert resp.status_code == 400
+
+
+def test_ais_node_upload_upserts_and_stamps_org_id():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value={"org_id": "org-A"})), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/ais", json={
+            "vessels": [{"mmsi": "123456789", "name": "MV TEST", "lat": 41.0, "lon": -73.9}],
+        })
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 1}
+    mock_set.assert_awaited_once()
+    (collection, doc_id, doc), kwargs = mock_set.await_args
+    assert collection == "vessels"
+    assert doc_id == "123456789"
+    assert doc["node_id"] == "node-1"
+    assert doc["org_id"] == "org-A"
+    assert kwargs.get("merge") is True
+
+
+def test_ais_node_upload_skips_entries_missing_mmsi():
+    _override({"node": True, "node_id": "node-1"})
+    with patch.object(telemetry.fstore, "doc_get_cached", AsyncMock(return_value=None)), \
+         patch.object(telemetry.fstore, "doc_set", AsyncMock()) as mock_set:
+        resp = client.post("/telemetry/ais", json={"vessels": [{"mmsi": ""}]})
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "count": 0}
+    mock_set.assert_not_awaited()
@@ -16,6 +16,7 @@ import type { CallRecord, IncidentRecord, NodeRecord, NodeStatus } from "@/lib/t
 import { isKnownSeverity, SEVERITY_COLORS, SEVERITY_LABEL, type Severity } from "@/lib/severity";
 import { MachineOutputNotice } from "@/components/ui/MachineOutputNotice";
 import { useAircraft } from "@/lib/useAircraft";
+import { useVessels } from "@/lib/useVessels";

 // ── Leaflet icon fix ──────────────────────────────────────────────────────────
 delete (L.Icon.Default.prototype as unknown as Record<string, unknown>)._getIconUrl;
@@ -125,6 +126,39 @@ function AircraftLayer() {
  );
 }

+// ── Vessel icon — node-26#9 second-SDR AIS overlay ─────────────────────────────
+function vesselIcon(headingDeg: number | null): L.DivIcon {
+  const size = 14;
+  const rotation = headingDeg ?? 0;
+  return L.divIcon({
+    className: "",
+    html: `<div style="width:${size}px;height:${size}px;transform:rotate(${rotation}deg)"><svg width="${size}" height="${size}" viewBox="0 0 24 24" fill="var(--accent)" stroke="var(--surface)" stroke-width="1"><path d="M12 2 L18 14 L18 20 L6 20 L6 14 Z"/></svg></div>`,
+    iconSize: [size, size],
+    iconAnchor: [size / 2, size / 2],
+  });
+}
+
+function VesselLayer() {
+  const { vessels } = useVessels();
+  return (
+    <>
+      {vessels
+        .filter((v) => v.lat != null && v.lon != null)
+        .map((v) => (
+          <Marker key={v.mmsi} position={[v.lat as number, v.lon as number]} icon={vesselIcon(v.heading_deg)}>
+            <Popup minWidth={160}>
+              <div className="space-y-1">
+                <div className="font-semibold">{v.name || v.mmsi}</div>
+                <div className="text-xs text-ink-muted">MMSI {v.mmsi}</div>
+                {v.speed_kt != null && <div className="text-xs">Speed: {Math.round(v.speed_kt)} kt</div>}
+              </div>
+            </Popup>
+          </Marker>
+        ))}
+    </>
+  );
+}
+
 function nodeFanIcon(members: NodeRecord[]): L.DivIcon {
  const n = members.length;
  const CARD = 13;
@@ -619,6 +653,13 @@ export default function MapView({ nodes, activeCalls, incidents = [], calls = []
            </FeatureGroup>
          </LayersControl.Overlay>

+          {/* Overlay: Vessels — node-26#9 second-SDR AIS live snapshot, opt-in */}
+          <LayersControl.Overlay name="Vessels">
+            <FeatureGroup>
+              <VesselLayer />
+            </FeatureGroup>
+          </LayersControl.Overlay>
+
          {/* Overlay: Weather Radar — NEXRAD via Iowa Env Mesonet; key forces remount on refresh */}
          <LayersControl.Overlay name="Weather Radar">
            <TileLayer
@@ -74,6 +74,18 @@ export interface AircraftTrack {
  last_seen: string;
 }

+export interface VesselTrack {
+  mmsi: string;
+  org_id?: string;
+  node_id: string;
+  name: string | null;
+  lat: number | null;
+  lon: number | null;
+  speed_kt: number | null;
+  heading_deg: number | null;
+  last_seen: string;
+}
+
 export interface VocabularyPendingTerm {
  term: string;
  source: "induction" | "correction";
@@ -0,0 +1,53 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { collection, onSnapshot, query, where, FirestoreError } from "firebase/firestore";
+import { onAuthStateChanged } from "firebase/auth";
+import { db, auth } from "@/lib/firebase";
+import { useAuth } from "@/components/AuthProvider";
+import type { VesselTrack } from "@/lib/types";
+
+// Same shape as useAircraft — `vessels` is a live snapshot (one per mmsi,
+// overwritten on every sighting, node-26#9), nothing prunes a doc when a
+// vessel goes out of range, so staleness is filtered client-side. AIS
+// position reports are much less frequent than ADS-B (minutes, not
+// seconds), so this window is longer than useAircraft's.
+const STALE_AFTER_MS = 10 * 60 * 1000;
+
+export function useVessels() {
+  const [vessels, setVessels] = useState<VesselTrack[]>([]);
+  const [loading, setLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const { orgId } = useAuth();
+
+  useEffect(() => {
+    let unsubFirestore: (() => void) | undefined;
+
+    const unsubAuth = onAuthStateChanged(auth, (user) => {
+      if (unsubFirestore) { unsubFirestore(); unsubFirestore = undefined; }
+
+      if (!user || !orgId) {
+        setVessels([]);
+        setLoading(false);
+        return;
+      }
+
+      const q = query(collection(db, "vessels"), where("org_id", "==", orgId));
+      unsubFirestore = onSnapshot(q, (snap) => {
+        const now = Date.now();
+        const fresh = snap.docs
+          .map((d) => d.data() as VesselTrack)
+          .filter((v) => now - new Date(v.last_seen).getTime() < STALE_AFTER_MS);
+        setVessels(fresh);
+        setLoading(false);
+      }, (err: FirestoreError) => { console.error("useVessels:", err); setError(err.message); setLoading(false); });
+    });
+
+    return () => {
+      unsubAuth();
+      if (unsubFirestore) unsubFirestore();
+    };
+  }, [orgId]);
+
+  return { vessels, loading, error };
+}
@@ -102,6 +102,11 @@ service cloud.firestore {
      allow write: if false;
    }

+    match /vessels/{mmsi} {
+      allow read: if docInMyOrg();
+      allow write: if false;
+    }
+
    match /alert_events/{alertId} {
      allow read: if docInMyOrg();
      allow write: if false;