"""Live map overlays: parsers, TTL cache, and upstream fetchers.

Moving objects (aircraft, vessels, trains) and alerts are vectors. Radar /
GIBS fire tiles are rasters served directly to the browser — this module
only returns tile *templates* and GeoJSON/JSON for vectors.

Third-party APIs that leak keys, lack CORS, or rate-limit by IP are fetched
here (FastAPI), never from Leaflet. Viewport bbox only; never a global ADS-B
or AIS poll.
"""

from __future__ import annotations

import asyncio
import logging
import math
import time
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable

import httpx

from config import OSINT_USER_AGENT

logger = logging.getLogger("osint.live_layers")

MARKER_FIELDS = ("id", "lat", "lon", "heading", "speed", "label", "extra")

ADSB_LOL_BASE = "https://api.adsb.lol"
AMTRAKER_TRAINS = "https://api.amtraker.com/v3/trains"
RAINVIEWER_MAPS = "https://api.rainviewer.com/public/weather-maps.json"
NWS_ALERTS = "https://api.weather.gov/alerts/active"
IEM_SBW = "https://mesonet.agron.iastate.edu/geojson/sbw.geojson"
WFIGS_INCIDENTS = (
    "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/"
    "WFIGS_Incident_Locations_Current/FeatureServer/0/query"
)
WFIGS_PERIMETERS = (
    "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/"
    "WFIGS_Interagency_Perimeters_Current/FeatureServer/0/query"
)
NHC_STORMS = "https://www.nhc.noaa.gov/CurrentStorms.json"

IEM_NEXRAD = "https://mesonet.agron.iastate.edu/cache/tile.py/1.0.0/nexrad-n0q/{z}/{x}/{y}.png"
GIBS_THERMAL = (
    "https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/"
    "VIIRS_SNPP_Thermal_Anomalies_375m_All/default/{time}/"
    "GoogleMapsCompatible_Level9/{z}/{y}/{x}.png"
)

CONUS = (-125.0, 24.0, -66.0, 50.0)
MAX_RADIUS_NM = 150
DEFAULT_LIMIT = 2000

_HTTP_TIMEOUT = httpx.Timeout(8.0, connect=3.0)
_http: httpx.AsyncClient | None = None
_COMPASS = {
    "N": 0, "NE": 45, "E": 90, "SE": 135,
    "S": 180, "SW": 225, "W": 270, "NW": 315,
    "NNE": 22, "ENE": 67, "ESE": 112, "SSE": 157,
    "SSW": 202, "WSW": 247, "WNW": 292, "NNW": 337,
}

_cache: dict[str, tuple[float, Any]] = {}
_key_locks: dict[str, asyncio.Lock] = {}
_key_locks_guard = asyncio.Lock()
_QUANT = 0.25  # degrees — pan jitter inside a cell reuses the TTL entry

# Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker.
vessel_last_known: dict[str, dict] = {}
vessel_lock = asyncio.Lock()
# Viewport-following accumulates vessels across every region visited in a
# session — keep the in-memory store bounded (oldest entries evicted).
_MAX_VESSELS = 6000
# Last ADS-B snapshot + WFIGS points for fire↔tanker correlation.
aircraft_last_known: dict[str, dict] = {}
fire_last_known: list[dict] = []


def overlay_catalog() -> dict:
    """Tile templates + vector endpoint map for the layer panel. No secrets."""
    today = datetime.now(timezone.utc).date().isoformat()
    return {
        "radar_iem": {
            "id": "radar_iem",
            "title": "IEM NEXRAD (CONUS)",
            "kind": "raster",
            "tileUrl": IEM_NEXRAD,
            "maxZoom": 18,
            "attribution": "Iowa Environmental Mesonet",
        },
        "radar_rainviewer": {
            "id": "radar_rainviewer",
            "title": "RainViewer (global)",
            "kind": "raster",
            "tileUrl": None,  # filled from /api/map/radar frames
            "maxZoom": 7,
            "attribution": '<a href="https://www.rainviewer.com/">Weather data by RainViewer</a>',
        },
        "gibs_thermal": {
            "id": "gibs_thermal",
            "title": "GIBS VIIRS thermal anomalies",
            "kind": "raster",
            "tileUrl": GIBS_THERMAL.replace("{time}", today),
            "timeTemplate": GIBS_THERMAL,
            "maxZoom": 9,
            "attribution": "NASA GIBS / EOSDIS",
        },
        "nws_alerts": {"id": "nws_alerts", "kind": "geojson", "endpoint": "/api/weather-alerts"},
        "wfigs_incidents": {"id": "wfigs_incidents", "kind": "points", "endpoint": "/api/fire-incidents"},
        "wfigs_perimeters": {"id": "wfigs_perimeters", "kind": "geojson", "endpoint": "/api/fire-perimeters"},
        "aircraft": {"id": "aircraft", "kind": "points", "endpoint": "/api/aircraft"},
        "vessels": {"id": "vessels", "kind": "points", "endpoint": "/api/vessels"},
        "trains": {"id": "trains", "kind": "points", "endpoint": "/api/trains"},
        "storms": {"id": "storms", "kind": "points", "endpoint": "/api/storms"},
    }


