"""AISStream WebSocket worker — server-side only.

aisstream.io forbids browser clients. Connect from the FastAPI/ingest
process, upsert last-known positions, and expose them via GET /api/vessels.

Idle (no crash) when AISSTREAM_API_KEY is unset. Reconnect with jittered
backoff and resend the full subscription within 3 seconds of each connect.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import random

from config import AISSTREAM_API_KEY, AISSTREAM_BBOX
from keystore import get_api_key
from live_layers import transform_ais_frame, upsert_vessel

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

WS_URL = "wss://stream.aisstream.io/v0/stream"
FILTER_TYPES = [
    "PositionReport",
    "StandardClassBPositionReport",
    "ExtendedClassBPositionReport",
    "ShipStaticData",
]


def _parse_boxes(raw: str) -> list[list[list[float]]]:
    """Env format: minlat,minlon,maxlat,maxlon[; ...]. AIS wants [[lat,lon],[lat,lon]]."""
    boxes = []
    for chunk in (raw or "").split(";"):
        parts = [p.strip() for p in chunk.split(",") if p.strip()]
        if len(parts) != 4:
            continue
        minlat, minlon, maxlat, maxlon = (float(p) for p in parts)
        boxes.append([[minlat, minlon], [maxlat, maxlon]])
    return boxes or [[[24.0, -125.0], [50.0, -66.0]]]


async def _resolve_key() -> str:
    return (os.getenv("AISSTREAM_API_KEY") or AISSTREAM_API_KEY
            or (await get_api_key("AISSTREAM_API_KEY")) or "").strip()


async def run_ais_worker() -> None:
    """Long-lived reconnect loop. Safe to spawn as an asyncio task."""
    try:
        import websockets
    except ImportError:
        logger.warning("websockets package not installed — AIS worker disabled")
        return

    backoff = 2.0
    while True:
        key = await _resolve_key()
        if not key:
            logger.warning(
                "AISSTREAM_API_KEY not set — AIS ingest idle. "
                "Create a free key at https://aisstream.io/account"
            )
            await asyncio.sleep(60)
            continue
        boxes = _parse_boxes(AISSTREAM_BBOX)
        try:
            async with websockets.connect(
                WS_URL,
                max_size=2 ** 22,
                ping_interval=20,
                ping_timeout=20,
                compression="deflate",
            ) as ws:
                sub = {
                    "APIKey": key,
                    "BoundingBoxes": boxes,
                    "FilterMessageTypes": FILTER_TYPES,
                }
                await ws.send(json.dumps(sub))
                logger.info("AISStream subscribed (%d bbox(es))", len(boxes))
                backoff = 2.0
                async for raw in ws:
                    if isinstance(raw, bytes):
                        raw = raw.decode("utf-8", errors="replace")
                    try:
                        frame = json.loads(raw)
                    except json.JSONDecodeError:
                        continue
                    marker = transform_ais_frame(frame)
                    if marker:
                        await upsert_vessel(marker)
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001
            logger.exception("AISStream disconnected")
        delay = backoff + random.uniform(0, 1.5)
        logger.info("AISStream reconnect in %.1fs", delay)
        await asyncio.sleep(delay)
        backoff = min(60.0, backoff * 1.7)