"""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 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 to_marker, upsert_vessel

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


# ── 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


# ── 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)
    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)