def parse_bbox(bbox: str) -> tuple[float, float, float, float]:
    """Parse 'minlon,minlat,maxlon,maxlat' into four floats."""
    parts = [p.strip() for p in (bbox or "").split(",")]
    if len(parts) != 4:
        raise ValueError("bbox must be 'minlon,minlat,maxlon,maxlat'")
    try:
        minlon, minlat, maxlon, maxlat = (float(p) for p in parts)
    except ValueError as exc:
        raise ValueError("bbox values must be floats") from exc
    return minlon, minlat, maxlon, maxlat


def quantize_bbox(
    minlon: float, minlat: float, maxlon: float, maxlat: float,
    step: float = _QUANT,
) -> tuple[float, float, float, float]:
    """Snap a viewport to a coarse cell so nearby pans share a cache key.

    The returned envelope is expanded to cover the original box.
    """
    def q_down(v: float, lo: float, hi: float) -> float:
        v = max(lo, min(hi, v))
        return math.floor(v / step) * step

    return (
        round(q_down(minlon, -180.0, 180.0), 4),
        round(q_down(minlat, -90.0, 90.0), 4),
        round(q_down(maxlon, -180.0, 180.0) + step, 4),
        round(q_down(maxlat, -90.0, 90.0) + step, 4),
    )


def bbox_cell_key(bbox: str | None) -> str:
    """Stable cache-key fragment for a viewport (or 'all')."""
    if not bbox:
        return "all"
    return ",".join(f"{v:.4f}" for v in quantize_bbox(*parse_bbox(bbox)))


def bbox_center_radius_nm(
    minlon: float, minlat: float, maxlon: float, maxlat: float,
) -> tuple[float, float, int]:
    """Viewport center + half-diagonal radius in nautical miles, clamped ≤ 150."""
    lat = (minlat + maxlat) / 2.0
    lon = (minlon + maxlon) / 2.0
    # Half the diagonal of the box, in nm (1 deg lat ≈ 60 nm).
    dlat = abs(maxlat - minlat) / 2.0
    dlon = abs(maxlon - minlon) / 2.0
    km = _haversine_km(lat, lon, lat + dlat, lon + dlon)
    nm = km / 1.852
    radius = max(1, min(MAX_RADIUS_NM, int(math.ceil(nm))))
    return lat, lon, radius


def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    r = 6371.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlmb = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
    return 2 * r * math.asin(min(1.0, math.sqrt(a)))


def to_marker(
    id_: str,
    lat: float | None,
    lon: float | None,
    heading: float | None = None,
    speed: float | None = None,
    label: str | None = None,
    extra: dict | None = None,
) -> dict:
    return {
        "id": str(id_),
        "lat": lat,
        "lon": lon,
        "heading": heading,
        "speed": speed,
        "label": label or str(id_),
        "extra": extra or {},
    }


def filter_points_bbox(
    points: list[dict],
    minlon: float, minlat: float, maxlon: float, maxlat: float,
    limit: int = DEFAULT_LIMIT,
) -> list[dict]:
    out = []
    for p in points:
        lat, lon = p.get("lat"), p.get("lon")
        if lat is None or lon is None:
            continue
        if minlat <= lat <= maxlat and minlon <= lon <= maxlon:
            out.append(p)
            if len(out) >= limit:
                break
    return out


_ALERT_KEEP = ("event", "severity", "headline", "areaDesc", "wfo", "source")


def slim_alert_properties(props: dict | None) -> dict:
    """Keep only the fields the map popup reads."""
    src = props or {}
    return {k: src.get(k) for k in _ALERT_KEEP}


def _walk_coords(coords: Any, acc: list[float]) -> None:
    if not coords:
        return
    first = coords[0]
    if isinstance(first, (int, float)):
        lon, lat = float(coords[0]), float(coords[1])
        acc[0] = min(acc[0], lon)
        acc[1] = min(acc[1], lat)
        acc[2] = max(acc[2], lon)
        acc[3] = max(acc[3], lat)
        return
    for child in coords:
        _walk_coords(child, acc)


def _geom_envelope(geom: dict | None) -> tuple[float, float, float, float] | None:
    if not geom or not isinstance(geom, dict):
        return None
    if geom.get("type") == "GeometryCollection":
        env: list[float] | None = None
        for g in geom.get("geometries") or []:
            e = _geom_envelope(g)
            if e is None:
                continue
            if env is None:
                env = list(e)
            else:
                env[0] = min(env[0], e[0])
                env[1] = min(env[1], e[1])
                env[2] = max(env[2], e[2])
                env[3] = max(env[3], e[3])
        return tuple(env) if env else None  # type: ignore[return-value]
    coords = geom.get("coordinates")
    if coords is None:
        return None
    acc = [180.0, 90.0, -180.0, -90.0]
    try:
        _walk_coords(coords, acc)
    except (TypeError, ValueError, IndexError):
        return None
    if acc[0] > acc[2]:
        return None
    return acc[0], acc[1], acc[2], acc[3]


