diff --git a/backend/main.py b/backend/main.py index 5f830b6..4166dbf 100644 --- a/backend/main.py +++ b/backend/main.py @@ -202,14 +202,40 @@ from fastapi.middleware.cors import CORSMiddleware from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.background import BackgroundTask from contextlib import asynccontextmanager -from services.data_fetcher import ( - start_scheduler, - stop_scheduler, - get_latest_data, - seed_startup_caches, -) -from services.ais_stream import start_ais_stream, stop_ais_stream -from services.carrier_tracker import start_carrier_tracker, stop_carrier_tracker +if not _MESH_ONLY: + from services.data_fetcher import ( + start_scheduler, + stop_scheduler, + get_latest_data, + seed_startup_caches, + ) + from services.ais_stream import start_ais_stream, stop_ais_stream + from services.carrier_tracker import start_carrier_tracker, stop_carrier_tracker +else: + # Lean mesh/wormhole process — avoid importing the OSINT fetcher graph. + def start_scheduler(*_a, **_k): # type: ignore[misc] + return None + + def stop_scheduler(*_a, **_k): # type: ignore[misc] + return None + + def get_latest_data(): # type: ignore[misc] + return {} + + def seed_startup_caches(*_a, **_k): # type: ignore[misc] + return None + + def start_ais_stream(*_a, **_k): # type: ignore[misc] + return None + + def stop_ais_stream(*_a, **_k): # type: ignore[misc] + return None + + def start_carrier_tracker(*_a, **_k): # type: ignore[misc] + return None + + def stop_carrier_tracker(*_a, **_k): # type: ignore[misc] + return None from slowapi import _rate_limit_exceeded_handler from slowapi.errors import RateLimitExceeded from services.schemas import HealthResponse, RefreshResponse @@ -352,28 +378,47 @@ def _load_optional_router(module_name: str) -> APIRouter: health_router = _load_optional_router("routers.health") -cctv_router = _load_optional_router("routers.cctv") -radio_router = _load_optional_router("routers.radio") -sigint_router = _load_optional_router("routers.sigint") -tools_router = _load_optional_router("routers.tools") -admin_router = _load_optional_router("routers.admin") -data_router = _load_optional_router("routers.data") mesh_peer_sync_router = _load_optional_router("routers.mesh_peer_sync") mesh_operator_router = _load_optional_router("routers.mesh_operator") mesh_oracle_router = _load_optional_router("routers.mesh_oracle") mesh_dm_router = _load_optional_router("routers.mesh_dm") mesh_public_router = _load_optional_router("routers.mesh_public") wormhole_router = _load_optional_router("routers.wormhole") -ai_intel_router = _load_optional_router("routers.ai_intel") -sar_router = _load_optional_router("routers.sar") infonet_router = _load_optional_router("routers.infonet") -road_corridors_router = _load_optional_router("routers.road_corridors") -osint_router = _load_optional_router("routers.osint") -scm_router = _load_optional_router("routers.scm") -entity_graph_router = _load_optional_router("routers.entity_graph") -intel_feeds_router = _load_optional_router("routers.intel_feeds") -analytics_router = _load_optional_router("routers.analytics") -agent_shell_router = _load_optional_router("routers.agent_shell") + +if _MESH_ONLY: + # Empty routers keep include_router() call sites unchanged without loading OSINT. + cctv_router = APIRouter() + radio_router = APIRouter() + sigint_router = APIRouter() + tools_router = APIRouter() + admin_router = APIRouter() + data_router = APIRouter() + ai_intel_router = APIRouter() + sar_router = APIRouter() + road_corridors_router = APIRouter() + osint_router = APIRouter() + scm_router = APIRouter() + entity_graph_router = APIRouter() + intel_feeds_router = APIRouter() + analytics_router = APIRouter() + agent_shell_router = APIRouter() +else: + cctv_router = _load_optional_router("routers.cctv") + radio_router = _load_optional_router("routers.radio") + sigint_router = _load_optional_router("routers.sigint") + tools_router = _load_optional_router("routers.tools") + admin_router = _load_optional_router("routers.admin") + data_router = _load_optional_router("routers.data") + ai_intel_router = _load_optional_router("routers.ai_intel") + sar_router = _load_optional_router("routers.sar") + road_corridors_router = _load_optional_router("routers.road_corridors") + osint_router = _load_optional_router("routers.osint") + scm_router = _load_optional_router("routers.scm") + entity_graph_router = _load_optional_router("routers.entity_graph") + intel_feeds_router = _load_optional_router("routers.intel_feeds") + analytics_router = _load_optional_router("routers.analytics") + agent_shell_router = _load_optional_router("routers.agent_shell") # --------------------------------------------------------------------------- diff --git a/backend/routers/data.py b/backend/routers/data.py index 4f2d581..5a98bda 100644 --- a/backend/routers/data.py +++ b/backend/routers/data.py @@ -3,6 +3,8 @@ import logging import math import os import threading +from collections import OrderedDict +from collections.abc import Callable from typing import Any from fastapi import APIRouter, Request, Response, Query, Depends from fastapi.responses import JSONResponse @@ -19,6 +21,11 @@ router = APIRouter() _refresh_lock = threading.Lock() +# ETag → serialized live-data body. Natural invalidation via etag change. +_LIVE_DATA_BYTES_CACHE_MAX = 32 +_LIVE_DATA_BYTES_CACHE: OrderedDict[str, bytes] = OrderedDict() +_LIVE_DATA_BYTES_CACHE_LOCK = threading.Lock() + class ViewportUpdate(BaseModel): s: float @@ -220,6 +227,38 @@ def _live_data_json_bytes(payload: dict) -> bytes: ) +def _cached_live_data_bytes(etag: str, build_payload_fn: Callable[[], dict]) -> bytes: + """Return orjson bytes for etag, building+serializing only on cache miss. + + Cache key is the full ETag string (includes version/bbox/prefix), so a + store version bump naturally misses without explicit invalidation. + """ + with _LIVE_DATA_BYTES_CACHE_LOCK: + cached = _LIVE_DATA_BYTES_CACHE.get(etag) + if cached is not None: + _LIVE_DATA_BYTES_CACHE.move_to_end(etag) + return cached + + body = _live_data_json_bytes(build_payload_fn()) + + with _LIVE_DATA_BYTES_CACHE_LOCK: + _LIVE_DATA_BYTES_CACHE[etag] = body + _LIVE_DATA_BYTES_CACHE.move_to_end(etag) + while len(_LIVE_DATA_BYTES_CACHE) > _LIVE_DATA_BYTES_CACHE_MAX: + _LIVE_DATA_BYTES_CACHE.popitem(last=False) + return body + + +def _item_lng(item: dict, lng_key: str = "lng"): + """Prefer ``lng``; fall back to ``lon`` (CCTV / KiwiSDR / PSK use lon).""" + lng = item.get(lng_key) + if lng is not None: + return lng + if lng_key != "lon": + return item.get("lon") + return item.get("lng") + + def _bbox_filter(items: list, s: float, w: float, n: float, e: float, lat_key: str = "lat", lng_key: str = "lng") -> list: pad_lat = (n - s) * 0.2 @@ -230,7 +269,7 @@ def _bbox_filter(items: list, s: float, w: float, n: float, e: float, out = [] for item in items: lat = item.get(lat_key) - lng = item.get(lng_key) + lng = _item_lng(item, lng_key) if lat is None or lng is None: out.append(item) continue @@ -293,6 +332,18 @@ def _cap_startup_items(items: list | None, max_items: int) -> list: return items[:max_items] +def _sample_items(items: list | None, max_items: int) -> list: + """Evenly spaced sample — stable across polls for the same store order.""" + if not items: + return [] + n = len(items) + if n <= max_items: + return items + # Deterministic stride so ETag byte cache stays coherent for a given version. + step = n / max_items + return [items[int(i * step)] for i in range(max_items)] + + def _cap_fast_startup_payload(payload: dict) -> dict: capped = dict(payload) capped["commercial_flights"] = _cap_startup_items(capped.get("commercial_flights"), 800) @@ -306,8 +357,37 @@ def _cap_fast_startup_payload(payload: dict) -> dict: return capped -def _cap_fast_dashboard_payload(payload: dict) -> dict: - return payload +# Dense layers only — never cap static infra (bases, plants, satellites, …). +# World caps keep first paint / zoomed-out views from shipping millions of rows; +# continental is looser; regional/city zoom skips capping (bbox already densifies). +_WORLD_FAST_CAPS: dict[str, int] = { + "commercial_flights": 1200, + "military_flights": 400, + "private_flights": 400, + "private_jets": 200, + "tracked_flights": 200, + "ships": 2000, + "cctv": 600, + "uavs": 200, + "liveuamap": 400, + "gps_jamming": 300, + "sigint": 800, + "trains": 200, +} +_CONTINENTAL_FAST_CAPS: dict[str, int] = { + "commercial_flights": 2500, + "military_flights": 800, + "private_flights": 800, + "private_jets": 400, + "tracked_flights": 400, + "ships": 4000, + "cctv": 1500, + "uavs": 400, + "liveuamap": 800, + "gps_jamming": 600, + "sigint": 1500, + "trains": 400, +} def _world_and_continental_scale(has_bbox: bool, s, w, n, e) -> tuple: @@ -317,6 +397,40 @@ def _world_and_continental_scale(has_bbox: bool, s, w, n, e) -> tuple: return world_scale, continental_scale +def _cap_fast_dashboard_payload(payload: dict, *, s=None, w=None, n=None, e=None) -> dict: + """Sample dense layers at world/continental zoom only. + + Regional pans rely on #288 bbox filtering for density — do not strip those. + Totals for sampled keys are preserved so the UI can show "N of M". + """ + has_bbox = _has_full_bbox(s, w, n, e) + world_scale, continental_scale = _world_and_continental_scale(has_bbox, s, w, n, e) + if not world_scale and not continental_scale: + payload["payload_scale"] = "regional" + return payload + + caps = _WORLD_FAST_CAPS if world_scale else _CONTINENTAL_FAST_CAPS + capped = dict(payload) + layer_totals = dict(capped.get("layer_totals") or {}) + sampled = False + for key, limit in caps.items(): + items = capped.get(key) + if not isinstance(items, list) or len(items) <= limit: + continue + layer_totals[key] = len(items) + capped[key] = _sample_items(items, limit) + sampled = True + # Keep cctv_total as the true DB/store count even when the array is sampled. + if "cctv" in layer_totals: + capped["cctv_total"] = layer_totals["cctv"] + if layer_totals: + capped["layer_totals"] = layer_totals + capped["payload_scale"] = "world" if world_scale else "continental" + if sampled: + capped["payload_sampled"] = True + return capped + + def _filter_sigint_by_layers(items: list, active_layers: dict) -> list: allow_aprs = bool(active_layers.get("sigint_aprs", True)) allow_mesh = bool(active_layers.get("sigint_meshtastic", True)) @@ -616,11 +730,13 @@ async def live_data(request: Request): etag = _current_etag(prefix="live|full|") if request.headers.get("if-none-match") == etag: return Response(status_code=304, headers={"ETag": etag, "Cache-Control": "no-cache"}) - from services.fetchers._store import get_latest_data_deepcopy_snapshot + from services.fetchers._store import get_latest_data_refs_snapshot + + def _build() -> dict: + return get_latest_data_refs_snapshot() - payload = get_latest_data_deepcopy_snapshot() return Response( - content=_live_data_json_bytes(payload), + content=_cached_live_data_bytes(etag, _build), media_type="application/json", headers={"ETag": etag, "Cache-Control": "no-cache"}, ) @@ -644,86 +760,223 @@ async def bootstrap_critical(request: Request): get_source_timestamps_snapshot, ) - d = get_latest_data_subset_refs( - "last_updated", "commercial_flights", "military_flights", "private_flights", - "private_jets", "tracked_flights", "ships", "uavs", "liveuamap", "gps_jamming", - "satellites", "satellite_source", "satellite_analysis", "sigint", "sigint_totals", - "trains", "news", "gdelt", "airports", "threat_level", "trending_markets", - "correlations", "fimi", "crowdthreat", - ) - freshness = get_source_timestamps_snapshot() - ships_enabled = any(active_layers.get(key, True) for key in ( - "ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts")) - sigint_items = _filter_sigint_by_layers(d.get("sigint") or [], active_layers) - payload = { - "last_updated": d.get("last_updated"), - "commercial_flights": _cap_startup_items( - (d.get("commercial_flights") or []) if active_layers.get("flights", True) else [], - 800, - ), - "military_flights": _cap_startup_items( - (d.get("military_flights") or []) if active_layers.get("military", True) else [], - 300, - ), - "private_flights": _cap_startup_items( - (d.get("private_flights") or []) if active_layers.get("private", True) else [], - 300, - ), - "private_jets": _cap_startup_items( - (d.get("private_jets") or []) if active_layers.get("jets", True) else [], - 150, - ), - "tracked_flights": _cap_startup_items( - (d.get("tracked_flights") or []) if active_layers.get("tracked", True) else [], - 250, - ), - "ships": _cap_startup_items((d.get("ships") or []) if ships_enabled else [], 1500), - "uavs": _cap_startup_items((d.get("uavs") or []) if active_layers.get("military", True) else [], 100), - "liveuamap": _cap_startup_items( - (d.get("liveuamap") or []) if active_layers.get("global_incidents", True) else [], - 300, - ), - "gps_jamming": _cap_startup_items( - (d.get("gps_jamming") or []) if active_layers.get("gps_jamming", True) else [], - 200, - ), - "satellites": _cap_startup_items( - (d.get("satellites") or []) if active_layers.get("satellites", True) else [], - 250, - ), - "satellite_source": d.get("satellite_source", "none"), - "satellite_analysis": (d.get("satellite_analysis") or {}) if active_layers.get("satellites", True) else {}, - "sigint": _cap_startup_items( - sigint_items if (active_layers.get("sigint_meshtastic", True) or active_layers.get("sigint_aprs", True)) else [], - 500, - ), - "sigint_totals": _sigint_totals_for_items(sigint_items), - "trains": _cap_startup_items((d.get("trains") or []) if active_layers.get("trains", True) else [], 100), - "news": _cap_startup_items(d.get("news") or [], 30), - "gdelt": _cap_startup_items((d.get("gdelt") or []) if active_layers.get("global_incidents", True) else [], 300), - "airports": _cap_startup_items(d.get("airports") or [], 500), - "threat_level": d.get("threat_level"), - "trending_markets": _cap_startup_items(d.get("trending_markets") or [], 10), - "correlations": _cap_startup_items( - (d.get("correlations") or []) if active_layers.get("correlations", True) else [], - 50, - ), - "fimi": d.get("fimi"), - "crowdthreat": _cap_startup_items( - (d.get("crowdthreat") or []) if active_layers.get("crowdthreat", True) else [], - 150, - ), - "freshness": freshness, - "bootstrap_ready": True, - "bootstrap_payload": True, - } + def _build() -> dict: + d = get_latest_data_subset_refs( + "last_updated", "commercial_flights", "military_flights", "private_flights", + "private_jets", "tracked_flights", "ships", "uavs", "liveuamap", "gps_jamming", + "satellites", "satellite_source", "satellite_analysis", "sigint", "sigint_totals", + "trains", "news", "gdelt", "airports", "threat_level", "trending_markets", + "correlations", "fimi", "crowdthreat", + ) + freshness = get_source_timestamps_snapshot() + ships_enabled = any(active_layers.get(key, True) for key in ( + "ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts")) + sigint_items = _filter_sigint_by_layers(d.get("sigint") or [], active_layers) + return { + "last_updated": d.get("last_updated"), + "commercial_flights": _cap_startup_items( + (d.get("commercial_flights") or []) if active_layers.get("flights", True) else [], + 800, + ), + "military_flights": _cap_startup_items( + (d.get("military_flights") or []) if active_layers.get("military", True) else [], + 300, + ), + "private_flights": _cap_startup_items( + (d.get("private_flights") or []) if active_layers.get("private", True) else [], + 300, + ), + "private_jets": _cap_startup_items( + (d.get("private_jets") or []) if active_layers.get("jets", True) else [], + 150, + ), + "tracked_flights": _cap_startup_items( + (d.get("tracked_flights") or []) if active_layers.get("tracked", True) else [], + 250, + ), + "ships": _cap_startup_items((d.get("ships") or []) if ships_enabled else [], 1500), + "uavs": _cap_startup_items((d.get("uavs") or []) if active_layers.get("military", True) else [], 100), + "liveuamap": _cap_startup_items( + (d.get("liveuamap") or []) if active_layers.get("global_incidents", True) else [], + 300, + ), + "gps_jamming": _cap_startup_items( + (d.get("gps_jamming") or []) if active_layers.get("gps_jamming", True) else [], + 200, + ), + "satellites": _cap_startup_items( + (d.get("satellites") or []) if active_layers.get("satellites", True) else [], + 250, + ), + "satellite_source": d.get("satellite_source", "none"), + "satellite_analysis": (d.get("satellite_analysis") or {}) if active_layers.get("satellites", True) else {}, + "sigint": _cap_startup_items( + sigint_items if (active_layers.get("sigint_meshtastic", True) or active_layers.get("sigint_aprs", True)) else [], + 500, + ), + "sigint_totals": _sigint_totals_for_items(sigint_items), + "trains": _cap_startup_items((d.get("trains") or []) if active_layers.get("trains", True) else [], 100), + "news": _cap_startup_items(d.get("news") or [], 30), + "gdelt": _cap_startup_items((d.get("gdelt") or []) if active_layers.get("global_incidents", True) else [], 300), + "airports": _cap_startup_items(d.get("airports") or [], 500), + "threat_level": d.get("threat_level"), + "trending_markets": _cap_startup_items(d.get("trending_markets") or [], 10), + "correlations": _cap_startup_items( + (d.get("correlations") or []) if active_layers.get("correlations", True) else [], + 50, + ), + "fimi": d.get("fimi"), + "crowdthreat": _cap_startup_items( + (d.get("crowdthreat") or []) if active_layers.get("crowdthreat", True) else [], + 150, + ), + "freshness": freshness, + "bootstrap_ready": True, + "bootstrap_payload": True, + } + return Response( - content=_live_data_json_bytes(payload), + content=_cached_live_data_bytes(etag, _build), media_type="application/json", headers={"ETag": etag, "Cache-Control": "no-cache"}, ) +_DELTA_FAST_KEYS: tuple[str, ...] = ( + "ships", + "commercial_flights", + "military_flights", + "tracked_flights", + "private_flights", + "private_jets", +) + + +def _parse_layer_versions(raw: str | None) -> dict[str, int] | None: + """Parse ``ships:3,commercial_flights:5`` into a dict. Empty/invalid → None.""" + if not raw or not str(raw).strip(): + return None + out: dict[str, int] = {} + for part in str(raw).split(","): + part = part.strip() + if not part or ":" not in part: + continue + key, _, ver = part.partition(":") + key = key.strip() + try: + out[key] = int(ver.strip()) + except (TypeError, ValueError): + continue + return out or None + + +def _attach_version_meta(payload: dict) -> dict: + from services.fetchers._store import get_data_version, get_layer_versions + + payload["mode"] = payload.get("mode") or "snapshot" + payload["store_version"] = get_data_version() + payload["layer_versions"] = get_layer_versions() + return payload + + +def _try_build_fast_delta( + client_lv: dict[str, int], + *, + s=None, + w=None, + n=None, + e=None, +) -> dict | None: + """Return a delta payload, or None to fall back to a full snapshot.""" + from services.fetchers._store import ( + active_layers, + compute_layer_row_delta, + get_data_version, + get_layer_versions, + get_latest_data_subset_refs, + get_source_timestamps_snapshot, + ) + + server_lv = get_layer_versions() + deltas: dict[str, Any] = {} + for key in _DELTA_FAST_KEYS: + since = client_lv.get(key) + if since is None: + # Client never saw this layer — force full snapshot recovery. + return None + patch = compute_layer_row_delta(key, since) + if patch is None: + return None + if not patch.get("unchanged"): + # Bbox-filter upserts when a regional viewport is active. + upserts = list(patch.get("upsert") or []) + if upserts and _has_full_bbox(s, w, n, e): + lat_span, lng_span = _bbox_spans(s, w, n, e) + if not (lng_span >= 300 or lat_span >= 120): + upserts = _bbox_filter(upserts, s, w, n, e) + deltas[key] = { + "upsert": upserts, + "delete": list(patch.get("delete") or []), + "version": patch.get("version", server_lv.get(key, 0)), + } + + # Non-delta fast keys: include full array only when the layer version advanced. + non_delta_keys = ( + "cctv", + "uavs", + "liveuamap", + "gps_jamming", + "satellites", + "satellite_analysis", + "sigint", + "trains", + ) + d = get_latest_data_subset_refs(*non_delta_keys, "satellite_source", "sigint_totals") + full_layers: dict[str, Any] = {} + for key in non_delta_keys: + client_ver = client_lv.get(key) + server_ver = int(server_lv.get(key, 0) or 0) + if client_ver is not None and int(client_ver) >= server_ver: + continue + if key == "cctv": + full_layers[key] = (d.get("cctv") or []) if active_layers.get("cctv", True) else [] + elif key == "uavs": + full_layers[key] = (d.get("uavs") or []) if active_layers.get("military", True) else [] + elif key == "liveuamap": + full_layers[key] = (d.get("liveuamap") or []) if active_layers.get("global_incidents", True) else [] + elif key == "gps_jamming": + full_layers[key] = (d.get("gps_jamming") or []) if active_layers.get("gps_jamming", True) else [] + elif key == "satellites": + full_layers[key] = (d.get("satellites") or []) if active_layers.get("satellites", True) else [] + full_layers["satellite_source"] = d.get("satellite_source", "none") + full_layers["satellite_analysis"] = ( + (d.get("satellite_analysis") or {}) if active_layers.get("satellites", True) else {} + ) + elif key == "sigint": + items = _filter_sigint_by_layers(d.get("sigint") or [], active_layers) + full_layers[key] = ( + items + if (active_layers.get("sigint_meshtastic", True) or active_layers.get("sigint_aprs", True)) + else [] + ) + full_layers["sigint_totals"] = _sigint_totals_for_items(items) + elif key == "trains": + full_layers[key] = (d.get("trains") or []) if active_layers.get("trains", True) else [] + + if full_layers and _has_full_bbox(s, w, n, e): + _apply_bbox_to_payload(full_layers, _FAST_BBOX_HEAVY_KEYS, s, w, n, e) + + return { + "mode": "delta", + "store_version": get_data_version(), + "layer_versions": server_lv, + "deltas": deltas, + "layers": full_layers, + "freshness": get_source_timestamps_snapshot(), + "cctv_total": len((d.get("cctv") or [])), + } + + @router.get("/api/live-data/fast") @limiter.limit("120/minute") async def live_data_fast( @@ -733,57 +986,89 @@ async def live_data_fast( n: float = Query(None, description="North bound (see s)", ge=-90, le=90), e: float = Query(None, description="East bound (see s)", ge=-180, le=180), initial: bool = Query(False, description="Return a capped startup payload for first paint"), + lv: str = Query( + None, + description="Optional per-layer versions for delta mode, e.g. ships:3,commercial_flights:5. " + "When provided and recoverable, response is mode=delta (upsert/delete). " + "On skew/miss, falls back to a full snapshot.", + ), ): bbox_suffix = _bbox_etag_suffix(s, w, n, e) + client_lv = None if initial else _parse_layer_versions(lv) + want_delta = client_lv is not None + + if want_delta: + delta_payload = _try_build_fast_delta(client_lv, s=s, w=w, n=n, e=e) + if delta_payload is not None: + # Tiny unchanged delta → 304 when ETag matches store version alone. + etag = _current_etag( + prefix="fast|delta|" + bbox_suffix.lstrip("|") + ("|" if bbox_suffix else "") + ) + if ( + not delta_payload.get("deltas") + and not delta_payload.get("layers") + and request.headers.get("if-none-match") == etag + ): + return Response(status_code=304, headers={"ETag": etag, "Cache-Control": "no-cache"}) + body = _live_data_json_bytes(delta_payload) + return Response( + content=body, + media_type="application/json", + headers={"ETag": etag, "Cache-Control": "no-cache", "X-SB-Mode": "delta"}, + ) + etag = _current_etag(prefix=("fast|initial|" if initial else "fast|full|") + bbox_suffix.lstrip("|") + ("|" if bbox_suffix else "")) if request.headers.get("if-none-match") == etag: return Response(status_code=304, headers={"ETag": etag, "Cache-Control": "no-cache"}) from services.fetchers._store import (active_layers, get_latest_data_subset_refs, get_source_timestamps_snapshot) - d = get_latest_data_subset_refs( - "last_updated", "commercial_flights", "military_flights", "private_flights", - "private_jets", "tracked_flights", "ships", "cctv", "uavs", "liveuamap", - "gps_jamming", "satellites", "satellite_source", "satellite_analysis", - "sigint", "sigint_totals", "trains", - ) - freshness = get_source_timestamps_snapshot() - ships_enabled = any(active_layers.get(key, True) for key in ( - "ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts")) - cctv_total = len(d.get("cctv") or []) - sigint_items = _filter_sigint_by_layers(d.get("sigint") or [], active_layers) - sigint_totals = _sigint_totals_for_items(sigint_items) - payload = { - "commercial_flights": (d.get("commercial_flights") or []) if active_layers.get("flights", True) else [], - "military_flights": (d.get("military_flights") or []) if active_layers.get("military", True) else [], - "private_flights": (d.get("private_flights") or []) if active_layers.get("private", True) else [], - "private_jets": (d.get("private_jets") or []) if active_layers.get("jets", True) else [], - "tracked_flights": (d.get("tracked_flights") or []) if active_layers.get("tracked", True) else [], - "ships": (d.get("ships") or []) if ships_enabled else [], - "cctv": (d.get("cctv") or []) if active_layers.get("cctv", True) else [], - "uavs": (d.get("uavs") or []) if active_layers.get("military", True) else [], - "liveuamap": (d.get("liveuamap") or []) if active_layers.get("global_incidents", True) else [], - "gps_jamming": (d.get("gps_jamming") or []) if active_layers.get("gps_jamming", True) else [], - "satellites": (d.get("satellites") or []) if active_layers.get("satellites", True) else [], - "satellite_source": d.get("satellite_source", "none"), - "satellite_analysis": (d.get("satellite_analysis") or {}) if active_layers.get("satellites", True) else {}, - "sigint": sigint_items if (active_layers.get("sigint_meshtastic", True) or active_layers.get("sigint_aprs", True)) else [], - "sigint_totals": sigint_totals, - "cctv_total": cctv_total, - "trains": (d.get("trains") or []) if active_layers.get("trains", True) else [], - "freshness": freshness, - } - if initial: - payload = _cap_fast_startup_payload(payload) - else: - payload = _cap_fast_dashboard_payload(payload) - # Issue #288: bbox filter heavy/dense layers only when all four bounds - # are supplied. Without bounds, behaviour is byte-for-byte identical - # to the pre-#288 implementation. - if _has_full_bbox(s, w, n, e): - payload = _apply_bbox_to_payload(payload, _FAST_BBOX_HEAVY_KEYS, s, w, n, e) + + def _build() -> dict: + d = get_latest_data_subset_refs( + "last_updated", "commercial_flights", "military_flights", "private_flights", + "private_jets", "tracked_flights", "ships", "cctv", "uavs", "liveuamap", + "gps_jamming", "satellites", "satellite_source", "satellite_analysis", + "sigint", "sigint_totals", "trains", + ) + freshness = get_source_timestamps_snapshot() + ships_enabled = any(active_layers.get(key, True) for key in ( + "ships_military", "ships_cargo", "ships_civilian", "ships_passenger", "ships_tracked_yachts")) + cctv_total = len(d.get("cctv") or []) + sigint_items = _filter_sigint_by_layers(d.get("sigint") or [], active_layers) + sigint_totals = _sigint_totals_for_items(sigint_items) + payload = { + "commercial_flights": (d.get("commercial_flights") or []) if active_layers.get("flights", True) else [], + "military_flights": (d.get("military_flights") or []) if active_layers.get("military", True) else [], + "private_flights": (d.get("private_flights") or []) if active_layers.get("private", True) else [], + "private_jets": (d.get("private_jets") or []) if active_layers.get("jets", True) else [], + "tracked_flights": (d.get("tracked_flights") or []) if active_layers.get("tracked", True) else [], + "ships": (d.get("ships") or []) if ships_enabled else [], + "cctv": (d.get("cctv") or []) if active_layers.get("cctv", True) else [], + "uavs": (d.get("uavs") or []) if active_layers.get("military", True) else [], + "liveuamap": (d.get("liveuamap") or []) if active_layers.get("global_incidents", True) else [], + "gps_jamming": (d.get("gps_jamming") or []) if active_layers.get("gps_jamming", True) else [], + "satellites": (d.get("satellites") or []) if active_layers.get("satellites", True) else [], + "satellite_source": d.get("satellite_source", "none"), + "satellite_analysis": (d.get("satellite_analysis") or {}) if active_layers.get("satellites", True) else {}, + "sigint": sigint_items if (active_layers.get("sigint_meshtastic", True) or active_layers.get("sigint_aprs", True)) else [], + "sigint_totals": sigint_totals, + "cctv_total": cctv_total, + "trains": (d.get("trains") or []) if active_layers.get("trains", True) else [], + "freshness": freshness, + } + if initial: + payload = _cap_fast_startup_payload(payload) + else: + # Issue #288: bbox densify first so world/continental caps sample + # the *visible* set, not a random world subset then clip. + if _has_full_bbox(s, w, n, e): + payload = _apply_bbox_to_payload(payload, _FAST_BBOX_HEAVY_KEYS, s, w, n, e) + payload = _cap_fast_dashboard_payload(payload, s=s, w=w, n=n, e=e) + return _attach_version_meta(payload) + return Response( - content=_live_data_json_bytes(payload), + content=_cached_live_data_bytes(etag, _build), media_type="application/json", - headers={"ETag": etag, "Cache-Control": "no-cache"}, + headers={"ETag": etag, "Cache-Control": "no-cache", "X-SB-Mode": "snapshot"}, ) @@ -801,96 +1086,100 @@ async def live_data_slow( if request.headers.get("if-none-match") == etag: return Response(status_code=304, headers={"ETag": etag, "Cache-Control": "no-cache"}) from services.fetchers._store import (active_layers, get_latest_data_subset_refs, get_source_timestamps_snapshot) - d = get_latest_data_subset_refs( - "last_updated", "news", "stocks", "financial_source", "oil", "weather", "traffic", - "earthquakes", "frontlines", "gdelt", "airports", "kiwisdr", "satnogs_stations", - "satnogs_observations", "tinygs_satellites", "space_weather", "internet_outages", - "firms_fires", "datacenters", "military_bases", "power_plants", "viirs_change_nodes", - "scanners", "weather_alerts", "ukraine_alerts", "air_quality", "volcanoes", - "fishing_activity", "psk_reporter", "correlations", "uap_sightings", "wastewater", - "crowdthreat", "threat_level", "trending_markets", "road_corridor_trends", - "malware_threats", "cyber_threats", "scm_suppliers", "telegram_osint", "gt_risk", - ) - freshness = get_source_timestamps_snapshot() - payload = { - "last_updated": d.get("last_updated"), - "threat_level": d.get("threat_level"), - "trending_markets": d.get("trending_markets", []), - "news": d.get("news", []), - "stocks": d.get("stocks", {}), - "financial_source": d.get("financial_source", ""), - "oil": d.get("oil", {}), - "weather": d.get("weather"), - "traffic": d.get("traffic", []), - "earthquakes": (d.get("earthquakes") or []) if active_layers.get("earthquakes", True) else [], - "frontlines": d.get("frontlines") if active_layers.get("ukraine_frontline", True) else None, - "gdelt": (d.get("gdelt") or []) if active_layers.get("global_incidents", True) else [], - "airports": d.get("airports") or [], - "kiwisdr": (d.get("kiwisdr") or []) if active_layers.get("kiwisdr", True) else [], - "satnogs_stations": (d.get("satnogs_stations") or []) if active_layers.get("satnogs", True) else [], - "satnogs_total": len(d.get("satnogs_stations") or []), - "satnogs_observations": (d.get("satnogs_observations") or []) if active_layers.get("satnogs", True) else [], - "tinygs_satellites": (d.get("tinygs_satellites") or []) if active_layers.get("tinygs", True) else [], - "tinygs_total": len(d.get("tinygs_satellites") or []), - "psk_reporter": (d.get("psk_reporter") or []) if active_layers.get("psk_reporter", True) else [], - "space_weather": d.get("space_weather"), - "internet_outages": (d.get("internet_outages") or []) if active_layers.get("internet_outages", True) else [], - "firms_fires": (d.get("firms_fires") or []) if active_layers.get("firms", True) else [], - "datacenters": (d.get("datacenters") or []) if active_layers.get("datacenters", True) else [], - "military_bases": (d.get("military_bases") or []) if active_layers.get("military_bases", True) else [], - "power_plants": (d.get("power_plants") or []) if active_layers.get("power_plants", True) else [], - "viirs_change_nodes": (d.get("viirs_change_nodes") or []) if active_layers.get("viirs_nightlights", True) else [], - "scanners": (d.get("scanners") or []) if active_layers.get("scanners", True) else [], - "weather_alerts": d.get("weather_alerts", []) if active_layers.get("weather_alerts", True) else [], - "ukraine_alerts": d.get("ukraine_alerts", []) if active_layers.get("ukraine_alerts", True) else [], - "air_quality": (d.get("air_quality") or []) if active_layers.get("air_quality", True) else [], - "volcanoes": (d.get("volcanoes") or []) if active_layers.get("volcanoes", True) else [], - "fishing_activity": (d.get("fishing_activity") or []) if active_layers.get("fishing_activity", True) else [], - "correlations": (d.get("correlations") or []) if active_layers.get("correlations", True) else [], - "uap_sightings": (d.get("uap_sightings") or []) if active_layers.get("uap_sightings", True) else [], - "wastewater": (d.get("wastewater") or []) if active_layers.get("wastewater", True) else [], - "crowdthreat": (d.get("crowdthreat") or []) if active_layers.get("crowdthreat", True) else [], - "road_corridor_trends": ( - d.get("road_corridor_trends") or {"updated_at": None, "corridors": []} + + def _build() -> dict: + d = get_latest_data_subset_refs( + "last_updated", "news", "stocks", "financial_source", "oil", "weather", "traffic", + "earthquakes", "frontlines", "gdelt", "airports", "kiwisdr", "satnogs_stations", + "satnogs_observations", "tinygs_satellites", "space_weather", "internet_outages", + "firms_fires", "datacenters", "military_bases", "power_plants", "viirs_change_nodes", + "scanners", "weather_alerts", "ukraine_alerts", "air_quality", "volcanoes", + "fishing_activity", "psk_reporter", "correlations", "uap_sightings", "wastewater", + "crowdthreat", "threat_level", "trending_markets", "road_corridor_trends", + "malware_threats", "cyber_threats", "scm_suppliers", "telegram_osint", "gt_risk", ) - if active_layers.get("road_corridor_trends", False) - else {"updated_at": None, "corridors": []}, - "malware_threats": ( - d.get("malware_threats") or {"threats": [], "total": 0} - ) - if active_layers.get("malware_c2", False) - else {"threats": [], "total": 0}, - "cyber_threats": ( - d.get("cyber_threats") or {"threats": [], "stats": {}} - ) - if active_layers.get("cyber_threats", False) - else {"threats": [], "stats": {}}, - "scm_suppliers": ( - d.get("scm_suppliers") or {"suppliers": [], "total": 0, "critical_count": 0} - ) - if active_layers.get("scm_suppliers", False) - else {"suppliers": [], "total": 0, "critical_count": 0}, - "telegram_osint": ( - d.get("telegram_osint") or {"posts": [], "total": 0, "geolocated": 0} - ) - if active_layers.get("telegram_osint", True) - else {"posts": [], "total": 0, "geolocated": 0}, - "gt_risk": ( - d.get("gt_risk") - or {"enabled": False, "heatmap": {"type": "FeatureCollection", "features": []}, "clusters": []} - ) - if active_layers.get("gt_risk", False) - else {"enabled": False, "heatmap": {"type": "FeatureCollection", "features": []}, "clusters": []}, - "freshness": freshness, - } - # Issue #288: bbox filter heavy/dense layers only when all four bounds - # are supplied. Static reference layers (datacenters, military bases, - # power_plants, etc.) deliberately stay world-scale so panning never - # hides the infrastructure overlay the operator already has on screen. - if _has_full_bbox(s, w, n, e): - payload = _apply_bbox_to_payload(payload, _SLOW_BBOX_HEAVY_KEYS, s, w, n, e) + freshness = get_source_timestamps_snapshot() + payload = { + "last_updated": d.get("last_updated"), + "threat_level": d.get("threat_level"), + "trending_markets": d.get("trending_markets", []), + "news": d.get("news", []), + "stocks": d.get("stocks", {}), + "financial_source": d.get("financial_source", ""), + "oil": d.get("oil", {}), + "weather": d.get("weather"), + "traffic": d.get("traffic", []), + "earthquakes": (d.get("earthquakes") or []) if active_layers.get("earthquakes", True) else [], + "frontlines": d.get("frontlines") if active_layers.get("ukraine_frontline", True) else None, + "gdelt": (d.get("gdelt") or []) if active_layers.get("global_incidents", True) else [], + "airports": d.get("airports") or [], + "kiwisdr": (d.get("kiwisdr") or []) if active_layers.get("kiwisdr", True) else [], + "satnogs_stations": (d.get("satnogs_stations") or []) if active_layers.get("satnogs", True) else [], + "satnogs_total": len(d.get("satnogs_stations") or []), + "satnogs_observations": (d.get("satnogs_observations") or []) if active_layers.get("satnogs", True) else [], + "tinygs_satellites": (d.get("tinygs_satellites") or []) if active_layers.get("tinygs", True) else [], + "tinygs_total": len(d.get("tinygs_satellites") or []), + "psk_reporter": (d.get("psk_reporter") or []) if active_layers.get("psk_reporter", True) else [], + "space_weather": d.get("space_weather"), + "internet_outages": (d.get("internet_outages") or []) if active_layers.get("internet_outages", True) else [], + "firms_fires": (d.get("firms_fires") or []) if active_layers.get("firms", True) else [], + "datacenters": (d.get("datacenters") or []) if active_layers.get("datacenters", True) else [], + "military_bases": (d.get("military_bases") or []) if active_layers.get("military_bases", True) else [], + "power_plants": (d.get("power_plants") or []) if active_layers.get("power_plants", True) else [], + "viirs_change_nodes": (d.get("viirs_change_nodes") or []) if active_layers.get("viirs_nightlights", True) else [], + "scanners": (d.get("scanners") or []) if active_layers.get("scanners", True) else [], + "weather_alerts": d.get("weather_alerts", []) if active_layers.get("weather_alerts", True) else [], + "ukraine_alerts": d.get("ukraine_alerts", []) if active_layers.get("ukraine_alerts", True) else [], + "air_quality": (d.get("air_quality") or []) if active_layers.get("air_quality", True) else [], + "volcanoes": (d.get("volcanoes") or []) if active_layers.get("volcanoes", True) else [], + "fishing_activity": (d.get("fishing_activity") or []) if active_layers.get("fishing_activity", True) else [], + "correlations": (d.get("correlations") or []) if active_layers.get("correlations", True) else [], + "uap_sightings": (d.get("uap_sightings") or []) if active_layers.get("uap_sightings", True) else [], + "wastewater": (d.get("wastewater") or []) if active_layers.get("wastewater", True) else [], + "crowdthreat": (d.get("crowdthreat") or []) if active_layers.get("crowdthreat", True) else [], + "road_corridor_trends": ( + d.get("road_corridor_trends") or {"updated_at": None, "corridors": []} + ) + if active_layers.get("road_corridor_trends", False) + else {"updated_at": None, "corridors": []}, + "malware_threats": ( + d.get("malware_threats") or {"threats": [], "total": 0} + ) + if active_layers.get("malware_c2", False) + else {"threats": [], "total": 0}, + "cyber_threats": ( + d.get("cyber_threats") or {"threats": [], "stats": {}} + ) + if active_layers.get("cyber_threats", False) + else {"threats": [], "stats": {}}, + "scm_suppliers": ( + d.get("scm_suppliers") or {"suppliers": [], "total": 0, "critical_count": 0} + ) + if active_layers.get("scm_suppliers", False) + else {"suppliers": [], "total": 0, "critical_count": 0}, + "telegram_osint": ( + d.get("telegram_osint") or {"posts": [], "total": 0, "geolocated": 0} + ) + if active_layers.get("telegram_osint", True) + else {"posts": [], "total": 0, "geolocated": 0}, + "gt_risk": ( + d.get("gt_risk") + or {"enabled": False, "heatmap": {"type": "FeatureCollection", "features": []}, "clusters": []} + ) + if active_layers.get("gt_risk", False) + else {"enabled": False, "heatmap": {"type": "FeatureCollection", "features": []}, "clusters": []}, + "freshness": freshness, + } + # Issue #288: bbox filter heavy/dense layers only when all four bounds + # are supplied. Static reference layers (datacenters, military bases, + # power_plants, etc.) deliberately stay world-scale so panning never + # hides the infrastructure overlay the operator already has on screen. + if _has_full_bbox(s, w, n, e): + payload = _apply_bbox_to_payload(payload, _SLOW_BBOX_HEAVY_KEYS, s, w, n, e) + return payload + return Response( - content=_live_data_json_bytes(payload), + content=_cached_live_data_bytes(etag, _build), media_type="application/json", headers={"ETag": etag, "Cache-Control": "no-cache"}, ) diff --git a/backend/routers/mesh_oracle.py b/backend/routers/mesh_oracle.py index 35e024f..4edaa0f 100644 --- a/backend/routers/mesh_oracle.py +++ b/backend/routers/mesh_oracle.py @@ -5,7 +5,7 @@ from fastapi.responses import JSONResponse from pydantic import BaseModel from limiter import limiter from auth import require_admin, require_local_operator, _scoped_view_authenticated -from services.data_fetcher import get_latest_data +from services.fetchers._store import get_latest_data_refs_snapshot as get_latest_data from services.mesh.mesh_protocol import normalize_payload from services.mesh.mesh_signed_events import ( MeshWriteExemption, diff --git a/backend/routers/mesh_public.py b/backend/routers/mesh_public.py index e88f8dd..302eb62 100644 --- a/backend/routers/mesh_public.py +++ b/backend/routers/mesh_public.py @@ -41,7 +41,7 @@ _scoped_view_authenticated = _main_delegate("_scoped_view_authenticated") _node_runtime_snapshot = _main_delegate("_node_runtime_snapshot") _verify_gate_access_main = _main_delegate("_verify_gate_access") from services.config import get_settings -from services.data_fetcher import get_latest_data +from services.fetchers._store import get_latest_data_refs_snapshot as get_latest_data from services.mesh.mesh_crypto import ( derive_node_id, normalize_peer_url, diff --git a/backend/services/cctv_pipeline.py b/backend/services/cctv_pipeline.py index f659722..174d276 100644 --- a/backend/services/cctv_pipeline.py +++ b/backend/services/cctv_pipeline.py @@ -1446,16 +1446,67 @@ def run_all_ingestors(): logger.warning(f"Ingestor {ingestor.__class__.__name__} failed during seed: {e}") -def get_all_cameras() -> List[Dict[str, Any]]: +# Columns the map / live-data path actually needs — skip refresh_rate / timestamps. +_CAMERA_SELECT_COLS = ( + "id", + "source_agency", + "lat", + "lon", + "direction_facing", + "media_url", + "media_type", +) + + +def get_camera_count() -> int: + """Cheap total for UI badges without loading the full table.""" + conn = sqlite3.connect(str(DB_PATH)) + try: + row = conn.execute("SELECT COUNT(*) FROM cameras").fetchone() + return int(row[0] if row else 0) + finally: + conn.close() + + +def get_all_cameras( + *, + south: float | None = None, + west: float | None = None, + north: float | None = None, + east: float | None = None, +) -> List[Dict[str, Any]]: + """Load map-facing camera rows, optionally SQL-scoped to a viewport. + + Without bounds, returns the full catalog (store still holds world list; + /api/live-data/fast applies #288 bbox + world-zoom sampling). + """ + cols = ", ".join(_CAMERA_SELECT_COLS) + sql = f"SELECT {cols} FROM cameras" + params: list[Any] = [] + if None not in (south, west, north, east): + # Inclusive bbox; antimeridian (west > east) uses OR on lon. + if west <= east: + sql += " WHERE lat BETWEEN ? AND ? AND lon BETWEEN ? AND ?" + params.extend([south, north, west, east]) + else: + sql += ( + " WHERE lat BETWEEN ? AND ?" + " AND (lon >= ? OR lon <= ?)" + ) + params.extend([south, north, west, east]) + conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row - cursor = conn.cursor() - cursor.execute("SELECT * FROM cameras") - rows = cursor.fetchall() - conn.close() + try: + rows = conn.execute(sql, params).fetchall() + finally: + conn.close() + cameras = [] for row in rows: - cam = dict(row) - cam["media_type"] = str(cam.get("media_type") or _detect_media_type(cam.get("media_url", "")) or "image") + cam = {k: row[k] for k in _CAMERA_SELECT_COLS} + cam["media_type"] = str( + cam.get("media_type") or _detect_media_type(cam.get("media_url", "")) or "image" + ) cameras.append(cam) return cameras diff --git a/backend/services/fetchers/_store.py b/backend/services/fetchers/_store.py index 33e9737..135dd8f 100644 --- a/backend/services/fetchers/_store.py +++ b/backend/services/fetchers/_store.py @@ -146,6 +146,123 @@ source_timestamps = {} source_freshness: dict[str, dict] = {} +# Layers that support row-level live-data deltas (P2). +_DELTA_LAYER_KEYS: frozenset[str] = frozenset( + { + "ships", + "commercial_flights", + "military_flights", + "tracked_flights", + "private_flights", + "private_jets", + } +) +# Ring of (layer_version, id→item) snapshots for delta computation. +_LAYER_ID_RING: dict[str, list[tuple[int, dict[str, Any]]]] = {} +_LAYER_ID_RING_MAX = 4 + + +def entity_id_for_layer(layer: str, item: dict) -> str: + """Stable entity id for delta upsert/delete keys.""" + if not isinstance(item, dict): + return "" + if layer == "ships": + return str(item.get("mmsi") or item.get("id") or "").strip() + return str( + item.get("icao24") or item.get("icao") or item.get("id") or item.get("hex") or "" + ).strip().lower() + + +def _track_fingerprint(item: dict) -> tuple: + lng = item.get("lng") + if lng is None: + lng = item.get("lon") + return ( + item.get("lat"), + lng, + item.get("heading") or item.get("hdg") or item.get("cog") or item.get("true_track"), + item.get("speed_knots") or item.get("sog") or item.get("spd"), + item.get("alt") or item.get("altitude") or item.get("alt_km"), + ) + + +def _capture_layer_id_map(layer: str, items: list) -> dict[str, Any]: + out: dict[str, Any] = {} + for item in items or []: + if not isinstance(item, dict): + continue + eid = entity_id_for_layer(layer, item) + if eid: + out[eid] = item + return out + + +def _record_delta_layer_snapshot_locked(layer: str) -> None: + """Caller must hold ``_data_lock``.""" + if layer not in _DELTA_LAYER_KEYS: + return + ver = _layer_versions.get(layer, 0) + val = latest_data.get(layer) + items = val if isinstance(val, list) else [] + id_map = _capture_layer_id_map(layer, items) + ring = _LAYER_ID_RING.setdefault(layer, []) + ring.append((ver, id_map)) + while len(ring) > _LAYER_ID_RING_MAX: + ring.pop(0) + + +def compute_layer_row_delta(layer: str, since_version: int) -> dict[str, Any] | None: + """Build upsert/delete delta vs a prior layer version. + + Returns ``None`` when the base version is no longer in the ring (client + must take a full snapshot). Returns an empty upsert/delete when unchanged. + """ + if layer not in _DELTA_LAYER_KEYS: + return None + try: + since = int(since_version) + except (TypeError, ValueError): + return None + + with _data_lock: + current_ver = int(_layer_versions.get(layer, 0) or 0) + if since == current_ver or since > current_ver: + return { + "upsert": [], + "delete": [], + "version": current_ver, + "unchanged": True, + } + ring = list(_LAYER_ID_RING.get(layer) or []) + base_map = None + for ver, id_map in ring: + if ver == since: + base_map = id_map + break + if base_map is None: + # Base version aged out of the ring — full snapshot required. + return None + if ring and ring[-1][0] == current_ver: + cur_map = ring[-1][1] + else: + val = latest_data.get(layer) + items = val if isinstance(val, list) else [] + cur_map = _capture_layer_id_map(layer, items) + + upsert: list[Any] = [] + for eid, item in cur_map.items(): + prev = base_map.get(eid) + if prev is None or _track_fingerprint(prev) != _track_fingerprint(item): + upsert.append(item) + delete = [eid for eid in base_map if eid not in cur_map] + return { + "upsert": upsert, + "delete": delete, + "version": current_ver, + "unchanged": not upsert and not delete, + } + + def _mark_fresh(*keys): """Record the current UTC time for one or more data source keys.""" now = datetime.utcnow().isoformat() @@ -159,6 +276,7 @@ def _mark_fresh(*keys): val = latest_data.get(k) count = len(val) if isinstance(val, list) else (1 if val is not None else 0) changed.append((k, _layer_versions[k], count)) + _record_delta_layer_snapshot_locked(k) # Publish partial fetch progress immediately so the frontend can # observe newly available data without waiting for the entire tier. _data_version += 1 @@ -273,7 +391,10 @@ def get_latest_data_subset(*keys: str) -> DashboardData: def get_latest_data_deepcopy_snapshot() -> DashboardData: - """Deep-copy the full dashboard for /api/health and legacy /api/live-data. + """Deep-copy the full dashboard for callers that need an isolated mutate-safe snap. + + Prefer ``get_latest_data_refs_snapshot`` / ``get_latest_data_subset_refs`` on + read-only hot paths (legacy ``/api/live-data`` uses refs + orjson). The per-value deepcopy runs OUTSIDE ``_data_lock`` so a large clone cannot block fetcher writers (#375). The store contract is replace-don't-mutate, @@ -310,6 +431,17 @@ def get_latest_data_subset_refs(*keys: str) -> DashboardData: return snap +def get_latest_data_refs_snapshot() -> DashboardData: + """Return a shallow dict of all top-level store refs (read-only callers). + + Copies the mapping under the lock without deep-copying values. Safe for + orjson serialization and other read-only consumers; callers MUST NOT + mutate nested objects. + """ + with _data_lock: + return {key: value for key, value in latest_data.items()} + + def get_source_timestamps_snapshot() -> dict[str, str]: """Return a stable copy of per-source freshness timestamps.""" with _data_lock: diff --git a/backend/services/layer_enable_refresh.py b/backend/services/layer_enable_refresh.py index 3052365..1fff1de 100644 --- a/backend/services/layer_enable_refresh.py +++ b/backend/services/layer_enable_refresh.py @@ -10,13 +10,14 @@ import logging logger = logging.getLogger(__name__) -# Inline — local DB / static files only. +# Inline — tiny local/static loads (ms). _INSTANT_LAYER_KEYS: frozenset[str] = frozenset( - {"cctv", "power_plants", "datacenters"} + {"power_plants", "datacenters"} ) -# Background — network-bound; may take seconds. +# Background — network-bound OR large local scans (full CCTV SELECT can stall +# the single uvicorn worker if run inline on enable). _SLOW_LAYER_KEYS: frozenset[str] = frozenset( - {"firms", "psk_reporter", "fishing_activity"} + {"cctv", "firms", "psk_reporter", "fishing_activity"} ) @@ -33,12 +34,6 @@ def _was_off_now_on(before: dict[str, bool], key: str) -> bool: def _instant_fetch(key: str) -> None: - if key == "cctv": - from services.fetchers.infrastructure import fetch_cctv - - fetch_cctv() - logger.info("CCTV loaded (layer enabled)") - return if key == "power_plants": from services.fetchers.infrastructure import fetch_power_plants @@ -55,6 +50,12 @@ def _instant_fetch(key: str) -> None: def _slow_fetch(key: str) -> None: + if key == "cctv": + from services.fetchers.infrastructure import fetch_cctv + + fetch_cctv() + logger.info("CCTV loaded (layer enabled)") + return if key == "firms": from services.fetchers.earth_observation import ( fetch_firms_country_fires, diff --git a/backend/services/mesh/mesh_hashchain.py b/backend/services/mesh/mesh_hashchain.py index c7420ad..771fd40 100644 --- a/backend/services/mesh/mesh_hashchain.py +++ b/backend/services/mesh/mesh_hashchain.py @@ -39,6 +39,7 @@ transports. """ import json +import orjson import os import time import hmac @@ -109,9 +110,30 @@ def _atomic_write_text(target: Path, content: str, encoding: str = "utf-8") -> N pass raise + +def _atomic_write_bytes(target: Path, content: bytes) -> None: + """Write binary content atomically via temp file + os.replace().""" + parent = target.parent + parent.mkdir(parents=True, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=str(parent), suffix=".tmp") + try: + with os.fdopen(fd, "wb") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_path, str(target)) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + DATA_DIR = Path(__file__).resolve().parents[2] / "data" CHAIN_FILE = DATA_DIR / "infonet.json" WAL_FILE = DATA_DIR / "infonet.wal" +CHAIN_COLD_DIR = DATA_DIR / "infonet_cold" GATE_STORE_DIR = DATA_DIR / "gate_messages" GATE_STORAGE_DOMAIN = "gates" @@ -1469,6 +1491,7 @@ class Infonet: self.event_index: dict[str, int] = {} # {event_id: index in events list} self.public_key_bindings: dict[str, str] = {} # {public_key: canonical node_id} self.revocations: dict[str, dict] = {} + self._cold_segments: list[dict] = [] # Archived prefix metadata (P8) self._replay_filter = ReplayFilter() self._last_validated_index: int = 0 # For incremental validation # Running counters — avoid O(N) scans in get_info() @@ -1549,9 +1572,15 @@ class Infonet: self.head_hash = data.get("head_hash", GENESIS_HASH) self.node_sequences = data.get("node_sequences", {}) self.sequence_domains = data.get("sequence_domains", {}) + cold = data.get("cold_segments") or [] + self._cold_segments = list(cold) if isinstance(cold, list) else [] self._rebuild_state() self._rebuild_revocations() self._rebuild_counters() + # Honor MAX_CHAIN_MEMORY after load (archive overflow, keep hot window). + if len(self.events) > MAX_CHAIN_MEMORY: + self._enforce_memory_cap() + self._flush() logger.info( f"Loaded Infonet: {len(self.events)} events, head={self.head_hash[:16]}..." ) @@ -1794,6 +1823,7 @@ class Infonet: and schedule a single write after _SAVE_INTERVAL seconds. Multiple rapid calls collapse into one I/O operation. """ + self._enforce_memory_cap() self._dirty = True with self._save_lock: if self._save_timer is None or not self._save_timer.is_alive(): @@ -1807,6 +1837,8 @@ class Infonet: Sprint 2 / Rec #8: clears the WAL only after the chain file has been durably written. A crash before _flush() succeeds leaves the WAL in place so _replay_wal() can recover on next boot. + + P8: compact orjson (no indent=2) — flush cost scales with chain size. """ if not self._dirty: return @@ -1818,9 +1850,10 @@ class Infonet: "head_hash": self.head_hash, "node_sequences": self.node_sequences, "sequence_domains": self.sequence_domains, + "cold_segments": list(getattr(self, "_cold_segments", []) or []), "events": self.events, } - _atomic_write_text(CHAIN_FILE, json.dumps(data, indent=2), encoding="utf-8") + _atomic_write_bytes(CHAIN_FILE, orjson.dumps(data, option=orjson.OPT_NON_STR_KEYS)) self._dirty = False # Chain file is now durable — safe to retire the WAL entry. self._clear_wal() @@ -1837,13 +1870,56 @@ class Infonet: "head_hash": self.head_hash, "node_sequences": self.node_sequences, "sequence_domains": self.sequence_domains, + "cold_segments": list(getattr(self, "_cold_segments", []) or []), "events": self.events, } - _atomic_write_text(CHAIN_FILE, json.dumps(data, indent=2), encoding="utf-8") + _atomic_write_bytes(CHAIN_FILE, orjson.dumps(data, option=orjson.OPT_NON_STR_KEYS)) except Exception as e: logger.error(f"Failed to materialize Infonet: {e}") raise + def _enforce_memory_cap(self) -> None: + """Spill oldest events to cold segments when RAM exceeds MAX_CHAIN_MEMORY. + + History is archived on disk (not deleted). Hot chain + indexes stay + coherent for head/sync; deep history remains in cold segments. + """ + if len(self.events) <= MAX_CHAIN_MEMORY: + return + overflow = len(self.events) - MAX_CHAIN_MEMORY + cold = self.events[:overflow] + CHAIN_COLD_DIR.mkdir(parents=True, exist_ok=True) + if not hasattr(self, "_cold_segments") or self._cold_segments is None: + self._cold_segments = [] + seg_no = len(self._cold_segments) + filename = f"cold_{seg_no:08d}.orjson" + path = CHAIN_COLD_DIR / filename + _atomic_write_bytes(path, orjson.dumps(cold, option=orjson.OPT_NON_STR_KEYS)) + self._cold_segments.append( + { + "filename": filename, + "count": len(cold), + "first_id": str(cold[0].get("event_id", "") or "") if cold else "", + "last_id": str(cold[-1].get("event_id", "") or "") if cold else "", + } + ) + self.events = self.events[overflow:] + # Rebuild indexes for the hot window only. + self.event_index = { + str(evt.get("event_id", "") or ""): idx + for idx, evt in enumerate(self.events) + if evt.get("event_id") + } + self._rebuild_counters() + self._invalidate_merkle_cache() + self._dirty = True + logger.info( + "Infonet memory cap: archived %d events to %s (hot=%d)", + overflow, + filename, + len(self.events), + ) + def confirmations_for_event(self, event_id: str) -> int: idx = self.event_index.get(event_id) if idx is None: diff --git a/backend/services/openclaw_watchdog.py b/backend/services/openclaw_watchdog.py index dc8270e..15f5262 100644 --- a/backend/services/openclaw_watchdog.py +++ b/backend/services/openclaw_watchdog.py @@ -112,9 +112,9 @@ def _evaluate_watches() -> list[dict[str, Any]]: # Load telemetry once for all watches try: - from services.telemetry import get_cached_telemetry, get_cached_slow_telemetry - fast = get_cached_telemetry() or {} - slow = get_cached_slow_telemetry() or {} + from services.telemetry import get_cached_telemetry_refs, get_cached_slow_telemetry_refs + fast = get_cached_telemetry_refs() or {} + slow = get_cached_slow_telemetry_refs() or {} except Exception: return [] diff --git a/backend/tests/test_issue_375_production_hardening.py b/backend/tests/test_issue_375_production_hardening.py index 8d655ec..8fe8f99 100644 --- a/backend/tests/test_issue_375_production_hardening.py +++ b/backend/tests/test_issue_375_production_hardening.py @@ -76,6 +76,7 @@ class TestLiveDataFullEndpoint: _store.latest_data["sigint"] = [ {"source": "aprs", "observed": datetime(2026, 1, 1, tzinfo=timezone.utc)}, ] + _store.bump_data_version() try: r = client.get("/api/live-data/fast") assert r.status_code == 200 @@ -83,6 +84,7 @@ class TestLiveDataFullEndpoint: finally: with _store._data_lock: _store.latest_data["sigint"] = prior + _store.bump_data_version() def test_live_data_serializes_non_json_native_values(self, client): from datetime import datetime, timezone @@ -94,6 +96,7 @@ class TestLiveDataFullEndpoint: _store.latest_data["gdelt"] = [ {"observed": datetime(2026, 1, 1, tzinfo=timezone.utc)}, ] + _store.bump_data_version() try: r = client.get("/api/live-data") assert r.status_code == 200 @@ -101,6 +104,7 @@ class TestLiveDataFullEndpoint: finally: with _store._data_lock: _store.latest_data["gdelt"] = prior + _store.bump_data_version() class TestSlowTaskConcurrency: diff --git a/backend/tests/test_layer_enable_refresh.py b/backend/tests/test_layer_enable_refresh.py index 0beff0b..a2518e6 100644 --- a/backend/tests/test_layer_enable_refresh.py +++ b/backend/tests/test_layer_enable_refresh.py @@ -38,7 +38,8 @@ def test_refresh_skips_when_layer_stays_off(): fetch_cctv.assert_not_called() -def test_refresh_cctv_runs_inline(): +def test_refresh_cctv_runs_on_slow_executor(): + """CCTV SELECT can be large — never block the API worker on enable.""" before = {**snapshot_active_layers(), "cctv": False} active_layers["cctv"] = True @@ -49,8 +50,9 @@ def test_refresh_cctv_runs_inline(): ): refresh_newly_enabled_layers(before) - fetch_cctv.assert_called_once() - bump.assert_called_once() - slow_exec.submit.assert_not_called() + fetch_cctv.assert_not_called() + bump.assert_not_called() + slow_exec.submit.assert_called_once() + assert slow_exec.submit.call_args[0][1] == ("cctv",) active_layers["cctv"] = before.get("cctv", False) diff --git a/backend/tests/test_live_data_etag_cache.py b/backend/tests/test_live_data_etag_cache.py new file mode 100644 index 0000000..90721c2 --- /dev/null +++ b/backend/tests/test_live_data_etag_cache.py @@ -0,0 +1,44 @@ +"""Unit tests for ETag-keyed live-data orjson byte cache (P4).""" +from __future__ import annotations + +from routers import data as data_router + + +def test_cached_live_data_bytes_invokes_factory_once_per_etag(): + calls = {"n": 0} + + def _build(): + calls["n"] += 1 + return {"ok": True, "n": calls["n"]} + + etag = "test|etag|cache-hit" + # Isolate from other tests / endpoint warm-up. + with data_router._LIVE_DATA_BYTES_CACHE_LOCK: + data_router._LIVE_DATA_BYTES_CACHE.pop(etag, None) + + first = data_router._cached_live_data_bytes(etag, _build) + second = data_router._cached_live_data_bytes(etag, _build) + + assert first == second + assert calls["n"] == 1 + assert b'"ok"' in first + + +def test_cached_live_data_bytes_misses_on_etag_change(): + calls = {"n": 0} + + def _build(): + calls["n"] += 1 + return {"n": calls["n"]} + + etag_a = "test|etag|a" + etag_b = "test|etag|b" + with data_router._LIVE_DATA_BYTES_CACHE_LOCK: + data_router._LIVE_DATA_BYTES_CACHE.pop(etag_a, None) + data_router._LIVE_DATA_BYTES_CACHE.pop(etag_b, None) + + a = data_router._cached_live_data_bytes(etag_a, _build) + b = data_router._cached_live_data_bytes(etag_b, _build) + + assert a != b + assert calls["n"] == 2 diff --git a/backend/tests/test_live_data_viewport_bbox.py b/backend/tests/test_live_data_viewport_bbox.py index 2cf6ff8..316b9c1 100644 --- a/backend/tests/test_live_data_viewport_bbox.py +++ b/backend/tests/test_live_data_viewport_bbox.py @@ -24,6 +24,12 @@ class TestFastBboxFiltering: def _seed_fast(self, monkeypatch): """Plant deterministic heavy + light fixtures across the globe.""" from services.fetchers import _store + from routers import data as data_router + + # Avoid cross-test ETag byte-cache pollution. + with data_router._LIVE_DATA_BYTES_CACHE_LOCK: + data_router._LIVE_DATA_BYTES_CACHE.clear() + _store.bump_data_version() # Heavy collections: dense across the world. commercial = [ @@ -35,7 +41,11 @@ class TestFastBboxFiltering: {"lat": -60.0, "lng": -120.0, "id": "s-sw"}, {"lat": 35.0, "lng": -75.0, "id": "s-ne"}, ] - cctv = [{"lat": 35.0, "lng": -75.0, "id": "c-1"}] + # Real CCTV rows use ``lon`` (not ``lng``) — bbox must honor that alias. + cctv = [ + {"lat": 35.0, "lon": -75.0, "id": "c-1"}, + {"lat": 35.0, "lon": 100.0, "id": "c-asia"}, + ] # Sigint heavy collection. sigint = [ @@ -85,7 +95,7 @@ class TestFastBboxFiltering: # Heavy layers: only the eastern-US fixture survives. assert {f["id"] for f in data["commercial_flights"]} == {"f-ne"} assert {s["id"] for s in data["ships"]} == {"s-ne"} - assert {c["id"] for c in data["cctv"]} == {"c-1"} + assert {c["id"] for c in data["cctv"]} == {"c-1"} # lon alias filtered assert {s["id"] for s in data["sigint"]} == {"sig-east"} def test_bbox_does_not_filter_light_layers(self, client, monkeypatch): diff --git a/backend/tests/test_perf_guardrail_caps.py b/backend/tests/test_perf_guardrail_caps.py new file mode 100644 index 0000000..8d2b182 --- /dev/null +++ b/backend/tests/test_perf_guardrail_caps.py @@ -0,0 +1,126 @@ +"""P5 / P11 guardrail tests — zoom-aware caps + CCTV lon bbox + enable path.""" +from __future__ import annotations + + +def test_sample_items_even_stride(): + from routers.data import _sample_items + + items = [{"id": i} for i in range(100)] + sampled = _sample_items(items, 10) + assert len(sampled) == 10 + assert sampled[0]["id"] == 0 + assert sampled[-1]["id"] == 90 + + +def test_world_zoom_caps_dense_layers_and_keeps_totals(): + from routers.data import _cap_fast_dashboard_payload + + flights = [{"lat": 0.0, "lng": float(i), "id": f"f-{i}"} for i in range(2000)] + cctv = [{"lat": 0.0, "lon": float(i % 180), "id": f"c-{i}"} for i in range(900)] + payload = { + "commercial_flights": flights, + "ships": [{"lat": 1.0, "lng": 1.0, "id": "s1"}], + "cctv": cctv, + "cctv_total": 900, + "satellites": [{"lat": -10.0, "lng": 10.0, "id": "sat"}], # never capped + } + out = _cap_fast_dashboard_payload(payload) + assert out["payload_scale"] == "world" + assert out["payload_sampled"] is True + assert len(out["commercial_flights"]) == 1200 + assert out["layer_totals"]["commercial_flights"] == 2000 + assert len(out["cctv"]) == 600 + assert out["cctv_total"] == 900 + assert len(out["satellites"]) == 1 + + +def test_regional_zoom_does_not_cap(): + from routers.data import _cap_fast_dashboard_payload + + flights = [{"lat": 35.0, "lng": -75.0, "id": f"f-{i}"} for i in range(1500)] + out = _cap_fast_dashboard_payload( + {"commercial_flights": flights}, + s=30, + w=-80, + n=40, + e=-70, + ) + assert out["payload_scale"] == "regional" + assert "payload_sampled" not in out + assert len(out["commercial_flights"]) == 1500 + + +def test_bbox_filter_respects_lon_alias(): + from routers.data import _bbox_filter + + items = [ + {"id": "in", "lat": 35.0, "lon": -75.0}, + {"id": "out", "lat": 35.0, "lon": 100.0}, + ] + filtered = _bbox_filter(items, 30, -80, 40, -70) + assert {c["id"] for c in filtered} == {"in"} + + +def test_get_all_cameras_column_subset(tmp_path, monkeypatch): + import sqlite3 + + from services import cctv_pipeline + + db = tmp_path / "cctv.db" + monkeypatch.setattr(cctv_pipeline, "DB_PATH", db) + conn = sqlite3.connect(str(db)) + conn.execute( + """ + CREATE TABLE cameras ( + id TEXT PRIMARY KEY, + source_agency TEXT, + lat REAL, + lon REAL, + direction_facing TEXT, + media_url TEXT, + media_type TEXT, + refresh_rate_seconds INTEGER, + last_updated TIMESTAMP, + secret_extra TEXT + ) + """ + ) + conn.execute( + "INSERT INTO cameras VALUES (?,?,?,?,?,?,?,?,?,?)", + ("cam-1", "dot", 35.0, -75.0, "N", "https://example.com/a.jpg", "image", 60, None, "nope"), + ) + conn.execute( + "INSERT INTO cameras VALUES (?,?,?,?,?,?,?,?,?,?)", + ("cam-2", "dot", 10.0, 10.0, "S", "https://example.com/b.jpg", "image", 60, None, "nope"), + ) + conn.commit() + conn.close() + + all_cams = cctv_pipeline.get_all_cameras() + assert len(all_cams) == 2 + assert "secret_extra" not in all_cams[0] + assert set(all_cams[0].keys()) <= set(cctv_pipeline._CAMERA_SELECT_COLS) + + boxed = cctv_pipeline.get_all_cameras(south=30, west=-80, north=40, east=-70) + assert [c["id"] for c in boxed] == ["cam-1"] + assert cctv_pipeline.get_camera_count() == 2 + + +def test_live_data_fast_world_samples_under_cap(client, monkeypatch): + from services.fetchers import _store + from routers import data as data_router + + with data_router._LIVE_DATA_BYTES_CACHE_LOCK: + data_router._LIVE_DATA_BYTES_CACHE.clear() + _store.bump_data_version() + + flights = [{"lat": 0.0, "lng": float(i % 170), "id": f"f-{i}"} for i in range(1500)] + monkeypatch.setitem(_store.latest_data, "commercial_flights", flights) + monkeypatch.setitem(_store.active_layers, "flights", True) + + r = client.get("/api/live-data/fast") + assert r.status_code == 200 + data = r.json() + assert data.get("payload_scale") == "world" + assert len(data["commercial_flights"]) == 1200 + assert data["layer_totals"]["commercial_flights"] == 1500 diff --git a/backend/tests/test_perf_p2_p8_p12.py b/backend/tests/test_perf_p2_p8_p12.py new file mode 100644 index 0000000..e6c511e --- /dev/null +++ b/backend/tests/test_perf_p2_p8_p12.py @@ -0,0 +1,148 @@ +"""P2 live-data delta + P8 hashchain memory + P12 mesh-only import guards.""" +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import patch + + +def test_compute_layer_row_delta_upsert_and_delete(monkeypatch): + from services.fetchers import _store + + monkeypatch.setitem(_store.latest_data, "ships", [ + {"mmsi": "1", "lat": 1.0, "lng": 1.0}, + {"mmsi": "2", "lat": 2.0, "lng": 2.0}, + ]) + _store._LAYER_ID_RING.clear() + with _store._data_lock: + _store._layer_versions["ships"] = 1 + _store._record_delta_layer_snapshot_locked("ships") + + monkeypatch.setitem(_store.latest_data, "ships", [ + {"mmsi": "1", "lat": 1.5, "lng": 1.0}, # moved + {"mmsi": "3", "lat": 3.0, "lng": 3.0}, # new + ]) + with _store._data_lock: + _store._layer_versions["ships"] = 2 + _store._record_delta_layer_snapshot_locked("ships") + + delta = _store.compute_layer_row_delta("ships", 1) + assert delta is not None + assert {u["mmsi"] for u in delta["upsert"]} == {"1", "3"} + assert delta["delete"] == ["2"] + assert delta["version"] == 2 + + +def test_compute_layer_row_delta_returns_none_when_base_missing(): + from services.fetchers import _store + + _store._LAYER_ID_RING.clear() + with _store._data_lock: + _store._layer_versions["ships"] = 5 + # Client holds an older version that is no longer in the ring. + assert _store.compute_layer_row_delta("ships", 1) is None + + +def test_live_data_fast_delta_mode(client, monkeypatch): + from services.fetchers import _store + + ships_v1 = [ + {"mmsi": "10", "lat": 10.0, "lng": 10.0}, + {"mmsi": "20", "lat": 20.0, "lng": 20.0}, + ] + monkeypatch.setitem(_store.latest_data, "ships", ships_v1) + monkeypatch.setitem(_store.active_layers, "ships_military", True) + _store._LAYER_ID_RING.clear() + with _store._data_lock: + _store._layer_versions["ships"] = 1 + for key in ( + "commercial_flights", + "military_flights", + "tracked_flights", + "private_flights", + "private_jets", + "cctv", + "uavs", + "liveuamap", + "gps_jamming", + "satellites", + "sigint", + "trains", + ): + _store._layer_versions.setdefault(key, 1) + _store._record_delta_layer_snapshot_locked("ships") + for key in ( + "commercial_flights", + "military_flights", + "tracked_flights", + "private_flights", + "private_jets", + ): + monkeypatch.setitem(_store.latest_data, key, []) + _store._record_delta_layer_snapshot_locked(key) + + # Advance ships + monkeypatch.setitem( + _store.latest_data, + "ships", + [ + {"mmsi": "10", "lat": 10.5, "lng": 10.0}, + {"mmsi": "30", "lat": 30.0, "lng": 30.0}, + ], + ) + with _store._data_lock: + _store._layer_versions["ships"] = 2 + _store._record_delta_layer_snapshot_locked("ships") + + lv = ( + "ships:1,commercial_flights:1,military_flights:1,tracked_flights:1," + "private_flights:1,private_jets:1,cctv:1,uavs:1,liveuamap:1," + "gps_jamming:1,satellites:1,sigint:1,trains:1" + ) + r = client.get(f"/api/live-data/fast?lv={lv}") + assert r.status_code == 200 + data = r.json() + assert data["mode"] == "delta" + assert "ships" in data["deltas"] + assert {u["mmsi"] for u in data["deltas"]["ships"]["upsert"]} == {"10", "30"} + assert data["deltas"]["ships"]["delete"] == ["20"] + + +def test_hashchain_flush_uses_orjson_compact(tmp_path, monkeypatch): + from services.mesh import mesh_hashchain as hc + + monkeypatch.setattr(hc, "DATA_DIR", tmp_path) + monkeypatch.setattr(hc, "CHAIN_FILE", tmp_path / "infonet.json") + monkeypatch.setattr(hc, "WAL_FILE", tmp_path / "infonet.wal") + monkeypatch.setattr(hc, "CHAIN_COLD_DIR", tmp_path / "infonet_cold") + monkeypatch.setattr(hc, "MAX_CHAIN_MEMORY", 3) + + net = hc.Infonet() + net.events = [{"event_id": f"e{i}", "event_type": "message", "payload": {}} for i in range(5)] + net.head_hash = "e4" + net._dirty = True + net._enforce_memory_cap() + net._flush() + + raw = (tmp_path / "infonet.json").read_bytes() + assert b"\n " not in raw # compact, not indent=2 + data = json.loads(raw) + assert len(data["events"]) == 3 + assert data["cold_segments"] + cold_file = tmp_path / "infonet_cold" / data["cold_segments"][0]["filename"] + assert cold_file.exists() + cold = json.loads(cold_file.read_bytes()) + assert len(cold) == 2 + + +def test_mesh_only_skips_osint_router_modules(): + """Source-level guard: MESH_ONLY branch must not load routers.data at import.""" + import inspect + from pathlib import Path + + src = Path("backend/main.py").read_text(encoding="utf-8") + assert "if _MESH_ONLY:" in src + assert 'data_router = APIRouter()' in src + assert "from services.data_fetcher import" in src + # data_fetcher import is gated + assert "if not _MESH_ONLY:" in src diff --git a/backend/tests/test_perf_safe_optimizations.py b/backend/tests/test_perf_safe_optimizations.py index 3845378..3699916 100644 --- a/backend/tests/test_perf_safe_optimizations.py +++ b/backend/tests/test_perf_safe_optimizations.py @@ -33,6 +33,25 @@ def test_health_uses_subset_refs_not_full_deepcopy(): assert "deepcopy" not in snap_source +def test_legacy_live_data_uses_refs_snapshot_not_deepcopy(): + from routers import data as data_router + + source = inspect.getsource(data_router.live_data) + assert "get_latest_data_refs_snapshot" in source + assert "get_latest_data_deepcopy_snapshot" not in source + assert "deepcopy" not in source + + +def test_openclaw_watchdog_uses_telemetry_refs(): + from services import openclaw_watchdog + + source = inspect.getsource(openclaw_watchdog._evaluate_watches) + assert "get_cached_telemetry_refs" in source + assert "get_cached_slow_telemetry_refs" in source + assert "get_cached_telemetry()" not in source + assert "get_cached_slow_telemetry()" not in source + + def test_active_layers_defaults_match_dashboard_first_paint(): """Backend must not prefetch layers the dashboard starts with disabled.""" from services.fetchers import _store diff --git a/frontend/src/__tests__/hooks/useDataPollingViewport.test.ts b/frontend/src/__tests__/hooks/useDataPollingViewport.test.ts index 39121c1..246e089 100644 --- a/frontend/src/__tests__/hooks/useDataPollingViewport.test.ts +++ b/frontend/src/__tests__/hooks/useDataPollingViewport.test.ts @@ -23,4 +23,13 @@ describe('viewport fast refetch wiring', () => { expect(handler).toHaveBeenCalledTimes(1); window.removeEventListener(VIEWPORT_COMMITTED_EVENT, handler); }); + + it('liveDataBoundsKey changes when the operator pans to a new region', async () => { + const { liveDataBoundsKey } = await import('@/lib/liveDataViewport'); + const before = liveDataBoundsKey(); + setLiveDataBounds({ south: 20, west: -130, north: 55, east: -60 }); + const after = liveDataBoundsKey(); + expect(before).not.toBe(after); + expect(after).toBe('20,-130,55,-60'); + }); }); diff --git a/frontend/src/__tests__/hooks/useDataStoreDelta.test.ts b/frontend/src/__tests__/hooks/useDataStoreDelta.test.ts new file mode 100644 index 0000000..fca42d4 --- /dev/null +++ b/frontend/src/__tests__/hooks/useDataStoreDelta.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from 'vitest'; +import { applyLayerDeltas, mergeData } from '@/hooks/useDataStore'; + +describe('applyLayerDeltas', () => { + it('upserts and deletes by entity id', () => { + mergeData({ + ships: [ + { mmsi: '1', lat: 1, lng: 1 }, + { mmsi: '2', lat: 2, lng: 2 }, + ], + }); + const ok = applyLayerDeltas({ + ships: { + upsert: [{ mmsi: '1', lat: 1.5, lng: 1 }], + delete: ['2'], + }, + }); + expect(ok).toBe(true); + // Re-read via merge of empty would not work; pull through another merge fingerprint + mergeData({}); + // Store is module singleton — verify via applying a no-op and checking through + // a follow-up delta that assumes state. + const ok2 = applyLayerDeltas({ + ships: { + upsert: [{ mmsi: '3', lat: 3, lng: 3 }], + delete: [], + }, + }); + expect(ok2).toBe(true); + }); +}); diff --git a/frontend/src/__tests__/hooks/useDataStoreMerge.test.ts b/frontend/src/__tests__/hooks/useDataStoreMerge.test.ts new file mode 100644 index 0000000..2480f1e --- /dev/null +++ b/frontend/src/__tests__/hooks/useDataStoreMerge.test.ts @@ -0,0 +1,120 @@ +import { act, cleanup, renderHook } from '@testing-library/react'; +import { afterEach, describe, expect, it } from 'vitest'; + +import { + arrayFingerprint, + mergeData, + useDataKey, + valuesEquivalent, +} from '@/hooks/useDataStore'; + +afterEach(() => { + cleanup(); +}); + +describe('arrayFingerprint', () => { + it('encodes length and id/lat/lng samples', () => { + const a = [ + { icao: 'ABC123', lat: 40.1234, lng: -74.5678 }, + { icao: 'DEF456', lat: 41.0, lng: -73.0 }, + ]; + expect(arrayFingerprint(a)).toBe('2|ABC123:4012:-7457|DEF456:4100:-7300'); + }); + + it('falls back through id/mmsi/hex/callsign/name', () => { + expect(arrayFingerprint([{ mmsi: '123', lat: 1.006, lng: 2.004 }])).toBe( + '1|123:101:200', + ); + expect(arrayFingerprint([{ hex: 'A1', lat: null, lng: null }])).toBe('1|A1::'); + }); +}); + +describe('valuesEquivalent', () => { + it('returns true for identical references and empty arrays', () => { + const arr: unknown[] = []; + expect(valuesEquivalent(arr, arr)).toBe(true); + expect(valuesEquivalent([], [])).toBe(true); + expect(valuesEquivalent(null, null)).toBe(true); + expect(valuesEquivalent(undefined, null)).toBe(false); + }); + + it('treats flight arrays with same ids/positions as equivalent', () => { + const prev = [{ icao: 'N1', lat: 10.001, lng: 20.002 }]; + const next = [{ icao: 'N1', lat: 10.004, lng: 20.002 }]; // rounds to same *100 + expect(valuesEquivalent(prev, next)).toBe(true); + }); + + it('detects position changes beyond rounding', () => { + const prev = [{ icao: 'N1', lat: 10.0, lng: 20.0 }]; + const next = [{ icao: 'N1', lat: 10.02, lng: 20.0 }]; + expect(valuesEquivalent(prev, next)).toBe(false); + }); + + it('shallow-compares plain objects', () => { + expect(valuesEquivalent({ a: 1, b: 2 }, { a: 1, b: 2 })).toBe(true); + expect(valuesEquivalent({ a: 1 }, { a: 1, b: 2 })).toBe(false); + expect(valuesEquivalent({ a: 1 }, { a: 2 })).toBe(false); + }); + + it('returns false for different lengths', () => { + expect(valuesEquivalent([{ id: 1 }], [{ id: 1 }, { id: 2 }])).toBe(false); + }); + + it('does not fingerprint non-geo arrays (news etc. still notify)', () => { + const prev = [{ id: 'a', title: 'Old headline' }]; + const next = [{ id: 'a', title: 'New headline' }]; + expect(valuesEquivalent(prev, next)).toBe(false); + }); +}); + +describe('mergeData stable references', () => { + it('keeps the previous array reference when content fingerprint matches', () => { + const first = [{ icao: 'KEEP', lat: 51.5, lng: -0.12 }]; + act(() => { + mergeData({ commercial_flights: first }); + }); + + const { result } = renderHook(() => useDataKey('commercial_flights')); + expect(result.current).toBe(first); + + act(() => { + mergeData({ + commercial_flights: [{ icao: 'KEEP', lat: 51.5, lng: -0.12 }], + }); + }); + expect(result.current).toBe(first); + }); + + it('replaces the reference when a tracked position moves', () => { + const first = [{ icao: 'MOVE', lat: 1.0, lng: 2.0 }]; + act(() => { + mergeData({ ships: first }); + }); + + const { result } = renderHook(() => useDataKey('ships')); + expect(result.current).toBe(first); + + const second = [{ icao: 'MOVE', lat: 1.5, lng: 2.0 }]; + act(() => { + mergeData({ ships: second }); + }); + expect(result.current).toBe(second); + expect(result.current).not.toBe(first); + }); + + it('does not notify for shallow-equivalent objects', () => { + const threat = { score: 3, level: 'ELEVATED' as const, color: '#f90', drivers: [] as string[] }; + act(() => { + mergeData({ threat_level: threat }); + }); + const { result } = renderHook(() => useDataKey('threat_level')); + expect(result.current).toBe(threat); + + act(() => { + mergeData({ + threat_level: { score: 3, level: 'ELEVATED', color: '#f90', drivers: threat.drivers }, + }); + }); + expect(result.current).toBe(threat); + }); +}); diff --git a/frontend/src/__tests__/map/applyDynamicLayerInterp.test.ts b/frontend/src/__tests__/map/applyDynamicLayerInterp.test.ts new file mode 100644 index 0000000..38c00e1 --- /dev/null +++ b/frontend/src/__tests__/map/applyDynamicLayerInterp.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from 'vitest'; +import { applyDynamicLayerInterp } from '@/components/map/applyDynamicLayerInterp'; +import type { DynamicMapLayersResult } from '@/components/map/dynamicMapLayers.worker'; + +function pointLayer( + props: Record, + coordinates: [number, number], +): GeoJSON.FeatureCollection { + return { + type: 'FeatureCollection', + features: [ + { + type: 'Feature', + properties: props, + geometry: { type: 'Point', coordinates }, + }, + ], + }; +} + +describe('applyDynamicLayerInterp', () => { + it('leaves layers unchanged when dtSeconds is 0', () => { + const layers: DynamicMapLayersResult = { + commercialFlightsGeoJSON: pointLayer( + { kind: 'flight', baseLat: 40, baseLng: -74, spd: 400, hdg: 90, alt: 30000 }, + [-74, 40], + ), + privateFlightsGeoJSON: null, + privateJetsGeoJSON: null, + militaryFlightsGeoJSON: null, + trackedFlightsGeoJSON: null, + shipsGeoJSON: null, + meshtasticGeoJSON: null, + aprsGeoJSON: null, + }; + expect(applyDynamicLayerInterp(layers, 0)).toBe(layers); + }); + + it('dead-reckons flight coordinates between polls', () => { + const layers: DynamicMapLayersResult = { + commercialFlightsGeoJSON: pointLayer( + { kind: 'flight', baseLat: 40, baseLng: -74, spd: 400, hdg: 90, alt: 30000 }, + [-74, 40], + ), + privateFlightsGeoJSON: null, + privateJetsGeoJSON: null, + militaryFlightsGeoJSON: null, + trackedFlightsGeoJSON: null, + shipsGeoJSON: null, + meshtasticGeoJSON: null, + aprsGeoJSON: null, + }; + const next = applyDynamicLayerInterp(layers, 30); + const coords = next.commercialFlightsGeoJSON?.features[0].geometry; + expect(coords && coords.type === 'Point' ? coords.coordinates[0] : null).not.toBe(-74); + expect(coords && coords.type === 'Point' ? coords.coordinates[1] : null).toBeCloseTo(40, 1); + }); + + it('does not move grounded flights', () => { + const layers: DynamicMapLayersResult = { + commercialFlightsGeoJSON: pointLayer( + { kind: 'flight', baseLat: 40, baseLng: -74, spd: 20, hdg: 90, alt: 50 }, + [-74, 40], + ), + privateFlightsGeoJSON: null, + privateJetsGeoJSON: null, + militaryFlightsGeoJSON: null, + trackedFlightsGeoJSON: null, + shipsGeoJSON: null, + meshtasticGeoJSON: null, + aprsGeoJSON: null, + }; + const next = applyDynamicLayerInterp(layers, 30); + expect(next.commercialFlightsGeoJSON).toBe(layers.commercialFlightsGeoJSON); + }); +}); diff --git a/frontend/src/app/layout.tsx b/frontend/src/app/layout.tsx index 44f6819..b38d4c7 100644 --- a/frontend/src/app/layout.tsx +++ b/frontend/src/app/layout.tsx @@ -1,6 +1,7 @@ import type { Metadata } from 'next'; import { JetBrains_Mono } from 'next/font/google'; import DesktopBridgeBootstrap from '@/components/DesktopBridgeBootstrap'; +import MotionRoot from '@/components/MotionRoot'; import { ThemeProvider } from '@/lib/ThemeContext'; import { I18nProvider } from '@/i18n'; import './globals.css'; @@ -33,8 +34,10 @@ export default function RootLayout({ - - {children} + + + {children} + diff --git a/frontend/src/app/page.tsx b/frontend/src/app/page.tsx index 4abbcb5..8931f49 100644 --- a/frontend/src/app/page.tsx +++ b/frontend/src/app/page.tsx @@ -2,7 +2,7 @@ import { useEffect, useState, useRef, useCallback, useMemo } from 'react'; import dynamic from 'next/dynamic'; -import { motion } from 'framer-motion'; +import { motion } from '@/lib/motion'; import { ChevronLeft, ChevronRight, ChevronUp, ChevronDown } from 'lucide-react'; import WorldviewLeftPanel from '@/components/WorldviewLeftPanel'; @@ -12,12 +12,9 @@ import FilterPanel from '@/components/FilterPanel'; import FindLocateBar from '@/components/FindLocateBar'; import TopRightControls from '@/components/TopRightControls'; import TimelinePanel from '@/components/TimelinePanel'; -import SettingsPanel from '@/components/SettingsPanel'; import MapLegend from '@/components/MapLegend'; import ScaleBar from '@/components/ScaleBar'; -import MeshTerminal from '@/components/MeshTerminal'; import MeshChat from '@/components/MeshChat'; -import InfonetTerminal from '@/components/InfonetTerminal'; import { endInfonetTerminalSession } from '@/lib/infonetTerminalSession'; import ShodanPanel from '@/components/ShodanPanel'; import ReconPanel from '@/components/ReconPanel'; @@ -64,6 +61,10 @@ import SarAoiEditorModal from '@/components/SarAoiEditorModal'; // Use dynamic loads for Maplibre to avoid SSR window is not defined errors const MaplibreViewer = dynamic(() => import('@/components/MaplibreViewer'), { ssr: false }); +// Heavy panels — defer until opened so they stay out of the critical path +const SettingsPanel = dynamic(() => import('@/components/SettingsPanel'), { ssr: false }); +const MeshTerminal = dynamic(() => import('@/components/MeshTerminal'), { ssr: false }); +const InfonetTerminal = dynamic(() => import('@/components/InfonetTerminal'), { ssr: false }); // LocateBar and SentinelInfoModal extracted to page-local modules (Sprint 4B) @@ -447,6 +448,24 @@ export default function Dashboard() { [], ); + const handleExpandEntityGraph = useCallback(() => { + if (isEntityGraphEligible(selectedEntity)) setShowEntityGraph(true); + }, [selectedEntity]); + + const handleArticleClick = useCallback( + (idx: number, lat?: number, lng?: number, title?: string) => { + if (lat !== undefined && lng !== undefined) { + setFlyToLocation({ lat, lng, ts: Date.now() }); + // Also highlight the corresponding map alert + if (title) { + const alertKey = `${title}|${lat},${lng}`; + setSelectedEntity({ id: alertKey, type: 'news' }); + } + } + }, + [], + ); + const handleMeasureClick = useCallback( (pt: { lat: number; lng: number }) => { setMeasurePoints((prev) => (prev.length >= 3 ? prev : [...prev, pt])); @@ -790,19 +809,8 @@ export default function Dashboard() { regionDossierLoading={regionDossierLoading} gtDossier={gtDossier} gtDossierLoading={gtDossierLoading} - onExpandEntityGraph={() => { - if (isEntityGraphEligible(selectedEntity)) setShowEntityGraph(true); - }} - onArticleClick={(idx, lat, lng, title) => { - if (lat !== undefined && lng !== undefined) { - setFlyToLocation({ lat, lng, ts: Date.now() }); - // Also highlight the corresponding map alert - if (title) { - const alertKey = `${title}|${lat},${lng}`; - setSelectedEntity({ id: alertKey, type: 'news' }); - } - } - }} + onExpandEntityGraph={handleExpandEntityGraph} + onArticleClick={handleArticleClick} /> diff --git a/frontend/src/components/AIIntelPanel.tsx b/frontend/src/components/AIIntelPanel.tsx index b1dc3df..903d6fa 100644 --- a/frontend/src/components/AIIntelPanel.tsx +++ b/frontend/src/components/AIIntelPanel.tsx @@ -3,7 +3,7 @@ import React, { useState, useEffect, useCallback } from 'react'; import ReactDOM from 'react-dom'; import { getBackendEndpoint } from '@/lib/backendEndpoint'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Brain, MapPin, diff --git a/frontend/src/components/AdvancedFilterModal.tsx b/frontend/src/components/AdvancedFilterModal.tsx index 4328f9e..ab20265 100644 --- a/frontend/src/components/AdvancedFilterModal.tsx +++ b/frontend/src/components/AdvancedFilterModal.tsx @@ -1,7 +1,7 @@ 'use client'; import { useState, useMemo, useRef, useCallback, useEffect } from 'react'; -import { motion } from 'framer-motion'; +import { motion } from '@/lib/motion'; import { Search, X, Check, GripHorizontal } from 'lucide-react'; interface FilterField { diff --git a/frontend/src/components/AlertToast.tsx b/frontend/src/components/AlertToast.tsx index 0c51245..d8437ff 100644 --- a/frontend/src/components/AlertToast.tsx +++ b/frontend/src/components/AlertToast.tsx @@ -1,7 +1,7 @@ 'use client'; import { useEffect, useState } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import type { ToastItem } from '@/hooks/useAlertToasts'; const TOAST_LIFETIME_MS = 5_000; diff --git a/frontend/src/components/ChangelogModal.tsx b/frontend/src/components/ChangelogModal.tsx index fa22111..c1eda38 100644 --- a/frontend/src/components/ChangelogModal.tsx +++ b/frontend/src/components/ChangelogModal.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useEffect } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { X, Network, diff --git a/frontend/src/components/FilterPanel.tsx b/frontend/src/components/FilterPanel.tsx index a855936..5b37acc 100644 --- a/frontend/src/components/FilterPanel.tsx +++ b/frontend/src/components/FilterPanel.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useMemo } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Minus, Plus, diff --git a/frontend/src/components/FindLocateBar.tsx b/frontend/src/components/FindLocateBar.tsx index 04374e6..deaebb1 100644 --- a/frontend/src/components/FindLocateBar.tsx +++ b/frontend/src/components/FindLocateBar.tsx @@ -2,7 +2,7 @@ import React, { useState, useMemo, useRef, useEffect } from 'react'; import { Search, Crosshair, Plane, Shield, Star, Ship, X, Database } from 'lucide-react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { trackedOperators } from '../lib/trackedData'; import { useDataKeys } from '@/hooks/useDataStore'; import { useTranslation } from '@/i18n'; diff --git a/frontend/src/components/GlobalTicker.tsx b/frontend/src/components/GlobalTicker.tsx index 64033fb..a25399e 100644 --- a/frontend/src/components/GlobalTicker.tsx +++ b/frontend/src/components/GlobalTicker.tsx @@ -1,7 +1,7 @@ 'use client'; import { useState } from 'react'; -import { motion } from 'framer-motion'; +import { motion } from '@/lib/motion'; import { ArrowUpRight, ArrowDownRight, TrendingUp, AlertTriangle, ChevronUp } from 'lucide-react'; import { useDataKeys } from '@/hooks/useDataStore'; diff --git a/frontend/src/components/InfonetTerminal/index.tsx b/frontend/src/components/InfonetTerminal/index.tsx index bea325a..9dabe98 100644 --- a/frontend/src/components/InfonetTerminal/index.tsx +++ b/frontend/src/components/InfonetTerminal/index.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useEffect } from 'react'; -import { AnimatePresence, motion } from 'framer-motion'; +import { AnimatePresence, motion } from '@/lib/motion'; import { X } from 'lucide-react'; import { beginInfonetTerminalSession } from '@/lib/infonetTerminalSession'; import InfonetShell from './InfonetShell'; diff --git a/frontend/src/components/KeyboardShortcutsOverlay.tsx b/frontend/src/components/KeyboardShortcutsOverlay.tsx index db27bbf..5e35aea 100644 --- a/frontend/src/components/KeyboardShortcutsOverlay.tsx +++ b/frontend/src/components/KeyboardShortcutsOverlay.tsx @@ -1,6 +1,6 @@ 'use client'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; const shortcuts = [ { key: 'L', desc: 'Toggle left panel (LAYERS)' }, diff --git a/frontend/src/components/MapLegend.tsx b/frontend/src/components/MapLegend.tsx index e10c66c..306b059 100644 --- a/frontend/src/components/MapLegend.tsx +++ b/frontend/src/components/MapLegend.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { X, ChevronDown, ChevronUp } from 'lucide-react'; import ExternalImage from '@/components/ExternalImage'; import { useTranslation } from '@/i18n'; diff --git a/frontend/src/components/MaplibreViewer.tsx b/frontend/src/components/MaplibreViewer.tsx index c4da9a9..0db53fc 100644 --- a/frontend/src/components/MaplibreViewer.tsx +++ b/frontend/src/components/MaplibreViewer.tsx @@ -150,6 +150,7 @@ import { EMPTY_FC } from '@/components/map/mapConstants'; import { useImperativeSource } from '@/components/map/hooks/useImperativeSource'; import { useDynamicMapLayersWorker } from '@/components/map/hooks/useDynamicMapLayersWorker'; import { useStaticMapLayersWorker } from '@/components/map/hooks/useStaticMapLayersWorker'; +import { applyDynamicLayerInterp } from '@/components/map/applyDynamicLayerInterp'; import { ClusterCountLabels, TrackedFlightLabels, @@ -303,7 +304,6 @@ function flightPayloadHasKnownRoute(entity: ReturnType) => { const coreData = useDataKeys([ 'tracked_flights', + 'commercial_flights', + 'military_flights', + 'private_flights', + 'private_jets', 'news', 'ships', 'uavs', @@ -1231,7 +1232,9 @@ const MaplibreViewer = ({ { bounds: mapBounds, serverBboxScoped: getLiveDataBounds() !== null, - dtSeconds: dtSeconds.current, + // Worker stamps base positions; main-thread applyDynamicLayerInterp + // dead-reckons between polls without rebuilding FeatureCollections. + dtSeconds: 0, trackedIcaos: Array.from(trackedIcaoSet), activeLayers: { flights: activeLayers.flights, @@ -1251,7 +1254,6 @@ const MaplibreViewer = ({ }, [ mapBounds, - interpTick, trackedIcaoSet, activeLayers.flights, activeLayers.private, @@ -1269,6 +1271,25 @@ const MaplibreViewer = ({ ], ); + const interpolatedDynamicMapLayers = useMemo( + () => { + void interpTick; + return applyDynamicLayerInterp(dynamicMapLayers, dtSeconds.current); + }, + [dynamicMapLayers, interpTick, dtSeconds], + ); + + const { + commercialFlightsGeoJSON: commFlightsGeoJSON, + privateFlightsGeoJSON: privFlightsGeoJSON, + privateJetsGeoJSON: privJetsGeoJSON, + militaryFlightsGeoJSON: milFlightsGeoJSON, + trackedFlightsGeoJSON, + shipsGeoJSON, + meshtasticGeoJSON, + aprsGeoJSON, + } = interpolatedDynamicMapLayers; + const staticMapLayers = useStaticMapLayersWorker( { cctv: staticCctv, @@ -1373,17 +1394,6 @@ const MaplibreViewer = ({ ], ); - const { - commercialFlightsGeoJSON: commFlightsGeoJSON, - privateFlightsGeoJSON: privFlightsGeoJSON, - privateJetsGeoJSON: privJetsGeoJSON, - militaryFlightsGeoJSON: milFlightsGeoJSON, - trackedFlightsGeoJSON, - shipsGeoJSON, - meshtasticGeoJSON, - aprsGeoJSON, - } = dynamicMapLayers; - const { cctvGeoJSON, kiwisdrGeoJSON, @@ -1805,6 +1815,24 @@ const MaplibreViewer = ({ useImperativeSource(mapForHook, 'trains', trainsGeoJSON, 60); useImperativeSource(mapForHook, 'sar-aois', sarAoisGeoJSON, 120); useImperativeSource(mapForHook, 'sar-anomalies', sarAnomaliesGeoJSON, 120); + // Remaining reactive sources → imperative (avoids React reconciling FeatureCollections) + useImperativeSource(mapForHook, 'night-overlay', nightGeoJSON, 250); + useImperativeSource(mapForHook, 'frontlines', frontlineGeoJSON, 200); + useImperativeSource(mapForHook, 'earthquakes', earthquakesGeoJSON, 100); + useImperativeSource(mapForHook, 'gps-jamming', jammingGeoJSON, 150); + useImperativeSource(mapForHook, 'gt-risk-source', gtRiskGeoJSON, 150); + useImperativeSource(mapForHook, 'correlations', correlationsGeoJSON, 100); + useImperativeSource(mapForHook, 'shodan-overlay', shodanGeoJSON, 100); + useImperativeSource(mapForHook, 'ai-intel-source', aiIntelGeoJSON, 80); + useImperativeSource(mapForHook, 'ukraine-alerts-source', ukraineAlertsGeoJSON, 120); + useImperativeSource(mapForHook, 'ukraine-alert-labels-source', ukraineAlertLabelsGeoJSON, 120); + useImperativeSource(mapForHook, 'weather-alerts-source', weatherAlertsGeoJSON, 120); + useImperativeSource(mapForHook, 'weather-alert-labels-source', weatherAlertLabelsGeoJSON, 120); + useImperativeSource(mapForHook, 'carriers', carriersGeoJSON, 75); + useImperativeSource(mapForHook, 'active-route', activeRouteGeoJSON, 50); + useImperativeSource(mapForHook, 'flight-trail', trailGeoJSON, 40); + useImperativeSource(mapForHook, 'predictive-path', predictiveGeoJSON, 40); + useImperativeSource(mapForHook, 'proximity-rings', proximityRingsGeoJSON, 60); const handleMouseMove = useCallback( (evt: MapLayerMouseEvent) => { @@ -2133,8 +2161,8 @@ const MaplibreViewer = ({ {/* SOLAR TERMINATOR — night overlay */} - {activeLayers.day_night && nightGeoJSON && ( - + {activeLayers.day_night && ( + + {/* GPS Jamming Zones — red translucent grid squares */} - + {/* Strategic Risk Heatmap — Bayesian posterior scores */} - + {/* Correlation Alerts — Emergent Intelligence grid squares */} - + {/* RF Anomaly — grey */} + {/* AI Intel Layer — pins from OpenClaw / AI co-pilot (data via useImperativeSource) */} + - )} {/* Military Bases — per-country colors */} @@ -3384,7 +3410,7 @@ const MaplibreViewer = ({ {/* Ukraine Air Raid Alerts — red/orange oblast polygons */} - + - + {/* Weather Alerts — severity-colored polygons with icon + label overlay */} - + - + - + - + {/* Flight trail history (where the aircraft has been) — altitude-colored gradient */} - + {/* Predictive vector (where entity is heading — 5 min forward projection) */} - + {/* Proximity range rings (10nm, 50nm, 100nm around selected entity) */} - + {children}; +} diff --git a/frontend/src/components/NewsFeed.tsx b/frontend/src/components/NewsFeed.tsx index c6c2257..7074ff7 100644 --- a/frontend/src/components/NewsFeed.tsx +++ b/frontend/src/components/NewsFeed.tsx @@ -1,7 +1,7 @@ "use client"; import { useState, useMemo } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { AlertTriangle, Clock, Minus, Plus, ExternalLink, Brain, Loader2, TrendingUp } from 'lucide-react'; import ConfirmDialog from '@/components/ui/ConfirmDialog'; import { usePredictionMarketsOptIn } from '@/hooks/usePredictionMarketsOptIn'; diff --git a/frontend/src/components/OnboardingModal.tsx b/frontend/src/components/OnboardingModal.tsx index 67e6b09..8bcd185 100644 --- a/frontend/src/components/OnboardingModal.tsx +++ b/frontend/src/components/OnboardingModal.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useEffect } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { X, ExternalLink, Key, Shield, Radar, Globe, Satellite, Ship, Radio, Bot, Copy, Check, Network } from 'lucide-react'; const CURRENT_ONBOARDING_VERSION = '0.9.81-agentic-onboarding-1'; diff --git a/frontend/src/components/PredictionsPanel.tsx b/frontend/src/components/PredictionsPanel.tsx index 44f042d..29d47a8 100644 --- a/frontend/src/components/PredictionsPanel.tsx +++ b/frontend/src/components/PredictionsPanel.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useEffect, useCallback, useRef } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Minus, Plus, diff --git a/frontend/src/components/RadioInterceptPanel.tsx b/frontend/src/components/RadioInterceptPanel.tsx index 8fd020c..bfee21a 100644 --- a/frontend/src/components/RadioInterceptPanel.tsx +++ b/frontend/src/components/RadioInterceptPanel.tsx @@ -2,7 +2,7 @@ import { API_BASE } from '@/lib/api'; import { useState, useEffect, useRef, useCallback } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { RadioReceiver, Activity, diff --git a/frontend/src/components/SarAoiEditorModal.tsx b/frontend/src/components/SarAoiEditorModal.tsx index 1bce45f..9a68f01 100644 --- a/frontend/src/components/SarAoiEditorModal.tsx +++ b/frontend/src/components/SarAoiEditorModal.tsx @@ -2,7 +2,7 @@ import React, { useState, useEffect, useCallback } from 'react'; import { createPortal } from 'react-dom'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { X, Radar, Plus, Trash2, MapPin, Crosshair } from 'lucide-react'; import { API_BASE } from '@/lib/api'; import type { SarAoi } from '@/types/dashboard'; diff --git a/frontend/src/components/SarModeChooserModal.tsx b/frontend/src/components/SarModeChooserModal.tsx index 1da4287..56b73ba 100644 --- a/frontend/src/components/SarModeChooserModal.tsx +++ b/frontend/src/components/SarModeChooserModal.tsx @@ -2,7 +2,7 @@ import React, { useState, useEffect } from 'react'; import { createPortal } from 'react-dom'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { X, ExternalLink, Radar, Check, Zap, Globe } from 'lucide-react'; import { API_BASE } from '@/lib/api'; diff --git a/frontend/src/components/SettingsPanel.tsx b/frontend/src/components/SettingsPanel.tsx index c43eaad..32f2282 100644 --- a/frontend/src/components/SettingsPanel.tsx +++ b/frontend/src/components/SettingsPanel.tsx @@ -59,7 +59,7 @@ import { type CompanionStatus, } from '@/lib/desktopCompanion'; import React, { useState, useEffect, useCallback } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Settings, ExternalLink, diff --git a/frontend/src/components/StartupWarmupModal.tsx b/frontend/src/components/StartupWarmupModal.tsx index b02eaa8..66e1fa9 100644 --- a/frontend/src/components/StartupWarmupModal.tsx +++ b/frontend/src/components/StartupWarmupModal.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useEffect, useState } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Database, Clock, X } from 'lucide-react'; const CURRENT_VERSION = '0.9.83'; diff --git a/frontend/src/components/WatchlistWidget.tsx b/frontend/src/components/WatchlistWidget.tsx index eeee733..af842bd 100644 --- a/frontend/src/components/WatchlistWidget.tsx +++ b/frontend/src/components/WatchlistWidget.tsx @@ -1,7 +1,7 @@ 'use client'; import { useState } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import type { WatchlistEntry } from '@/hooks/useWatchlist'; import { Eye, X, Trash2, ChevronUp, ChevronDown, Crosshair } from 'lucide-react'; diff --git a/frontend/src/components/WorldviewLeftPanel.tsx b/frontend/src/components/WorldviewLeftPanel.tsx index 0f38f9e..3a71dad 100644 --- a/frontend/src/components/WorldviewLeftPanel.tsx +++ b/frontend/src/components/WorldviewLeftPanel.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useEffect, useRef, useMemo, useCallback } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Layers, Minus, diff --git a/frontend/src/components/WorldviewRightPanel.tsx b/frontend/src/components/WorldviewRightPanel.tsx index d233edd..0303afb 100644 --- a/frontend/src/components/WorldviewRightPanel.tsx +++ b/frontend/src/components/WorldviewRightPanel.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useState, useEffect } from 'react'; -import { motion, AnimatePresence } from 'framer-motion'; +import { motion, AnimatePresence } from '@/lib/motion'; import { Plus, Minus } from 'lucide-react'; import type { MapEffects } from '@/types/dashboard'; diff --git a/frontend/src/components/map/FishingDestinationRoute.tsx b/frontend/src/components/map/FishingDestinationRoute.tsx index 12a8feb..c7d7c91 100644 --- a/frontend/src/components/map/FishingDestinationRoute.tsx +++ b/frontend/src/components/map/FishingDestinationRoute.tsx @@ -1,7 +1,7 @@ 'use client'; import React, { useEffect, useState, useRef } from 'react'; -import { Source, Layer, Marker } from 'react-map-gl/maplibre'; +import { Source, Layer } from 'react-map-gl/maplibre'; import { API_BASE } from '@/lib/api'; interface Props { @@ -20,7 +20,10 @@ export default function FishingDestinationRoute({ vesselLat, vesselLng, destinat const prevDest = useRef(''); useEffect(() => { - if (!destination) { setDestCoords(null); return; } + if (!destination) { + setDestCoords(null); + return; + } const query = destination.trim(); if (!query || query === prevDest.current) return; prevDest.current = query; @@ -43,7 +46,9 @@ export default function FishingDestinationRoute({ vesselLat, vesselLng, destinat setDestCoords(null); } })(); - return () => { cancelled = true; }; + return () => { + cancelled = true; + }; }, [destination]); if (!destCoords) return null; @@ -56,7 +61,10 @@ export default function FishingDestinationRoute({ vesselLat, vesselLng, destinat properties: { type: 'fishing-route' }, geometry: { type: 'LineString', - coordinates: [[vesselLng, vesselLat], destCoords], + coordinates: [ + [vesselLng, vesselLat], + destCoords, + ], }, }, { @@ -71,49 +79,47 @@ export default function FishingDestinationRoute({ vesselLat, vesselLng, destinat }; return ( - <> - - - - - - + + + + + ); } diff --git a/frontend/src/components/map/applyDynamicLayerInterp.ts b/frontend/src/components/map/applyDynamicLayerInterp.ts new file mode 100644 index 0000000..49c25c8 --- /dev/null +++ b/frontend/src/components/map/applyDynamicLayerInterp.ts @@ -0,0 +1,95 @@ +import { interpolatePosition } from '@/utils/positioning'; +import type { DynamicMapLayersResult } from '@/components/map/dynamicMapLayers.worker'; + +const UNBOUNDED_INTERP_SECONDS = Number.POSITIVE_INFINITY; + +type InterpProps = { + baseLat?: number; + baseLng?: number; + spd?: number | null; + hdg?: number | null; + alt?: number | null; + kind?: 'flight' | 'ship'; +}; + +function interpCoords(props: InterpProps, dtSeconds: number): [number, number] | null { + const lat = props.baseLat; + const lng = props.baseLng; + if (typeof lat !== 'number' || typeof lng !== 'number') return null; + const spd = props.spd; + if (!spd || spd <= 0 || dtSeconds <= 0) return [lng, lat]; + + if (props.kind === 'flight') { + if (props.alt != null && props.alt <= 100) return [lng, lat]; + if (dtSeconds < 1) return [lng, lat]; + } + + const heading = props.hdg || 0; + const [newLat, newLng] = interpolatePosition( + lat, + lng, + heading, + spd, + dtSeconds, + 0, + UNBOUNDED_INTERP_SECONDS, + ); + return [newLng, newLat]; +} + +function applyInterpToCollection( + fc: GeoJSON.FeatureCollection | null, + dtSeconds: number, +): GeoJSON.FeatureCollection | null { + if (!fc?.features?.length || dtSeconds <= 0) return fc; + + let changed = false; + const features = fc.features.map((feature) => { + const props = (feature.properties || {}) as InterpProps; + if (props.baseLat == null || props.baseLng == null) return feature; + if (!props.spd || props.spd <= 0) return feature; + + const coords = interpCoords(props, dtSeconds); + if (!coords) return feature; + + const geom = feature.geometry; + if (!geom || geom.type !== 'Point') return feature; + const [oldLng, oldLat] = geom.coordinates; + if (oldLng === coords[0] && oldLat === coords[1]) return feature; + + changed = true; + return { + ...feature, + geometry: { + type: 'Point' as const, + coordinates: coords, + }, + }; + }); + + if (!changed) return fc; + return { type: 'FeatureCollection', features }; +} + +/** + * Re-apply dead-reckoning between backend polls without rebuilding layer + * FeatureCollections in the dynamic map worker. + * + * Expects features stamped with baseLat/baseLng/spd/hdg(/alt/kind) by the worker. + */ +export function applyDynamicLayerInterp( + layers: DynamicMapLayersResult, + dtSeconds: number, +): DynamicMapLayersResult { + if (dtSeconds <= 0) return layers; + return { + commercialFlightsGeoJSON: applyInterpToCollection(layers.commercialFlightsGeoJSON, dtSeconds), + privateFlightsGeoJSON: applyInterpToCollection(layers.privateFlightsGeoJSON, dtSeconds), + privateJetsGeoJSON: applyInterpToCollection(layers.privateJetsGeoJSON, dtSeconds), + militaryFlightsGeoJSON: applyInterpToCollection(layers.militaryFlightsGeoJSON, dtSeconds), + trackedFlightsGeoJSON: applyInterpToCollection(layers.trackedFlightsGeoJSON, dtSeconds), + shipsGeoJSON: applyInterpToCollection(layers.shipsGeoJSON, dtSeconds), + meshtasticGeoJSON: layers.meshtasticGeoJSON, + aprsGeoJSON: layers.aprsGeoJSON, + }; +} diff --git a/frontend/src/components/map/dynamicMapLayers.worker.ts b/frontend/src/components/map/dynamicMapLayers.worker.ts index 4fc178d..1455d30 100644 --- a/frontend/src/components/map/dynamicMapLayers.worker.ts +++ b/frontend/src/components/map/dynamicMapLayers.worker.ts @@ -1,6 +1,5 @@ /// -import { interpolatePosition } from '@/utils/positioning'; import { classifyAircraft } from '@/utils/aircraftClassification'; import type { Flight, Ship, SigintSignal } from '@/types/dashboard'; import type { FlightLayerConfig } from '@/components/map/geoJSONBuilders'; @@ -99,8 +98,6 @@ const EMPTY_RESULT: DynamicMapLayersResult = { aprsGeoJSON: null, }; -const UNBOUNDED_INTERP_SECONDS = Number.POSITIVE_INFINITY; - const TRACKED_GROUNDED_ICON_MAP: Record = { airliner: 'svgAirlinerGrey', turboprop: 'svgTurbopropGrey', @@ -213,38 +210,6 @@ function flightDisplayLabel(f: Flight): string { return ''; } -function interpFlightPosition(f: Flight, dtSeconds: number): [number, number] { - if (!f.speed_knots || f.speed_knots <= 0 || dtSeconds <= 0) return [f.lng, f.lat]; - if (f.alt != null && f.alt <= 100) return [f.lng, f.lat]; - if (dtSeconds < 1) return [f.lng, f.lat]; - const heading = f.true_track || f.heading || 0; - const [newLat, newLng] = interpolatePosition( - f.lat, - f.lng, - heading, - f.speed_knots, - dtSeconds, - 0, - UNBOUNDED_INTERP_SECONDS, - ); - return [newLng, newLat]; -} - -function interpShipPosition(s: Ship, dtSeconds: number): [number, number] { - if (typeof s.sog !== 'number' || !s.sog || s.sog <= 0 || dtSeconds <= 0) return [s.lng, s.lat]; - const heading = (typeof s.cog === 'number' ? s.cog : 0) || s.heading || 0; - const [newLat, newLng] = interpolatePosition( - s.lat, - s.lng, - heading, - s.sog, - dtSeconds, - 0, - UNBOUNDED_INTERP_SECONDS, - ); - return [newLng, newLat]; -} - function buildFlightLayerGeoJSONWorker( flights: Flight[] | undefined, config: FlightLayerConfig, @@ -257,11 +222,14 @@ function buildFlightLayerGeoJSONWorker( const { colorMap, groundedMap, typeLabel, idPrefix, milSpecialMap, useTrackHeading } = config; const features: GeoJSON.Feature[] = []; + // Geometry is stamped at the reported position (dt=0). Main-thread + // applyDynamicLayerInterp dead-reckons between polls using baseLat/spd/hdg + // so we do not rebuild this FeatureCollection every interp tick. + void dtSeconds; for (let i = 0; i < flights.length; i += 1) { const f = flights[i]; if (f.lat == null || f.lng == null) continue; - const [iLng, iLat] = interpFlightPosition(f, dtSeconds); - if (!passesViewFilter(iLat, iLng, bounds, serverBboxScoped)) continue; + if (!passesViewFilter(f.lat, f.lng, bounds, serverBboxScoped)) continue; if (f.icao24 && trackedIcaos.has(f.icao24.toLowerCase())) continue; const acType = classifyAircraft(f.model, f.aircraft_category); @@ -289,8 +257,14 @@ function buildFlightLayerGeoJSONWorker( callsign: flightDisplayLabel(f), rotation, iconId, + kind: 'flight', + baseLat: f.lat, + baseLng: f.lng, + spd: f.speed_knots ?? null, + hdg: rotation, + alt: f.alt ?? null, }, - geometry: { type: 'Point', coordinates: [iLng, iLat] }, + geometry: { type: 'Point', coordinates: [f.lng, f.lat] }, }); } @@ -306,11 +280,11 @@ function buildTrackedFlightsGeoJSONWorker( if (!flights?.length) return null; const features: GeoJSON.Feature[] = []; + void dtSeconds; for (let i = 0; i < flights.length; i += 1) { const f = flights[i]; if (f.lat == null || f.lng == null) continue; - const [lng, lat] = interpFlightPosition(f, dtSeconds); - if (!passesViewFilter(lat, lng, bounds, serverBboxScoped)) continue; + if (!passesViewFilter(f.lat, f.lng, bounds, serverBboxScoped)) continue; const alertColor = ('alert_color' in f ? f.alert_color : '') || 'white'; const acType = classifyAircraft(f.model, f.aircraft_category); @@ -326,6 +300,7 @@ function buildTrackedFlightsGeoJSONWorker( TRACKED_ICON_MAP.airliner[alertColor] || 'svgAirlinerWhite'; const displayName = flightDisplayLabel(f); + const rotation = f.heading || 0; features.push({ type: 'Feature', @@ -333,10 +308,16 @@ function buildTrackedFlightsGeoJSONWorker( id: f.icao24 || i, type: 'tracked_flight', callsign: String(displayName), - rotation: f.heading || 0, + rotation, iconId, + kind: 'flight', + baseLat: f.lat, + baseLng: f.lng, + spd: f.speed_knots ?? null, + hdg: rotation, + alt: f.alt ?? null, }, - geometry: { type: 'Point', coordinates: [lng, lat] }, + geometry: { type: 'Point', coordinates: [f.lng, f.lat] }, }); } @@ -363,12 +344,12 @@ function buildShipsGeoJSONWorker( return null; } + void dtSeconds; const features: GeoJSON.Feature[] = []; for (let i = 0; i < ships.length; i += 1) { const s = ships[i]; if (s.lat == null || s.lng == null) continue; - const [iLng, iLat] = interpShipPosition(s, dtSeconds); - if (!passesViewFilter(iLat, iLng, bounds, serverBboxScoped)) continue; + if (!passesViewFilter(s.lat, s.lng, bounds, serverBboxScoped)) continue; if (s.type === 'carrier') continue; const isTrackedYacht = Boolean(s.yacht_alert); @@ -389,16 +370,22 @@ function buildShipsGeoJSONWorker( else if (s.type === 'yacht' || isPassenger) iconId = 'svgShipWhite'; else if (isMilitary) iconId = 'svgShipAmber'; + const rotation = s.heading || (typeof s.cog === 'number' ? s.cog : 0) || 0; features.push({ type: 'Feature', properties: { id: s.mmsi || s.name || `ship-${i}`, type: 'ship', name: s.name, - rotation: s.heading || 0, + rotation, iconId, + kind: 'ship', + baseLat: s.lat, + baseLng: s.lng, + spd: typeof s.sog === 'number' ? s.sog : null, + hdg: rotation, }, - geometry: { type: 'Point', coordinates: [iLng, iLat] }, + geometry: { type: 'Point', coordinates: [s.lng, s.lat] }, }); } diff --git a/frontend/src/components/map/hooks/useInterpolation.ts b/frontend/src/components/map/hooks/useInterpolation.ts index b2c0a57..a1117bd 100644 --- a/frontend/src/components/map/hooks/useInterpolation.ts +++ b/frontend/src/components/map/hooks/useInterpolation.ts @@ -12,8 +12,9 @@ const UNBOUNDED_INTERP_SECONDS = Number.POSITIVE_INFINITY; * to smoothly animate entity positions between API updates. * * The interp functions read dtSeconds from a ref so their references stay stable. - * This prevents 7 GeoJSON useMemos from re-firing every tick — GeoJSON only rebuilds - * when source data actually changes (new API fetch), not on every interpolation tick. + * Dynamic flight/ship GeoJSON is rebuilt by the worker only when source data / + * filters / bounds change; applyDynamicLayerInterp then dead-reckons coordinates + * on each interpTick without another full worker rebuild. */ export function useInterpolation() { const dataTimestamp = useRef(Date.now()); @@ -21,7 +22,7 @@ export function useInterpolation() { const [interpTick, setInterpTick] = useState(0); // Update dtSeconds on each tick and bump a lightweight counter so moving - // layers actually rebuild between backend refreshes. + // markers can advance between backend refreshes. useEffect(() => { const iv = setInterval(() => { dtRef.current = (Date.now() - dataTimestamp.current) / 1000; diff --git a/frontend/src/hooks/useDataPolling.ts b/frontend/src/hooks/useDataPolling.ts index caf10be..d911921 100644 --- a/frontend/src/hooks/useDataPolling.ts +++ b/frontend/src/hooks/useDataPolling.ts @@ -1,15 +1,86 @@ import { useEffect, useRef } from "react"; import { API_BASE } from "@/lib/api"; -import { mergeData, setBackendStatus as setStoreBackendStatus } from "./useDataStore"; +import { + applyLayerDeltas, + mergeData, + setBackendStatus as setStoreBackendStatus, +} from "./useDataStore"; import { appendLiveDataBoundsParams, liveDataBoundsKey } from "@/lib/liveDataViewport"; import { VIEWPORT_COMMITTED_EVENT } from "@/components/map/hooks/useViewportBounds"; export type BackendStatus = 'connecting' | 'connected' | 'disconnected'; +const DELTA_LAYER_KEYS = [ + "ships", + "commercial_flights", + "military_flights", + "tracked_flights", + "private_flights", + "private_jets", + "cctv", + "uavs", + "liveuamap", + "gps_jamming", + "satellites", + "sigint", + "trains", +] as const; + +function formatLayerVersions(lv: Record | null): string | null { + if (!lv) return null; + const parts: string[] = []; + for (const key of DELTA_LAYER_KEYS) { + const ver = lv[key]; + if (typeof ver === "number" && Number.isFinite(ver)) { + parts.push(`${key}:${ver}`); + } + } + // Need at least the primary delta layers before requesting deltas. + if (!parts.some((p) => p.startsWith("ships:") || p.startsWith("commercial_flights:"))) { + return null; + } + return parts.join(","); +} + +function ingestFastPayload( + json: Record, + layerVersionsRef: { current: Record | null }, +): boolean { + const mode = String(json.mode || "snapshot"); + if (json.layer_versions && typeof json.layer_versions === "object") { + layerVersionsRef.current = { + ...(layerVersionsRef.current || {}), + ...(json.layer_versions as Record), + }; + } + + if (mode === "delta") { + const deltas = (json.deltas || {}) as Record< + string, + { upsert?: unknown[]; delete?: string[]; version?: number } + >; + const ok = applyLayerDeltas(deltas); + if (!ok) return false; + const layers = (json.layers || {}) as Record; + if (layers && Object.keys(layers).length > 0) { + mergeData(layers); + } + if (json.freshness) mergeData({ freshness: json.freshness }); + if (json.cctv_total != null) mergeData({ cctv_total: json.cctv_total }); + if (json.sigint_totals) mergeData({ sigint_totals: json.sigint_totals }); + return true; + } + + mergeData(json); + return true; +} + // --------------------------------------------------------------------------- // Polling pause/resume — used by Time Machine snapshot playback // --------------------------------------------------------------------------- let _pollingPaused = false; +/** True while the browser tab is hidden — timers are cleared; refresh on focus. */ +let _tabHidden = false; let _fastEtagRef: { current: string | null } | null = null; let _slowEtagRef: { current: string | null } | null = null; @@ -100,7 +171,10 @@ const VIEWPORT_FAST_REFETCH_MIN_INTERVAL_MS = 2500; * the shared ETag cache exactly like the pre-#288 behaviour. * * Viewport commits trigger a debounced fast-tier refetch so regional pans - * refill aircraft/ships without waiting for the 15s poll cadence. + * refill aircraft/ships without waiting for the 15s poll cadence. That refetch + * clears layer-version delta state — row deltas are not bbox-aware for the + * client's existing arrays, so an empty delta after a pan would leave the + * previous region's aircraft on screen (or none, after inView culls them). * * The AIS stream viewport POST (/api/viewport) is still handled separately * by useViewportBounds to limit upstream AIS ingestion. @@ -113,15 +187,18 @@ export function useDataPolling() { // Expose refs so pausePolling/resumePolling can invalidate ETags _fastEtagRef = fastEtag; _slowEtagRef = slowEtag; + _tabHidden = typeof document !== 'undefined' && document.hidden; let hasData = false; let fetchedStartupFastPayload = false; let fastTimerId: ReturnType | null = null; let slowTimerId: ReturnType | null = null; let viewportDebounceTimer: ReturnType | null = null; + let layerToggleRetryTimer: ReturnType | null = null; const fastAbortRef = { current: null as AbortController | null }; const slowAbortRef = { current: null as AbortController | null }; const fastFetchGenRef = { current: 0 }; + const layerVersionsRef = { current: null as Record | null }; let lastViewportFetchKey: string | null = null; let lastViewportFetchAt = 0; @@ -174,9 +251,13 @@ export function useDataPolling() { const useStartupPayload = !fetchedStartupFastPayload && !fastEtag.current; const headers: Record = {}; if (!useStartupPayload && fastEtag.current) headers['If-None-Match'] = fastEtag.current; - const url = appendLiveDataBoundsParams( + let url = appendLiveDataBoundsParams( `${API_BASE}/api/live-data/fast${useStartupPayload ? '?initial=1' : ''}`, ); + const lvParam = !useStartupPayload ? formatLayerVersions(layerVersionsRef.current) : null; + if (lvParam) { + url += (url.includes('?') ? '&' : '?') + `lv=${encodeURIComponent(lvParam)}`; + } const res = await fetch(url, { headers, signal: controller.signal, @@ -195,8 +276,14 @@ export function useDataPolling() { if (useStartupPayload) fetchedStartupFastPayload = true; const json = await res.json(); if (fetchGen !== fastFetchGenRef.current) return; - mergeData(json); - if (hasMeaningfulFastData(json)) hasData = true; + const applied = ingestFastPayload(json, layerVersionsRef); + if (!applied) { + // Delta apply failed — force a full snapshot next tick. + layerVersionsRef.current = null; + fastEtag.current = null; + } else if (hasMeaningfulFastData(json) || json.mode === 'delta') { + hasData = true; + } } } catch (e) { const aborted = @@ -256,6 +343,8 @@ export function useDataPolling() { // Adaptive polling: retry every 3s during startup, back off to normal cadence once data arrives const scheduleNext = (tier: 'fast' | 'slow', fetchGen?: number) => { + // Pause scheduling while the tab is backgrounded; visibilitychange resumes with a refresh. + if (_tabHidden || document.hidden) return; if (tier === 'fast') { if (fetchGen !== undefined && fetchGen !== fastFetchGenRef.current) return; const delay = hasData ? 15000 : 3000; // 3s startup retry → 15s steady state @@ -267,57 +356,101 @@ export function useDataPolling() { } }; - const queueViewportFastRefetch = () => { - if (_pollingPaused) return; + const clearPollTimers = () => { + if (fastTimerId) { + clearTimeout(fastTimerId); + fastTimerId = null; + } + if (slowTimerId) { + clearTimeout(slowTimerId); + slowTimerId = null; + } + }; - const key = liveDataBoundsKey(); - if (!key) { - lastViewportFetchKey = null; + const onVisibilityChange = () => { + if (document.hidden) { + _tabHidden = true; + clearPollTimers(); return; } - if (key === lastViewportFetchKey) return; + _tabHidden = false; + // Resume with one refresh on focus so the UI doesn't feel stuck. + // Keep ETags — 304 is fine when nothing changed. Still respect Time Machine pause. + if (_pollingPaused) return; + void fetchFastData(); + void fetchSlowData(); + }; + + const fireViewportFastRefetch = () => { + if (_pollingPaused || _tabHidden || document.hidden) return; + + // null bounds → world-scale fetch; still refetch when leaving a region + // so the store is not stuck with the previous bbox-filtered arrays. + const currentFetchKey = liveDataBoundsKey() ?? '__world__'; + if (currentFetchKey === lastViewportFetchKey) return; + + const now = Date.now(); + const waitMs = VIEWPORT_FAST_REFETCH_MIN_INTERVAL_MS - (now - lastViewportFetchAt); + if (waitMs > 0) { + // Do not drop the pan — retry when the rate window opens. + if (viewportDebounceTimer) clearTimeout(viewportDebounceTimer); + viewportDebounceTimer = setTimeout(() => { + viewportDebounceTimer = null; + fireViewportFastRefetch(); + }, waitMs); + return; + } + + lastViewportFetchKey = currentFetchKey; + lastViewportFetchAt = now; + fastEtag.current = null; + // Force a bbox-scoped snapshot. Delta mode only patches rows that changed + // in the store; it cannot replace "Europe flights" with "America flights". + layerVersionsRef.current = null; + void fetchFastData(); + }; + + const queueViewportFastRefetch = () => { + if (_pollingPaused || _tabHidden || document.hidden) return; + + const fetchKey = liveDataBoundsKey() ?? '__world__'; + if (fetchKey === lastViewportFetchKey) return; if (viewportDebounceTimer) clearTimeout(viewportDebounceTimer); viewportDebounceTimer = setTimeout(() => { viewportDebounceTimer = null; - if (_pollingPaused) return; - - const currentKey = liveDataBoundsKey(); - if (!currentKey || currentKey === lastViewportFetchKey) return; - - const now = Date.now(); - if (now - lastViewportFetchAt < VIEWPORT_FAST_REFETCH_MIN_INTERVAL_MS) return; - - lastViewportFetchKey = currentKey; - lastViewportFetchAt = now; - fastEtag.current = null; - void fetchFastData(); + fireViewportFastRefetch(); }, VIEWPORT_FAST_REFETCH_DEBOUNCE_MS); }; - // When a layer toggle fires, refetch live tiers immediately and retry a few - // times so network-heavy on-enable fetches (FIRMS, PSK, …) can finish in the - // background without blocking POST /api/layers on the single API worker. + // When a layer toggle fires, refetch live tiers immediately and one follow-up + // retry so network-heavy on-enable fetches (FIRMS, PSK, …) can land without + // a multi-retry storm (was 1s/2.5s/5s → up to 8 requests). const onLayerToggle = () => { slowEtag.current = null; fastEtag.current = null; + // Force full snapshot after toggle — deltas may miss cold-layer fills. + layerVersionsRef.current = null; if (slowTimerId) clearTimeout(slowTimerId); slowTimerId = null; + if (layerToggleRetryTimer) { + clearTimeout(layerToggleRetryTimer); + layerToggleRetryTimer = null; + } void fetchFastData(); void fetchSlowData(); - const retryDelaysMs = [1000, 2500, 5000]; - for (const delay of retryDelaysMs) { - setTimeout(() => { - if (_pollingPaused) return; - slowEtag.current = null; - fastEtag.current = null; - void fetchSlowData(); - void fetchFastData(); - }, delay); - } + layerToggleRetryTimer = setTimeout(() => { + layerToggleRetryTimer = null; + if (_pollingPaused || _tabHidden || document.hidden) return; + slowEtag.current = null; + fastEtag.current = null; + void fetchSlowData(); + void fetchFastData(); + }, 2500); }; window.addEventListener(LAYER_TOGGLE_EVENT, onLayerToggle); window.addEventListener(VIEWPORT_COMMITTED_EVENT, queueViewportFastRefetch); + document.addEventListener('visibilitychange', onVisibilityChange); void (async () => { await fetchCriticalBootstrap(); @@ -329,9 +462,10 @@ export function useDataPolling() { return () => { window.removeEventListener(LAYER_TOGGLE_EVENT, onLayerToggle); window.removeEventListener(VIEWPORT_COMMITTED_EVENT, queueViewportFastRefetch); - if (fastTimerId) clearTimeout(fastTimerId); - if (slowTimerId) clearTimeout(slowTimerId); + document.removeEventListener('visibilitychange', onVisibilityChange); + clearPollTimers(); if (viewportDebounceTimer) clearTimeout(viewportDebounceTimer); + if (layerToggleRetryTimer) clearTimeout(layerToggleRetryTimer); abortInFlightFastFetch(); if (slowAbortRef.current) slowAbortRef.current.abort(); }; diff --git a/frontend/src/hooks/useDataStore.ts b/frontend/src/hooks/useDataStore.ts index 03a2318..cd767dc 100644 --- a/frontend/src/hooks/useDataStore.ts +++ b/frontend/src/hooks/useDataStore.ts @@ -25,17 +25,97 @@ const store: Record = {}; let backendStatus: BackendStatus = "connecting"; const statusListeners = new Set(); +// ── Content-aware equality (stable merge) ──────────────────────────────── + +/** Cheap O(n) fingerprint for dashboard arrays — id + rounded lat/lng. */ +export function arrayFingerprint(arr: unknown[]): string { + let h = String(arr.length); + for (let i = 0; i < arr.length; i++) { + const item = arr[i] as Record | null | undefined; + if (item == null || typeof item !== "object") { + h += `|${item}`; + continue; + } + const id = + item.icao ?? item.id ?? item.mmsi ?? item.hex ?? item.callsign ?? item.name ?? i; + const lat = item.lat != null ? Math.round(Number(item.lat) * 100) : ""; + const lng = item.lng != null ? Math.round(Number(item.lng) * 100) : ""; + h += `|${id}:${lat}:${lng}`; + } + return h; +} + +/** + * Content-aware equality for mergeData: keep the previous reference when + * values are equivalent enough that subscribers need not re-render. + */ +export function valuesEquivalent(prev: unknown, next: unknown): boolean { + if (prev === next) return true; + if (prev == null || next == null) return prev === next; + if (typeof prev !== typeof next) return false; + if (Array.isArray(prev) && Array.isArray(next)) { + if (prev.length !== next.length) return false; + if (prev.length === 0) return true; + // Geo / track layers: cheap id+position fingerprint. Other arrays + // (news, etc.) fall through as not-equivalent so new refs still notify. + const sample = prev[0]; + const isGeoish = + sample != null && + typeof sample === "object" && + ("lat" in (sample as object) || + "lng" in (sample as object) || + "icao" in (sample as object) || + "icao24" in (sample as object) || + "mmsi" in (sample as object)); + if (!isGeoish) return false; + return arrayFingerprint(prev) === arrayFingerprint(next); + } + if (typeof prev === "object" && typeof next === "object") { + // Shallow: same keys and === values + const pk = Object.keys(prev as object); + const nk = Object.keys(next as object); + if (pk.length !== nk.length) return false; + for (const k of pk) { + if ((prev as Record)[k] !== (next as Record)[k]) { + return false; + } + } + return true; + } + return prev === next; +} + // ── Write API (called from useDataPolling) ─────────────────────────────── /** Merge a partial payload into the store, notifying only affected keys. */ export function mergeData(patch: Record) { const changedKeys: string[] = []; for (const key of Object.keys(patch)) { - const next = patch[key]; - if (store[key] !== next) { - store[key] = next; - changedKeys.push(key); + // Protocol / meta fields from live-data — never materialize as layers. + if ( + key === "mode" || + key === "deltas" || + key === "layers" || + key === "store_version" || + key === "layer_versions" || + key === "payload_scale" || + key === "payload_sampled" || + key === "layer_totals" || + key === "startup_payload" || + key === "bootstrap_payload" || + key === "bootstrap_ready" + ) { + continue; } + const next = patch[key]; + const prev = store[key]; + if (valuesEquivalent(prev, next)) { + // Keep previous reference so subscribers do not fire + store[key] = prev; + continue; + } + store[key] = next; + changedKeys.push(key); } // Notify per-key subscribers for (const key of changedKeys) { @@ -48,6 +128,56 @@ export function mergeData(patch: Record) { } } +function entityIdForLayer(layer: string, item: Record): string { + if (layer === "ships") { + return String(item.mmsi ?? item.id ?? "").trim(); + } + return String(item.icao24 ?? item.icao ?? item.id ?? item.hex ?? "") + .trim() + .toLowerCase(); +} + +/** + * Apply row-level upsert/delete patches from `/api/live-data/fast` delta mode. + * Returns false if a layer patch cannot be applied safely (caller should full-resync). + */ +export function applyLayerDeltas( + deltas: Record, +): boolean { + const changedKeys: string[] = []; + for (const [key, patch] of Object.entries(deltas || {})) { + if (!patch || typeof patch !== "object") return false; + const prev = store[key]; + const base = Array.isArray(prev) ? (prev as Record[]) : []; + const byId = new Map>(); + for (const item of base) { + if (!item || typeof item !== "object") continue; + const id = entityIdForLayer(key, item); + if (id) byId.set(id, item); + } + for (const id of patch.delete || []) { + byId.delete(String(id)); + } + for (const raw of patch.upsert || []) { + if (!raw || typeof raw !== "object") continue; + const item = raw as Record; + const id = entityIdForLayer(key, item); + if (!id) continue; + byId.set(id, item); + } + store[key] = Array.from(byId.values()); + changedKeys.push(key); + } + for (const key of changedKeys) { + const set = keyListeners.get(key); + if (set) for (const fn of set) fn(); + } + if (changedKeys.length > 0) { + for (const fn of globalListeners) fn(); + } + return true; +} + export function setBackendStatus(next: BackendStatus) { if (backendStatus === next) return; backendStatus = next; diff --git a/frontend/src/lib/motion.tsx b/frontend/src/lib/motion.tsx new file mode 100644 index 0000000..aaf3905 --- /dev/null +++ b/frontend/src/lib/motion.tsx @@ -0,0 +1,27 @@ +'use client'; + +/** + * Shared framer-motion entry — LazyMotion + domAnimation keeps the animation + * feature set small instead of pulling the full motion bundle into every panel. + * + * Import `motion` / `AnimatePresence` from here (not `framer-motion`) so the + * app stays on the lightweight `m` component under LazyMotion. + */ +import type { ReactNode } from 'react'; +import { + LazyMotion, + domAnimation, + m, + AnimatePresence, +} from 'framer-motion'; + +export { LazyMotion, domAnimation, AnimatePresence }; +export const motion = m; + +export function MotionProvider({ children }: { children: ReactNode }) { + return ( + + {children} + + ); +} diff --git a/frontend/src/types/dashboard.ts b/frontend/src/types/dashboard.ts index f8d4d65..868f361 100644 --- a/frontend/src/types/dashboard.ts +++ b/frontend/src/types/dashboard.ts @@ -825,6 +825,10 @@ export interface DashboardData { satellite_source?: string; financial_source?: string; cctv_total?: number; + /** World/continental sampling metadata from /api/live-data/fast (P5). */ + payload_scale?: 'world' | 'continental' | 'regional'; + payload_sampled?: boolean; + layer_totals?: Record; satnogs_total?: number; tinygs_total?: number; bootstrap_ready?: boolean;