"""
V3 wiring for `LiveReadThroughAdapter` — the §6 seam #1 (data → engine) reading **V3 natively**.

`LiveReadThroughAdapter` was designed DI-friendly from day one: it takes `fetch_positions` (live
snapshot) + `fetch_static` (meta / course / teams) as injectable callables, with V2 HTTP as the
default. This module is the V3 production wiring — both callables back into the V3 `Store`
(`reads.boat_position_at` for live fixes, `broadcast_feed.build_feed` for static parts) so the
broadcast pixel + interactive paths run end-to-end off V3 with **no V2 dependency**. The
StreamingScorer, LiveFeedServer, and renderer downstream are unchanged.

The boat-ref join key (`{userId}:boat:<b>`) flows through as the engine's `deviceId` — exactly as
`broadcast_feed.resolve_teams` already does — so the engine joins frames ↔ teams identically.
"""
from __future__ import annotations

import sys
import time
from collections.abc import Callable
from pathlib import Path

_HERE = Path(__file__).resolve().parent
_INGEST = _HERE.parent.parent / "ingest"
sys.path.insert(0, str(_HERE))                                  # noqa: E402
sys.path.insert(0, str(_INGEST))                                # noqa: E402

from broadcast_feed import resolve_static  # noqa: E402  (V3 ingest, flat-import)
from live_adapter import LiveReadThroughAdapter  # noqa: E402  (this dir)
from reads import DEFAULT_MAX_AGE_MS, DEFAULT_PRIMARY_MOUNT, boat_position_at  # noqa: E402
from store import Store  # noqa: E402


def v3_fetch_positions(
    store: Store,
    boats: list[str],
    *,
    now_ms: Callable[[], int] | None = None,
    primary_mount: str = DEFAULT_PRIMARY_MOUNT,
    max_age_ms: int = DEFAULT_MAX_AGE_MS,
) -> Callable[[], dict]:
    """A `fetch_positions` that snapshots the V3 store at call time — the V3 analogue of the V2
    `/api/viz/{i}/positions` polling. For each boat, returns its **canonical** position (primary mount
    + failover, §13.1) at `now_ms()` if it has a fresh fix within `max_age_ms`, else omits it. Output
    shape matches what `positions_to_frames` expects: `{boatRef: {lat, lon, speed, course,
    timestampUtcMs}}`. `now_ms` is injectable for tests; default is wall-clock UTC."""
    _now = now_ms or (lambda: int(time.time() * 1000))

    def fetch() -> dict:
        t = _now()
        out: dict[str, dict] = {}
        for boat in boats:
            fix = boat_position_at(store, boat, t, primary_mount=primary_mount, max_age_ms=max_age_ms)
            if fix is None:
                continue
            out[boat] = {
                "lat": fix.latitude, "lon": fix.longitude,
                "speed": fix.speed or 0.0, "course": fix.course or 0.0,
                "timestampUtcMs": fix.t_ms,
            }
        return out
    return fetch


def http_fetch_positions(
    read_base: str,
    instance_id: str,
    *,
    now_ms: Callable[[], int] | None = None,
) -> Callable[[], dict]:
    """A `fetch_positions` that polls the **gcloud read API** `GET /v3/view/{instance}/positions`
    instead of querying the store — Phase 0 of `docs/plans/live-broadcast-feed-in-gcloud.md`. That
    endpoint runs co-located with the DB + a warm `LiveStateCache`, so ONE cached HTTP call returns all
    boats fast, replacing the ~N cross-boundary SQL round-trips per tick that `v3_fetch_positions` does
    from the render PC. Output shape matches `positions_to_frames` (`{boatRef: {lat, lon, speed, course,
    timestampUtcMs}}`). A transient read miss yields `{}` (boats hold last position), same effect as a
    stale-fix omit. Set `$VR_READ_TOKEN` to reach a non-public instance; a public instance needs none."""
    import json
    import os
    import urllib.request
    base = read_base.rstrip("/")
    _now = now_ms or (lambda: int(time.time() * 1000))
    token = os.environ.get("VR_READ_TOKEN")

    def fetch() -> dict:
        url = f"{base}/v3/view/{instance_id}/positions?t={_now()}"
        req = urllib.request.Request(url)
        if token:
            req.add_header("Authorization", f"Bearer {token}")
        try:
            with urllib.request.urlopen(req, timeout=8) as r:      # noqa: S310 - fixed read-API host
                data = json.loads(r.read().decode("utf-8"))
        except Exception:                                          # noqa: BLE001 - transient miss → hold last
            return {}
        return {ref: {"lat": p.get("lat"), "lon": p.get("lon"),
                      "speed": p.get("speed") or 0.0, "course": p.get("course") or 0.0,
                      "timestampUtcMs": p.get("timestampUtcMs")}
                for ref, p in (data or {}).items()
                if isinstance(p, dict) and p.get("lat") is not None}

    return fetch