def clip_fc_to_bbox(
    fc: dict | None,
    minlon: float, minlat: float, maxlon: float, maxlat: float,
) -> dict:
    """Drop features whose geometry envelope misses the viewport. No shapely."""
    box = (minlon, minlat, maxlon, maxlat)
    out = []
    for feat in (fc or {}).get("features") or []:
        if not isinstance(feat, dict):
            continue
        env = _geom_envelope(feat.get("geometry"))
        if env is None:
            continue
        if env[0] <= box[2] and env[2] >= box[0] and env[1] <= box[3] and env[3] >= box[1]:
            out.append(feat)
    return {"type": "FeatureCollection", "features": out}


def _f(value: object) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _heading(value: object) -> float | None:
    if value is None or value == "":
        return None
    if isinstance(value, str):
        key = value.strip().upper()
        if key in _COMPASS:
            return float(_COMPASS[key])
    num = _f(value)
    if num is None:
        return None
    if num < 0 or num > 360:
        return None
    return num


def _s(value: object) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


# ADS-B emitter category (DO-260B). A3 airliners, A5 heavies, A7 helicopters.
_EMITTER = {
    "A0": "unknown", "A1": "light", "A2": "small", "A3": "large",
    "A4": "high vortex", "A5": "heavy", "A6": "high performance", "A7": "rotorcraft",
    "B0": "unknown", "B1": "glider", "B2": "airship", "B3": "parachute",
    "B4": "ultralight", "B6": "UAV", "B7": "space",
    "C0": "ground unknown", "C1": "emergency vehicle", "C2": "service vehicle",
    "D0": "unknown", "D1": "emergency",
}

# Combat / dedicated-military ICAO types. C-130/C-17 omitted — those also fly
# as fire tankers and civil contractors; dbFlags/hex catch the real mil ones.
_MIL_ICAO = frozenset({
    "F15", "F16", "F18", "FA18", "F22", "F35", "F117", "A10", "A10A",
    "B1", "B1B", "B2", "B52", "AV8B", "F4", "F5", "F14",
    "SU27", "SU30", "SU34", "SU35", "SU57",
    "MG29", "MIG29", "MG31", "MIG31", "J10", "J11", "J15", "J16", "J20",
    "EUFI", "RFAL", "TOR", "E3TF", "E3CF", "E6", "E8", "P8",
    "MQ9", "MQ1", "RQ4", "V22", "AH64", "H64",
})

# US DoD Mode-S block AE0000–AEFFFF.
_US_DOD_HEX_LO, _US_DOD_HEX_HI = 0xAE0000, 0xAEFFFF

_MIL_CS_PREFIX = ("RCH", "NAVY", "ARMY", "MARINE", "GOTOF", "REACH")


def classify_adsb(ac: dict) -> tuple[str, str]:
    """Return (role, role_src). Prefer readsb dbFlags bit0, then type/hex/cs."""
    flags = ac.get("dbFlags")
    try:
        flags_i = int(flags) if flags is not None else 0
    except (TypeError, ValueError):
        flags_i = 0
    if flags_i & 1:
        return "military", "dbFlags"
    icao = str(ac.get("t") or "").strip().upper()
    if icao in _MIL_ICAO:
        return "military", "type"
    try:
        hx = int(str(ac.get("hex") or "").strip(), 16)
    except ValueError:
        hx = -1
    if _US_DOD_HEX_LO <= hx <= _US_DOD_HEX_HI:
        return "military", "hex"
    cs = str(ac.get("flight") or "").strip().upper()
    if cs.startswith(_MIL_CS_PREFIX):
        return "military", "callsign"
    return "civilian", "default"


def _adsb_extra(ac: dict, hex_id: str) -> dict[str, Any]:
    role, src = classify_adsb(ac)
    cat = str(ac.get("category") or "").strip().upper()
    extra: dict[str, Any] = {
        "hex": hex_id,
        "reg": _s(ac.get("r")),
        "type": _s(ac.get("t")),
        "alt_baro": ac.get("alt_baro"),
        "squawk": _s(ac.get("squawk")),
        "emergency": _s(ac.get("emergency")),
        "category": cat or None,
        "emitter": _EMITTER.get(cat),
        "seen_pos": ac.get("seen_pos"),
        "role": role,
        "role_src": src,
        "src": "adsb.lol",
    }
    desc = _s(ac.get("desc"))
    if desc:
        extra["desc"] = desc
    own = _s(ac.get("ownOp") or ac.get("ownOpName") or ac.get("ownop"))
    if own:
        extra["ownOp"] = own
    if ac.get("alt_geom") is not None:
        extra["alt_geom"] = ac.get("alt_geom")
    vs = ac.get("baro_rate")
    if vs is None:
        vs = ac.get("geom_rate")
    if vs is not None:
        extra["vs"] = vs
    try:
        raw_flags = ac.get("dbFlags")
        flags_i = int(raw_flags) if raw_flags is not None else 0
    except (TypeError, ValueError):
        flags_i = 0
    if flags_i:
        extra["dbFlags"] = flags_i
        extra["interesting"] = bool(flags_i & 2)
    return extra


