"""VesselAPI REST poller — quota-capped AIS for the Middle East (free tier 150 calls/mo).

VesselAPI and AISStream are two independent, first-class vessel providers —
not a primary/fallback pair. AISStream (WebSocket) owns live US-coast AIS;
VesselAPI (REST) covers the Strait of Hormuz (default box) where AISStream
has no coverage. Missing one key never disables the other. This worker polls
the REST ``GET /v1/location/vessels/bounding-box`` endpoint at most
``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day
and upserts the results into the shared ``vessel_last_known`` store.

Idle (no crash) when VESSELAPI_API_KEY is unset. Never called from the GET
/api/vessels path — map pans must not hit upstream. One request per poll,
``pagination.limit=50``, never follow ``nextToken``, never send ``filter.sat``.
"""

from __future__ import annotations

import asyncio
import calendar
import json
import logging
import os
from datetime import date, datetime, timezone

import httpx
from sqlalchemy import Column, Date, DateTime, Integer, Table, func, select, text

from config import (
    OSINT_USER_AGENT,
    VESSELAPI_API_KEY,
    VESSELAPI_BBOX,
    VESSELAPI_INTERVAL,
    VESSELAPI_MAX_CALLS_PER_DAY,
)
from database import async_session, engine, metadata
from live_layers import parse_bbox, to_marker, upsert_vessel, vessel_last_known, vessel_lock

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

BASE_URL = "https://api.vesselapi.com/v1"
ENDPOINT = f"{BASE_URL}/location/vessels/bounding-box"
MAX_SPAN_DEG = 4.0  # |dLat| + |dLon| — VesselAPI 400s above this.
PAGE_LIMIT = 50  # pagination.limit; never follow nextToken on the free tier.

_client: httpx.AsyncClient | None = None
_client_lock = asyncio.Lock()


# ── Box parsing / span validation ─────────────────────────────────────────

class BboxError(ValueError):
    """A configured VesselAPI box violates the 4° span rule or is malformed."""


def validate_bbox_span(
    minlat: float, minlon: float, maxlat: float, maxlon: float,
) -> None:
    """Reject boxes VesselAPI would 400 on (span > 4°, bad order, bad range)."""
    if not (-90 <= minlat <= 90 and -90 <= maxlat <= 90
            and -180 <= minlon <= 180 and -180 <= maxlon <= 180):
        raise BboxError("coordinates out of range")
    if minlat >= maxlat or minlon >= maxlon:
        raise BboxError("bbox must have min < max on both axes")
    dlat = abs(maxlat - minlat)
    dlon = abs(maxlon - minlon)
    if dlat + dlon > MAX_SPAN_DEG:
        raise BboxError(
            f"span |dLat|+|dLon| = {dlat + dlon:.2f}° exceeds {MAX_SPAN_DEG}° cap"
        )


def parse_boxes(raw: str) -> list[tuple[float, float, float, float]]:
    """Env format: ``minlat,minlon,maxlat,maxlon[; ...]`` (lat/lon order)."""
    out: list[tuple[float, float, float, float]] = []
    for chunk in (raw or "").split(";"):
        parts = [p.strip() for p in chunk.split(",") if p.strip()]
        if len(parts) != 4:
            continue
        try:
            minlat = float(parts[0])
            minlon = float(parts[1])
            maxlat = float(parts[2])
            maxlon = float(parts[3])
        except ValueError:
            continue
        out.append((minlat, minlon, maxlat, maxlon))
    return out


def parse_boxes_validated(raw: str) -> list[tuple[float, float, float, float]]:
    """Parse boxes, log + skip any that violate the span/order/range rules."""
    valid: list[tuple[float, float, float, float]] = []
    for box in parse_boxes(raw):
        try:
            validate_bbox_span(*box)
            valid.append(box)
        except BboxError as exc:
            logger.warning("VesselAPI bbox %r skipped: %s", box, exc)
    return valid


# ── Position → marker transform ───────────────────────────────────────────

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


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