def v3_fetch_static(
    store: Store,
    instance_id: str,
    boats: list[str],
    *,
    meta: dict,
    course: dict,
    team_config: dict | None = None,
    course_id: str | None = None,
) -> Callable[[], tuple[dict, dict, list]]:
    """A `fetch_static` that returns `(meta, course, teams)` from V3 reads via `broadcast_feed.resolve_static`
    — the V3 replacement for the V2 Firestore resolution. The injected `meta` and `course` come from the
    management plane (§15.1); `teams` is derived from `entry_occupancy` at the gun. `course_id` enables
    wind merge into `meta`. Uses `resolve_static` (NOT `build_feed`) so the live path never materialises
    frames across the open-ended live window — that walks every second of a 24 h span and hangs against a
    real DB; live positions come from the separate `fetch_positions` poll, so frames here are unneeded."""
    def fetch() -> tuple[dict, dict, list]:
        feed = resolve_static(
            store, instance_id, boats,
            meta=meta, course=course, team_config=team_config, course_id=course_id,
        )
        return feed["meta"], feed["course"], feed["teams"]
    return fetch


def build_v3_live_adapter(
    store: Store,
    instance_id: str,
    group_id: str,
    boats: list[str],
    *,
    meta: dict,
    course: dict,
    team_config: dict | None = None,
    course_id: str | None = None,
    wind: dict | None = None,
    now_ms: Callable[[], int] | None = None,
    primary_mount: str = DEFAULT_PRIMARY_MOUNT,
    max_age_ms: int = DEFAULT_MAX_AGE_MS,
    read_base: str | None = None,
) -> LiveReadThroughAdapter:
    """Construct a `LiveReadThroughAdapter` wired to V3 reads — the production path replacing V2
    polling. The downstream contract (StreamingScorer ← frames; LiveFeedServer ← deltas) is unchanged.
    Usage:

        adapter = build_v3_live_adapter(store, "flight-1", "g-day1", boats,
                                        meta=meta, course=course, team_config=cfg)
        meta_s, course_s, teams = adapter.resolve_meta_course_teams()
        scorer = StreamingScorer(meta_s, course_s, teams)
        for batch in adapter.poll_frames(interval_s=1.0):
            for sec, devs in batch.items():
                scorer.push_frame(int(sec), devs)
    """
    # Phase 0: `read_base` set → poll the cached gcloud /positions over HTTP (no per-tick DB queries
    # from the render PC); else the original store-backed read. `fetch_static` stays store-backed (one
    # read at startup, not per-tick) until the producer fully moves to gcloud (Phase 1).
    fetch_positions = (
        http_fetch_positions(read_base, instance_id, now_ms=now_ms) if read_base
        else v3_fetch_positions(
            store, boats, now_ms=now_ms, primary_mount=primary_mount, max_age_ms=max_age_ms,
        )
    )
    return LiveReadThroughAdapter(
        instance_id, group_id, wind=wind,
        fetch_positions=fetch_positions,
        fetch_static=v3_fetch_static(
            store, instance_id, boats, meta=meta, course=course,
            team_config=team_config, course_id=course_id,
        ),
    )