_NAVSTAT = {
    0: "underway",
    1: "at anchor",
    2: "not under command",
    3: "restricted manoeuvre",
    4: "constrained by draught",
    5: "moored",
    6: "aground",
    7: "fishing",
    8: "sailing",
    14: "AIS-SART",
    15: "undefined",
}

# Compact MID → country for the flags that actually show up on AIS.
_MID_COUNTRY = {
    211: "Germany", 218: "Germany",
    219: "Denmark", 220: "Denmark",
    224: "Spain", 225: "Spain",
    226: "France", 227: "France", 228: "France",
    232: "United Kingdom", 233: "United Kingdom", 234: "United Kingdom", 235: "United Kingdom",
    236: "Gibraltar", 237: "Greece", 239: "Greece", 240: "Greece", 241: "Greece",
    244: "Netherlands", 245: "Netherlands", 246: "Netherlands",
    247: "Italy", 249: "Malta", 250: "Ireland", 251: "Iceland",
    255: "Portugal", 256: "Malta",
    257: "Norway", 258: "Norway", 259: "Norway",
    261: "Poland", 263: "Portugal", 265: "Sweden", 266: "Sweden",
    271: "Turkey", 273: "Russia", 276: "Estonia", 277: "Lithuania",
    301: "Anguilla", 303: "United States", 310: "Bermuda", 316: "Canada",
    319: "Cayman Islands", 338: "United States", 339: "Jamaica",
    345: "Mexico", 352: "Panama", 353: "Panama", 354: "Panama",
    355: "Panama", 356: "Panama", 357: "Panama",
    366: "United States", 367: "United States", 368: "United States", 369: "United States",
    370: "Panama", 371: "Panama", 372: "Panama", 373: "Panama", 374: "Panama",
    375: "St Vincent", 376: "St Vincent", 377: "St Vincent",
    412: "China", 413: "China", 414: "China", 416: "Taiwan",
    419: "India", 431: "Japan", 432: "Japan", 440: "South Korea", 441: "South Korea",
    477: "Hong Kong", 503: "Australia", 525: "Indonesia", 533: "Malaysia",
    538: "Marshall Islands", 548: "Philippines", 563: "Singapore",
    564: "Singapore", 565: "Singapore", 566: "Singapore", 567: "Thailand",
    574: "Vietnam", 636: "Liberia", 637: "Liberia",
    710: "Brazil", 725: "Chile", 730: "Colombia", 760: "Peru",
}


def _mmsi_country(mmsi: object) -> str | None:
    digits = "".join(ch for ch in str(mmsi or "") if ch.isdigit())
    if len(digits) < 3:
        return None
    try:
        mid = int(digits[:3])
    except ValueError:
        return None
    return _MID_COUNTRY.get(mid)


def classify_ais_type(type_code: int | None) -> tuple[str, str]:
    """Return (role, kind) from ITU-R M.1371 ship-and-cargo type."""
    if type_code is None:
        return "civilian", "unknown"
    t = int(type_code)
    tens = t // 10
    if t == 35:
        return "military", "military"
    if t == 30:
        return "civilian", "fishing"
    if t in (31, 32, 52):
        return "civilian", "tug"
    if t == 33:
        return "civilian", "dredger"
    if t == 34:
        return "civilian", "diving"
    if t == 36:
        return "civilian", "sailing"
    if t == 37:
        return "civilian", "pleasure"
    if t == 50:
        return "government", "pilot"
    if t == 51:
        return "government", "SAR"
    if t == 55:
        return "government", "law"
    if t == 54:
        return "government", "anti-pollution"
    if t == 58:
        return "government", "medical"
    if tens == 4:
        return "civilian", "HSC"
    if tens == 6:
        return "civilian", "passenger"
    if tens == 7:
        return "civilian", "cargo"
    if tens == 8:
        return "civilian", "tanker"
    if tens in (5, 9) or t in (53, 56, 57, 59):
        return "civilian", "special"
    return "civilian", "other"