def transform_vesselapi_position(obj: dict | None) -> dict | None:
    """Map one VesselAPI position object to the shared marker contract.

    Returns None for glitch rows, missing MMSI, or missing coordinates.
    """
    if not obj or not isinstance(obj, dict):
        return None
    if obj.get("suspected_glitch") is True:
        return None
    mmsi = obj.get("mmsi")
    if mmsi is None:
        return None
    lat = _f(obj.get("latitude"))
    lon = _f(obj.get("longitude"))
    if lat is None or lon is None:
        return None
    mmsi_s = str(mmsi)
    name = _s(obj.get("vessel_name") or obj.get("name"))
    heading = _f(obj.get("heading"))
    if heading is None:
        heading = _f(obj.get("cog"))
    sog = _f(obj.get("sog"))
    extra: dict = {
        "src": "vesselapi",
        "mmsi": mmsi_s,
        "cog": obj.get("cog"),
        "sog": obj.get("sog"),
        "navstat": obj.get("nav_status"),
    }
    imo = obj.get("imo")
    if imo:
        extra["imo"] = imo
    dest = _s(obj.get("dest") or obj.get("destination"))
    if dest:
        extra["dest"] = dest
    ts = obj.get("timestamp") or obj.get("processed_timestamp")
    if ts:
        extra["timestamp"] = ts
    return to_marker(
        mmsi_s, lat, lon,
        heading=heading,
        speed=sog,
        label=name or mmsi_s,
        extra=extra,
    )


def transform_vesselapi_payload(payload: dict | None) -> list[dict]:
    """Flatten a bounding-box response ``{vessels: [...]}`` to markers."""
    if not payload or not isinstance(payload, dict):
        return []
    rows = payload.get("vessels") or []
    out = []
    for row in rows:
        marker = transform_vesselapi_position(row)
        if marker:
            out.append(marker)
    return out


def utc_day_start(now: datetime) -> datetime:
    """Floor ``now`` to 00:00:00 UTC."""
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    now = now.astimezone(timezone.utc)
    return now.replace(hour=0, minute=0, second=0, microsecond=0)


def pick_poll_at(poll_times: list[datetime], as_of: datetime) -> datetime | None:
    """Latest poll timestamp at or before ``as_of`` (DVR as-of)."""
    if as_of.tzinfo is None:
        as_of = as_of.replace(tzinfo=timezone.utc)
    else:
        as_of = as_of.astimezone(timezone.utc)
    eligible: list[datetime] = []
    for raw in poll_times:
        ts = raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc)
        ts = ts.astimezone(timezone.utc)
        if ts <= as_of:
            eligible.append(ts)
    return max(eligible) if eligible else None


def snapshot_as_of(rows: list[dict], as_of: datetime) -> list[dict]:
    """Keep only rows from the latest poll_at ≤ ``as_of``."""
    chosen = pick_poll_at(
        [r["poll_at"] for r in rows if r.get("poll_at") is not None],
        as_of,
    )
    if chosen is None:
        return []
    out = []
    for row in rows:
        ts = row.get("poll_at")
        if ts is None:
            continue
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=timezone.utc)
        if ts.astimezone(timezone.utc) == chosen:
            out.append(row)
    return out


# ── Durable daily quota (Postgres, survives restarts) ─────────────────────
# Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork.

vesselapi_quota = Table(
    "vesselapi_quota",
    metadata,
    Column("day", Date, primary_key=True),
    Column("calls", Integer, nullable=False, server_default="0"),
    Column("remaining", Integer, nullable=True),
    Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
)

_CREATE_QUOTA_SQL = text(
    """
    CREATE TABLE IF NOT EXISTS vesselapi_quota (
        day        DATE PRIMARY KEY,
        calls      INTEGER NOT NULL DEFAULT 0,
        remaining  INTEGER,
        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
    """
)

_quota_lock = asyncio.Lock()
_quota_ensured = False


async def ensure_quota_table() -> None:
    global _quota_ensured
    if _quota_ensured:
        return
    async with _quota_lock:
        if _quota_ensured:
            return
        async with engine.begin() as conn:
            await conn.execute(_CREATE_QUOTA_SQL)
        _quota_ensured = True