def transform_adsb_lol(payload: dict | list | None) -> list[dict]:
    """Map ADSB.lol v2 aircraft list to shared markers. Dedup on hex."""
    if payload is None:
        return []
    if isinstance(payload, list):
        aircraft = payload
    else:
        aircraft = payload.get("ac") or payload.get("aircraft") or []
    seen: set[str] = set()
    out: list[dict] = []
    for ac in aircraft:
        hex_id = str(ac.get("hex") or "").strip().lower()
        lat, lon = _f(ac.get("lat")), _f(ac.get("lon"))
        if not hex_id or lat is None or lon is None:
            continue
        if hex_id in seen:
            continue
        seen.add(hex_id)
        flight = str(ac.get("flight") or "").strip() or hex_id
        out.append(to_marker(
            hex_id, lat, lon,
            heading=_heading(ac.get("track")),
            speed=_f(ac.get("gs")),
            label=flight,
            extra=_adsb_extra(ac, hex_id),
        ))
    return out


def transform_amtraker(payload: dict | None) -> list[dict]:
    """Flatten Amtraker `{trainNum: [Train, ...]}` to one marker per trainID."""
    if not payload or not isinstance(payload, dict):
        return []
    out: list[dict] = []
    for _num, trains in payload.items():
        if not isinstance(trains, list):
            continue
        for tr in trains:
            if not isinstance(tr, dict):
                continue
            lat, lon = _f(tr.get("lat")), _f(tr.get("lon"))
            if lat is None or lon is None:
                continue
            tid = str(tr.get("trainID") or tr.get("trainId") or "").strip()
            tnum = str(tr.get("trainNum") or _num)
            route = str(tr.get("routeName") or "").strip()
            label = f"{route} #{tnum}".strip() if route else f"Train {tnum}"
            late = tr.get("late")
            if late is None:
                late = tr.get("lateMin")
            out.append(to_marker(
                tid or tnum, lat, lon,
                heading=_heading(tr.get("heading")),
                speed=_f(tr.get("velocity") or tr.get("speed")),
                label=label,
                extra={
                    "route": route,
                    "trainNum": tnum,
                    "late_min": late,
                    "iconColor": tr.get("iconColor"),
                    "stations": tr.get("stations") or [],
                    "src": "amtraker",
                },
            ))
    return out


def transform_ais_frame(frame: dict | None) -> dict | None:
    """Decode one AISStream JSON envelope to a marker (or None if unusable)."""
    if not frame or not isinstance(frame, dict):
        return None
    meta = frame.get("MetaData") or {}
    mmsi = meta.get("MMSI") or meta.get("mmsi")
    if mmsi is None:
        return None
    name = str(meta.get("ShipName") or meta.get("shipName") or "").strip()
    lat = _f(meta.get("Latitude") if "Latitude" in meta else meta.get("latitude"))
    lon = _f(meta.get("Longitude") if "Longitude" in meta else meta.get("longitude"))
    msg = frame.get("Message") or {}
    pos = (
        msg.get("PositionReport")
        or msg.get("StandardClassBPositionReport")
        or msg.get("ExtendedClassBPositionReport")
        or {}
    )
    extra: dict[str, Any] = {"src": "aisstream", "mmsi": mmsi}
    country = _mmsi_country(mmsi)
    if country:
        extra["country"] = country
    if frame.get("MessageType") == "ShipStaticData":
        static = msg.get("ShipStaticData") or {}
        dest = str(static.get("Destination") or static.get("destination") or "").strip()
        extra["dest"] = dest
        extra["static"] = True
        if not name:
            name = str(static.get("Name") or static.get("name") or "").strip()
        cs = _s(static.get("CallSign") or static.get("callSign"))
        if cs:
            extra["callsign"] = cs
        try:
            imo = int(static.get("ImoNumber") or static.get("imoNumber") or 0)
        except (TypeError, ValueError):
            imo = 0
        if imo:
            extra["imo"] = imo
        type_code = static.get("Type") if "Type" in static else static.get("type")
        try:
            type_i = int(type_code) if type_code is not None else None
        except (TypeError, ValueError):
            type_i = None
        if type_i is not None:
            extra["type_code"] = type_i
            role, kind = classify_ais_type(type_i)
            extra["role"] = role
            extra["kind"] = kind
        dim = static.get("Dimension") or static.get("dimension") or {}
        if isinstance(dim, dict):
            a, b = _f(dim.get("A")), _f(dim.get("B"))
            c, d = _f(dim.get("C")), _f(dim.get("D"))
            if a is not None and b is not None:
                extra["length"] = int(round(a + b))
            if c is not None and d is not None:
                extra["beam"] = int(round(c + d))
        draught = _f(static.get("MaximumStaticDraught") or static.get("maximumStaticDraught"))
        if draught is not None:
            extra["draught"] = draught
        eta = static.get("Eta") or static.get("eta") or {}
        if isinstance(eta, dict) and eta.get("Month"):
            extra["eta"] = (
                f"{int(eta.get('Month') or 0):02d}-{int(eta.get('Day') or 0):02d} "
                f"{int(eta.get('Hour') or 0):02d}:{int(eta.get('Minute') or 0):02d}"
            )
        if lat is None or lon is None:
            # Static-only update: caller merges onto last-known by MMSI.
            return to_marker(str(mmsi), None, None, label=name or str(mmsi), extra=extra)
    heading = pos.get("TrueHeading")
    if heading in (511, 511.0, None):
        heading = pos.get("Cog")
    sog = pos.get("Sog")
    navstat = pos.get("NavigationalStatus")
    extra["navstat"] = navstat
    try:
        extra["nav"] = _NAVSTAT.get(int(navstat)) if navstat is not None else None
    except (TypeError, ValueError):
        extra["nav"] = None
    extra["cog"] = pos.get("Cog")
    if not extra.get("dest"):
        extra.pop("dest", None)
    if lat is None or lon is None:
        return None
    return to_marker(
        str(mmsi), lat, lon,
        heading=_heading(heading),
        speed=_f(sog),
        label=name or str(mmsi),
        extra=extra,
    )


def transform_wfigs_incidents(fc: dict | None) -> list[dict]:
    if not fc:
        return []
    out: list[dict] = []
    for feat in fc.get("features") or []:
        geom = feat.get("geometry") or {}
        coords = geom.get("coordinates") or []
        if len(coords) < 2:
            continue
        lon, lat = _f(coords[0]), _f(coords[1])
        if lat is None or lon is None:
            continue
        props = feat.get("properties") or {}
        name = str(props.get("IncidentName") or "Incident").strip()
        out.append(to_marker(
            name, lat, lon,
            label=name,
            extra={
                "acres": props.get("IncidentSize"),
                "contained": props.get("PercentContained"),
                "state": props.get("POOState"),
                "category": props.get("IncidentTypeCategory"),
                "cause": props.get("FireCause"),
                "discovered": props.get("FireDiscoveryDateTime"),
                "src": "wfigs",
            },
        ))
    return out


def transform_nhc_storms(payload: dict | None) -> list[dict]:
    if not payload:
        return []
    storms = payload.get("activeStorms") or []
    class_label = {
        "TD": "Tropical Depression",
        "TS": "Tropical Storm",
        "HU": "Hurricane",
        "STD": "Subtropical Depression",
        "STS": "Subtropical Storm",
        "PTC": "Potential Tropical Cyclone",
        "PC": "Post-Tropical Cyclone",
    }
    out: list[dict] = []
    for s in storms:
        lat = _f(s.get("latitudeNumeric") or s.get("lat"))
        lon = _f(s.get("longitudeNumeric") or s.get("lon"))
        if lat is None or lon is None:
            continue
        sid = str(s.get("id") or s.get("binNumber") or "").strip()
        name = str(s.get("name") or "Storm").strip()
        klass = str(s.get("classification") or "").upper()
        kind = class_label.get(klass, klass or "Storm")
        out.append(to_marker(
            sid or name, lat, lon,
            heading=_heading(s.get("movementDir")),
            speed=_f(s.get("movementSpeed")),
            label=f"{kind} {name}".strip(),
            extra={
                "classification": klass,
                "intensity_kt": s.get("intensity"),
                "src": "nhc",
            },
        ))
    return out


def rainviewer_tile_url(host: str, path: str, size: int = 256, color: int = 2,
                        smooth: int = 1, snow: int = 1) -> str:
    host = host.rstrip("/")
    path = path if path.startswith("/") else f"/{path}"
    return f"{host}{path}/{size}/{{z}}/{{x}}/{{y}}/{color}/{smooth}_{snow}.png"


def _headers() -> dict[str, str]:
    return {"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"}


async def _lock_for(key: str) -> asyncio.Lock:
    async with _key_locks_guard:
        lock = _key_locks.get(key)
        if lock is None:
            lock = asyncio.Lock()
            _key_locks[key] = lock
        return lock


async def _ttl_get(key: str, ttl: float, factory: Callable[[], Awaitable[Any]]) -> Any:
    now = time.monotonic()
    hit = _cache.get(key)
    if hit and now - hit[0] < ttl:
        return hit[1]
    lock = await _lock_for(key)
    async with lock:
        hit = _cache.get(key)
        if hit and time.monotonic() - hit[0] < ttl:
            return hit[1]
        value = await factory()
        _cache[key] = (time.monotonic(), value)
        return value


async def init_http() -> None:
    """Shared outbound client — one TLS pool for all overlay upstreams."""
    global _http
    if _http is None:
        _http = httpx.AsyncClient(
            timeout=_HTTP_TIMEOUT,
            follow_redirects=True,
            headers=_headers(),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )


async def close_http() -> None:
    global _http
    if _http is not None:
        await _http.aclose()
        _http = None