class PgQuotaStore:
    """Postgres-backed daily call counter. Injected for tests."""

    async def calls_today(self, day: date) -> int:
        await ensure_quota_table()
        async with async_session() as session:
            row = (await session.execute(
                select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
            )).scalar()
        return int(row) if row else 0

    async def remaining_today(self, day: date) -> int | None:
        await ensure_quota_table()
        async with async_session() as session:
            row = (await session.execute(
                select(vesselapi_quota.c.remaining).where(vesselapi_quota.c.day == day)
            )).scalar()
        return int(row) if row is not None else None

    async def bump(self, day: date, remaining: int | None) -> int:
        await ensure_quota_table()
        now = datetime.now(timezone.utc)
        async with async_session() as session:
            existing = (await session.execute(
                select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day)
            )).scalar()
            if existing is None:
                await session.execute(
                    vesselapi_quota.insert().values(
                        day=day, calls=1, remaining=remaining, updated_at=now,
                    )
                )
            else:
                await session.execute(
                    vesselapi_quota.update()
                    .where(vesselapi_quota.c.day == day)
                    .values(
                        calls=vesselapi_quota.c.calls + 1,
                        remaining=remaining,
                        updated_at=now,
                    )
                )
            await session.commit()
        return (int(existing) if existing else 0) + 1


# ── Daily VesselAPI snapshots (DVR as-of + survive restarts) ──────────────
# Cleared at the UTC day boundary so the table holds today's 5 polls only.

_CREATE_VESSELS_SQL = text(
    """
    CREATE TABLE IF NOT EXISTS vessels (
        mmsi     TEXT NOT NULL,
        poll_at  TIMESTAMPTZ NOT NULL,
        lat      DOUBLE PRECISION NOT NULL,
        lon      DOUBLE PRECISION NOT NULL,
        heading  DOUBLE PRECISION,
        speed    DOUBLE PRECISION,
        label    TEXT,
        extra    JSONB,
        PRIMARY KEY (mmsi, poll_at)
    )
    """
)
_CREATE_VESSELS_POLL_IDX = text(
    "CREATE INDEX IF NOT EXISTS ix_vessels_poll_at ON vessels (poll_at DESC)"
)
_CREATE_VESSELS_BBOX_IDX = text(
    "CREATE INDEX IF NOT EXISTS ix_vessels_bbox ON vessels (lon, lat)"
)

_vessels_lock = asyncio.Lock()
_vessels_ensured = False


async def ensure_vessels_table() -> None:
    global _vessels_ensured
    if _vessels_ensured:
        return
    async with _vessels_lock:
        if _vessels_ensured:
            return
        async with engine.begin() as conn:
            await conn.execute(_CREATE_VESSELS_SQL)
            await conn.execute(_CREATE_VESSELS_POLL_IDX)
            await conn.execute(_CREATE_VESSELS_BBOX_IDX)
        _vessels_ensured = True


def _marker_from_vessel_row(r) -> dict:
    extra = r.get("extra") or {}
    if isinstance(extra, str):
        try:
            extra = json.loads(extra)
        except (TypeError, ValueError):
            extra = {}
    if not isinstance(extra, dict):
        extra = {}
    extra.setdefault("src", "vesselapi")
    poll_at = r.get("poll_at")
    if poll_at is not None and hasattr(poll_at, "isoformat"):
        extra["poll_at"] = poll_at.isoformat()
    marker = to_marker(
        str(r["id"]), r["lat"], r["lon"],
        heading=r.get("heading"), speed=r.get("speed"),
        label=r.get("label") or str(r["id"]),
        extra=extra,
    )
    marker["seen_at"] = extra.get("poll_at") or datetime.now(timezone.utc).isoformat()
    return marker


async def persist_vessel_snapshot(markers: list[dict], poll_at: datetime) -> None:
    """Write one VesselAPI poll into ``vessels`` (today's snapshots)."""
    await ensure_vessels_table()
    if not markers:
        return
    async with async_session() as session:
        for m in markers:
            vid = str(m.get("id") or "")
            lat, lon = m.get("lat"), m.get("lon")
            if not vid or lat is None or lon is None:
                continue
            extra = dict(m.get("extra") or {})
            extra.setdefault("src", "vesselapi")
            await session.execute(
                text(
                    """
                    INSERT INTO vessels
                        (mmsi, poll_at, lat, lon, heading, speed, label, extra)
                    VALUES
                        (:mmsi, :poll_at, :lat, :lon, :heading, :speed, :label,
                         CAST(:extra AS jsonb))
                    ON CONFLICT (mmsi, poll_at) DO UPDATE SET
                        lat = EXCLUDED.lat,
                        lon = EXCLUDED.lon,
                        heading = EXCLUDED.heading,
                        speed = EXCLUDED.speed,
                        label = EXCLUDED.label,
                        extra = EXCLUDED.extra
                    """
                ),
                {
                    "mmsi": vid,
                    "poll_at": poll_at,
                    "lat": float(lat),
                    "lon": float(lon),
                    "heading": m.get("heading"),
                    "speed": m.get("speed"),
                    "label": m.get("label") or vid,
                    "extra": json.dumps(extra),
                },
            )
        await session.commit()


async def purge_old_vessels(before: datetime | None = None) -> None:
    """Drop snapshots from before the current UTC day (or ``before``)."""
    await ensure_vessels_table()
    cutoff = before or utc_day_start(datetime.now(timezone.utc))
    async with async_session() as session:
        await session.execute(
            text("DELETE FROM vessels WHERE poll_at < :cutoff"),
            {"cutoff": cutoff},
        )
        await session.commit()


async def fetch_vessels_as_of(
    ts: datetime,
    bbox: str | None = None,
    limit: int = 2000,
) -> list[dict]:
    """Latest VesselAPI poll at or before ``ts`` (DVR as-of, not exact minute)."""
    try:
        await ensure_vessels_table()
        async with async_session() as session:
            poll = (await session.execute(
                text("SELECT max(poll_at) FROM vessels WHERE poll_at <= :ts"),
                {"ts": ts},
            )).scalar()
            if poll is None:
                return []
            sql = """
                SELECT mmsi AS id, lat, lon, heading, speed, label, extra, poll_at
                FROM vessels
                WHERE poll_at = :poll
            """
            params: dict = {"poll": poll, "limit": limit}
            if bbox:
                minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
                sql += (
                    " AND lon BETWEEN :minlon AND :maxlon"
                    " AND lat BETWEEN :minlat AND :maxlat"
                )
                params.update(
                    minlon=minlon, minlat=minlat, maxlon=maxlon, maxlat=maxlat,
                )
            sql += " LIMIT :limit"
            rows = (await session.execute(text(sql), params)).mappings().all()
        return [_marker_from_vessel_row(r) for r in rows]
    except Exception:
        logger.exception("VesselAPI snapshot fetch failed")
        return []


async def hydrate_last_known() -> int:
    """Seed in-memory last-known from today's latest poll (app boot)."""
    try:
        rows = await fetch_vessels_as_of(datetime.now(timezone.utc))
    except Exception:
        logger.exception("VesselAPI hydrate failed")
        return 0
    if not rows:
        return 0
    async with vessel_lock:
        for m in rows:
            vid = str(m.get("id") or "")
            if vid:
                vessel_last_known[vid] = m
    return len(rows)


# ── Budget / scheduling (pure, unit-testable) ─────────────────────────────

def days_left_in_month(now: datetime) -> int:
    """UTC days remaining in the current month, inclusive of today."""
    _, last = calendar.monthrange(now.year, now.month)
    return last - now.day + 1


def budget_allows(
    calls_today: int,
    remaining: int | None,
    days_left: int,
    max_per_day: int,
) -> bool:
    """True if another poll is permitted today.

    Local hard cap: fewer than ``max_per_day`` successful calls today.
    Monthly floor: if ``X-RateLimit-Remaining`` is known, keep at least
    ``max_per_day * days_left`` in reserve for the rest of the month.
    """
    if calls_today >= max_per_day:
        return False
    if remaining is not None and remaining <= max_per_day * days_left:
        return False
    return True


def choose_box(
    boxes: list[tuple[float, float, float, float]],
    calls_today: int,
    max_per_day: int,
) -> int:
    """Index into ``boxes`` for the next poll.

    Prefer refreshing the first (primary) box rather than spraying one call
    across every region — round-robin only when the remaining daily budget is
    enough to cover all boxes.
    """
    if len(boxes) <= 1:
        return 0
    budget_left = max_per_day - calls_today
    if budget_left >= len(boxes):
        return calls_today % len(boxes)
    return 0


# ── HTTP / poll ───────────────────────────────────────────────────────────