async def _get_json(url: str, params: dict | None = None) -> Any:
    if _http is None:
        async with httpx.AsyncClient(
            timeout=_HTTP_TIMEOUT, follow_redirects=True, headers=_headers(),
        ) as client:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
            return resp.json()
    resp = await _http.get(url, params=params)
    resp.raise_for_status()
    return resp.json()


async def fetch_aircraft(
    bbox: str, limit: int = DEFAULT_LIMIT, persist: bool = False,
) -> list[dict]:
    """Viewport ADS-B.

    GET path (persist=False) serves in-memory last-known and never writes
    tracks/geofences. Background refresh (persist=True) hits ADSB.lol and
    then persist_aircraft_snapshot.
    """
    minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
    if not persist:
        cached = [
            dict(v) for v in aircraft_last_known.values()
            if v.get("lat") is not None and v.get("lon") is not None
        ]
        filtered = filter_points_bbox(cached, minlon, minlat, maxlon, maxlat, limit)
        if filtered:
            return filtered
    qminlon, qminlat, qmaxlon, qmaxlat = quantize_bbox(minlon, minlat, maxlon, maxlat)
    lat, lon, radius = bbox_center_radius_nm(qminlon, qminlat, qmaxlon, qmaxlat)
    cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}"

    async def _load():
        url = f"{ADSB_LOL_BASE}/v2/lat/{lat:.4f}/lon/{lon:.4f}/dist/{radius}"
        return transform_adsb_lol(await _get_json(url))

    rows = await _ttl_get(cache_key, 8.0, _load)
    for m in rows:
        aircraft_last_known[str(m.get("id"))] = m
    if persist:
        await persist_aircraft_snapshot(rows)
    return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)


async def persist_aircraft_snapshot(rows: list[dict]) -> None:
    """Track / geofence / WS / fire correlation — never on the GET path."""
    from ws_manager import manager
    from tracks import record_position
    from geofence import record_and_notify

    for m in rows:
        mlat, mlon = m.get("lat"), m.get("lon")
        if mlat is None or mlon is None:
            continue
        await record_position("aircraft", m)
        if manager.has_clients():
            await manager.publish_point("adsb", m, lat=mlat, lon=mlon)
        await record_and_notify(
            source_kind="adsb", entity_id=str(m.get("id")),
            lat=mlat, lon=mlon, payload=m,
        )
    if fire_last_known:
        from fire_aircraft import correlate_and_notify
        await correlate_and_notify(fire_last_known, rows)


async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]:
    async def _load():
        return transform_amtraker(await _get_json(AMTRAKER_TRAINS))

    rows = await _ttl_get("amtraker:trains", 20.0, _load)
    if bbox:
        minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
        return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
    return rows[:limit]


async def fetch_vessels(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]:
    async with vessel_lock:
        rows = [dict(v) for v in vessel_last_known.values()
                if v.get("lat") is not None and v.get("lon") is not None]
    if bbox:
        minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
        return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)
    return rows[:limit]


async def upsert_vessel(marker: dict) -> None:
    """Merge an AIS marker into last-known by MMSI. Static-only updates names."""
    vid = str(marker.get("id") or "")
    if not vid:
        return
    async with vessel_lock:
        prev = vessel_last_known.get(vid, {})
        extra = {**(prev.get("extra") or {}), **(marker.get("extra") or {})}
        lat = marker.get("lat") if marker.get("lat") is not None else prev.get("lat")
        lon = marker.get("lon") if marker.get("lon") is not None else prev.get("lon")
        label = marker.get("label")
        if not label or label == vid:
            label = prev.get("label") or vid
        stored = {
            **to_marker(
                vid, lat, lon,
                heading=marker.get("heading") if marker.get("heading") is not None else prev.get("heading"),
                speed=marker.get("speed") if marker.get("speed") is not None else prev.get("speed"),
                label=label,
                extra=extra,
            ),
            "seen_at": datetime.now(timezone.utc).isoformat(),
        }
        vessel_last_known[vid] = stored
        if len(vessel_last_known) > _MAX_VESSELS:
            excess = len(vessel_last_known) - int(_MAX_VESSELS * 0.9)
            oldest = sorted(
                vessel_last_known,
                key=lambda k: vessel_last_known[k].get("seen_at", ""),
            )[:excess]
            for k in oldest:
                vessel_last_known.pop(k, None)
    if lat is not None and lon is not None:
        from ws_manager import manager
        from tracks import record_position
        from geofence import record_and_notify
        await manager.publish_point("ais", stored, lat=lat, lon=lon)
        await record_position("vessel", stored)
        await record_and_notify(
            source_kind="ais", entity_id=vid, lat=lat, lon=lon, payload=stored,
        )