async def _resolve_key() -> str:
    from keystore import get_api_key
    return (
        os.getenv("VESSELAPI_API_KEY")
        or VESSELAPI_API_KEY
        or (await get_api_key("VESSELAPI_API_KEY"))
        or ""
    ).strip()


async def _get_client() -> httpx.AsyncClient:
    global _client
    if _client is None:
        async with _client_lock:
            if _client is None:
                _client = httpx.AsyncClient(
                    timeout=httpx.Timeout(15.0, connect=5.0),
                    follow_redirects=True,
                    headers={"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"},
                    limits=httpx.Limits(max_connections=1, max_keepalive_connections=1),
                )
    return _client


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


def _int_header(value: str | None) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: str) -> bool:
    """One quota-checked poll. Returns True if a successful 2xx was made.

    Only successful 2xx responses count against the monthly quota; 4xx/5xx/429
    are skipped without retry-storming (Retry-After respected by simply
    sleeping the interval).
    """
    now = datetime.now(timezone.utc)
    today = now.date()
    calls = await store.calls_today(today)
    remaining = await store.remaining_today(today)
    days_left = days_left_in_month(now)
    if not budget_allows(calls, remaining, days_left, VESSELAPI_MAX_CALLS_PER_DAY):
        logger.info(
            "VesselAPI quota reached (calls_today=%d, remaining=%s, days_left=%d) — skip poll",
            calls, remaining, days_left,
        )
        return False

    idx = choose_box(boxes, calls, VESSELAPI_MAX_CALLS_PER_DAY)
    minlat, minlon, maxlat, maxlon = boxes[idx]
    client = await _get_client()
    params = {
        "filter.latBottom": str(minlat),
        "filter.latTop": str(maxlat),
        "filter.lonLeft": str(minlon),
        "filter.lonRight": str(maxlon),
        "pagination.limit": str(PAGE_LIMIT),
    }
    headers = {"Authorization": f"Bearer {key}"}
    try:
        resp = await client.get(ENDPOINT, params=params, headers=headers)
    except httpx.HTTPError as exc:
        logger.warning("VesselAPI request failed: %s", exc)
        return False

    if resp.status_code == 429:
        logger.warning(
            "VesselAPI rate-limited (Retry-After=%s) — skip poll",
            resp.headers.get("Retry-After"),
        )
        return False
    if resp.status_code >= 400:
        logger.warning("VesselAPI HTTP %d — not counted against quota", resp.status_code)
        return False

    # 2xx success — counts against the monthly quota.
    remaining = _int_header(resp.headers.get("X-RateLimit-Remaining"))
    calls = await store.bump(today, remaining)
    try:
        data = resp.json()
    except ValueError:
        logger.warning("VesselAPI 2xx with non-JSON body — counted but ignored")
        return True
    markers = transform_vesselapi_payload(data)
    for m in markers:
        await upsert_vessel(m)
    try:
        await persist_vessel_snapshot(markers, now)
        await purge_old_vessels(utc_day_start(now))
    except Exception:  # noqa: BLE001 — live overlay must not die on persist
        logger.exception("VesselAPI snapshot persist failed")
    logger.info(
        "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)",
        len(markers), remaining, calls,
    )
    return True


# ── Worker loop ───────────────────────────────────────────────────────────

async def run_vesselapi_worker(store: PgQuotaStore | None = None) -> None:
    """Long-lived poll loop. Idle when the key is unset; never crashes the app."""
    if store is None:
        store = PgQuotaStore()
    boxes = parse_boxes_validated(VESSELAPI_BBOX)
    if not boxes:
        logger.warning(
            "VESSELAPI_BBOX has no valid boxes (span ≤ %.1f°) — poller idle", MAX_SPAN_DEG,
        )
    while True:
        try:
            if not boxes:
                await asyncio.sleep(VESSELAPI_INTERVAL)
                continue
            key = await _resolve_key()
            if not key:
                logger.warning(
                    "VESSELAPI_API_KEY not set — VesselAPI poller idle. "
                    "Create a free key at https://dashboard.vesselapi.com/"
                )
                await asyncio.sleep(VESSELAPI_INTERVAL)
                continue
            await poll_once(store, boxes, key)
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001 — keep the loop alive across transient failures
            logger.exception("VesselAPI poll error")
        await asyncio.sleep(VESSELAPI_INTERVAL)