def _wfigs_params(bbox: str | None, *, offset_m: float = 250.0) -> dict:
    params = {
        "where": "1=1",
        "outSR": "4326",
        "f": "geojson",
        "resultRecordCount": 500,
        "maxAllowableOffset": offset_m / 111_320.0,  # metres → degrees
        "geometryPrecision": 5,
    }
    if bbox:
        minlon, minlat, maxlon, maxlat = quantize_bbox(*parse_bbox(bbox))
        params["geometry"] = f"{minlon},{minlat},{maxlon},{maxlat}"
        params["geometryType"] = "esriGeometryEnvelope"
        params["inSR"] = "4326"
        params["spatialRel"] = "esriSpatialRelIntersects"
    return params


async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]:
    params = _wfigs_params(bbox)
    params["outFields"] = (
        "IncidentName,IncidentSize,FireDiscoveryDateTime,POOState,"
        "PercentContained,IncidentTypeCategory,FireCause"
    )

    async def _load():
        return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params))

    rows = await _ttl_get(f"wfigs:inc:{bbox_cell_key(bbox)}", 600.0, _load)
    fire_last_known[:] = list(rows)
    if rows and aircraft_last_known:
        from fire_aircraft import correlate_and_notify
        await correlate_and_notify(rows, list(aircraft_last_known.values()))
    return rows[:limit]


async def fetch_fire_perimeters(bbox: str | None) -> dict:
    params = _wfigs_params(bbox)
    params["outFields"] = (
        "poly_IncidentName,poly_GISAcres,attr_IncidentSize,"
        "attr_PercentContained,attr_FireDiscoveryDateTime"
    )

    async def _load():
        return await _get_json(WFIGS_PERIMETERS, params)

    fc = await _ttl_get(f"wfigs:per:{bbox_cell_key(bbox)}", 600.0, _load)
    if not isinstance(fc, dict):
        return {"type": "FeatureCollection", "features": []}
    return fc


async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict:
    """Cached NWS active alerts + IEM storm-based warning polygons."""

    async def _load_iem():
        try:
            data = await _get_json(IEM_SBW)
        except Exception:
            return {"type": "FeatureCollection", "features": []}
        return data if isinstance(data, dict) else {"type": "FeatureCollection", "features": []}

    async def _load():
        nws_params: dict[str, str] = {"status": "actual"}
        clip_box = None
        if area:
            nws_params["area"] = area.upper()
        if bbox:
            clip_box = quantize_bbox(*parse_bbox(bbox))
        # NWS /alerts/active 400s on bbox= — never send it; clip locally.
        nws_fc: dict = {"features": []}
        nws_ok = True
        try:
            nws_fc = await _get_json(NWS_ALERTS, nws_params)
        except Exception as exc:
            logger.warning("NWS alerts fetch failed: %s", exc)
            nws_ok = False
            nws_fc = {"features": []}
        sbw_fc = await _ttl_get("iem:sbw", 45.0, _load_iem)
        features = []
        for feat in nws_fc.get("features") or []:
            if not isinstance(feat, dict):
                continue
            props = dict(feat.get("properties") or {})
            props["source"] = "nws"
            features.append({**feat, "properties": slim_alert_properties(props)})
        for feat in (sbw_fc or {}).get("features") or []:
            if not isinstance(feat, dict):
                continue
            props = dict(feat.get("properties") or {})
            props["source"] = "iem-sbw"
            if "event" not in props:
                props["event"] = props.get("ps") or "Storm-based warning"
            features.append({**feat, "properties": slim_alert_properties(props)})
        merged = {"type": "FeatureCollection", "features": features, "nws_ok": nws_ok}
        if clip_box:
            clipped = clip_fc_to_bbox(merged, *clip_box)
            clipped["nws_ok"] = nws_ok
            merged = clipped
        elif bbox:
            clipped = clip_fc_to_bbox(merged, *parse_bbox(bbox))
            clipped["nws_ok"] = nws_ok
            merged = clipped
        return merged

    key = f"alerts:{area or ''}:{bbox_cell_key(bbox) if bbox else ''}"
    return await _ttl_get(key, 30.0, _load)


async def fetch_radar_meta() -> dict:
    async def _load():
        data = await _get_json(RAINVIEWER_MAPS)
        host = data.get("host") or "https://tilecache.rainviewer.com"
        radar = (data.get("radar") or {})
        past = radar.get("past") or []
        nowcast = radar.get("nowcast") or []
        frames = []
        for fr in past + nowcast:
            path = fr.get("path")
            if not path:
                continue
            frames.append({
                "time": fr.get("time"),
                "path": path,
                "tileUrl": rainviewer_tile_url(host, path),
            })
        latest = frames[-1]["tileUrl"] if frames else None
        return {
            "provider": "rainviewer",
            "host": host,
            "tileUrl": latest,
            "frames": frames,
            "attribution": "Weather data by RainViewer",
            "iemTileUrl": IEM_NEXRAD,
        }

    return await _ttl_get("radar:rv", 300.0, _load)


async def fetch_storms() -> list[dict]:
    async def _load():
        return transform_nhc_storms(await _get_json(NHC_STORMS))

    return await _ttl_get("nhc:storms", 300.0, _load)