"""
Live **V3** broadcast driver — broadcast the live V3 `Store` to the 3D scene over SSE.

The production counterpart to `serve_live.py` (which replays a recorded feed file): this reads live
nav out of the V3 `Store` via `build_v3_live_adapter` (§6 seam #1 — `v3_adapter.py`), scores it
incrementally through `StreamingScorer`, and publishes `init`/`delta` over `LiveFeedServer` — the same
renderer, the same BroadcastFeed contract, unchanged. It closes the gap that `build_v3_live_adapter`
had no driver wiring it to the live feed.

Two modes:

  * ``--seed`` (default): an in-memory store seeded with a **moving two-boat race** around Stockholm,
    so the entire 3D pipeline (scene + standings + commentary, plus the optional AIS overlay) runs on a
    laptop with **no rig** — the 3D analogue of `v3/ingest/demo_serve.py` (which drives the 2D map).
  * production: pass a Postgres-backed store + the management-plane ``meta``/``course``/``team_config``
    (call ``run(...)`` directly), and live nav from the boats drives the broadcast.

Run:
    .venv-demo/Scripts/python v3/broadcast/live/serve_v3_live.py            # seeded offline demo
    # open http://localhost:8765/core/web/index.html?feed=live
    AISSTREAM_API_KEY=… .venv-demo/Scripts/python v3/broadcast/live/serve_v3_live.py --ais
"""
from __future__ import annotations

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

_HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(_HERE))
sys.path.insert(0, str(_HERE.parent.parent / "ingest"))

from live_feed_server import LiveFeedServer, program_state  # noqa: E402
from race_queue import OnAirQueue, StartedRaces  # noqa: E402
from streaming_score import StreamingScorer  # noqa: E402
from v3_adapter import build_v3_live_adapter  # noqa: E402


def _simple_commentator(meta: dict, teams: list, course: dict):   # noqa: ANN001, ANN201
    """Build the streaming factual (team-racing-free) commentator (`--simple`) for non-team-race formats
    — e.g. the GSYS Aug-2 sprint. Lazily imported so the v3/scoring path is only needed when used."""
    sys.path.insert(0, str(_HERE.parents[1] / "scoring"))
    from simple_commentary import SimpleCommentator
    return SimpleCommentator(meta, teams, course)


def _centroid(positions: dict) -> tuple[float, float] | None:
    """Mean `(lat, lon)` of a tick's boat positions — the point the AIS overlay box follows."""
    pts = [(p["lat"], p["lon"]) for p in positions.values()
           if p.get("lat") is not None and p.get("lon") is not None]
    if not pts:
        return None
    return (sum(la for la, _ in pts) / len(pts), sum(lo for _, lo in pts) / len(pts))


def _own_mmsis(store, instance_id: str, t_ms: int, candidates) -> set:  # noqa: ANN001
    """MMSIs to hide as the tracked boats' own vessels — the instance's boats' `boat_meta` resolved
    against `candidates` (live AIS targets or replay observations) via `ais_feed.resolve_own_mmsis`."""
    from ais_feed import resolve_own_mmsis
    from attribution import resolve_roster
    roster = resolve_roster(
        store.entries_covering_instance(instance_id, t_ms),
        t_ms, instance_id=instance_id)
    boats = sorted({b for team in roster.get(instance_id, {}).values() for b in team})
    metas = [m for b in boats if (m := store.boat_meta_for(b)) is not None]
    return resolve_own_mmsis(metas, candidates)


def media_from_store(store, boats: list[str], track_start_ms: int, *,  # noqa: ANN001 — Store
                     clip_url: Callable) -> dict:
    """The PRODUCTION live boat-cam media map — derived from the store's `video_segments` (§19.10),
    not from any driver-side bookkeeping: `{boatRef: [{tSec, file, video}, …]}` in the exact shape the
    renderer's `mediaAt` consumes. `clip_url(segment) -> str | None` maps a segment's `gcs_path` to a
    URL the renderer can fetch — a v4-signed GCS URL in production (`gcs_clip_url`), a static-root
    relative path in the local demo; `None` skips the clip. (Video only — stills live in the photos
    table and are not part of the live tile timeline.)"""
    media: dict[str, list] = {}
    for boat in boats:
        clips: list[dict] = []
        for slot in store.slots_of_boat(boat):
            for seg in store.video_segments_for(slot):
                url = clip_url(seg)
                if not url:
                    continue
                clips.append({"tSec": max(0, (seg.start_ms - track_start_ms) // 1000),
                              "file": url, "video": True})
        if clips:
            clips.sort(key=lambda c: c["tSec"])
            media[boat] = clips
    return media


def gcs_clip_url(segment) -> str | None:                # noqa: ANN001 — VideoSegment
    """Production `clip_url`: a short-lived signed HTTPS URL for a `gs://` clip (the renderer's plain
    <video> can't send auth headers). None for non-GCS paths (a LocalUploader dev path)."""
    from read_server import _signed_clip_url
    try:
        return _signed_clip_url(segment.gcs_path)
    except Exception:                                   # noqa: BLE001 — an unsignable clip is skipped
        return None


def event_id_for_instance(store, instance_id: str) -> str | None:   # noqa: ANN001 — Store
    """The competition this instance belongs to (race_instances.event_id), or None (standalone)."""
    inst = store.get_race_instance(instance_id)
    return getattr(inst, "event_id", None) if inst else None


def apply_event_config(store, event_id: str | None, *, meta: dict) -> dict:   # noqa: ANN001
    """Overlay the EVENT-level configuration (event_config.py) onto derived meta — venue, boat
    class, presentation flags. Called BEFORE `apply_broadcast_config`, so per-instance keys win:
    engine-derived defaults < event config < instance broadcast_config. `event_id` None or no
    config stored → meta unchanged (full backwards compatibility). The config's `integrations`
    (secret refs) are NEVER copied into meta — they are server-side only."""
    if event_id is None:
        return meta
    cfg = store.get_event_config(event_id) or {}
    if not cfg:
        return meta
    out = dict(meta)
    venue = cfg.get("venue") or {}
    if venue:
        v = dict(out.get("venue") or {})
        if venue.get("name"):
            v["name"] = str(venue["name"])
        if venue.get("cityBearingDeg") is not None:
            v["cityBearingDeg"] = float(venue["cityBearingDeg"])
        if venue.get("world"):
            v["world"] = str(venue["world"])
        if venue.get("navMarks"):
            v["navMarks"] = venue["navMarks"]           # real venue aids to navigation (renderer draws them)
        out["venue"] = v
    if cfg.get("boatClass"):
        out["boatClass"] = str(cfg["boatClass"])
    pres = cfg.get("presentation") or {}
    if pres.get("noDownwindSails") is not None:
        out["noDownwindSails"] = bool(pres["noDownwindSails"])
    if pres.get("rcVessels") is not None:
        out["rcVessels"] = bool(pres["rcVessels"])
    if pres.get("hullLivery"):                             # named fleet paint scheme (renderer j80.js;
        out["hullLivery"] = str(pres["hullLivery"])        # 'gsys' = the GSYS hulls TRWC races on)
    if pres.get("mode"):
        out["presentationMode"] = str(pres["mode"])        # "race" | "track" (individual tracks)
    if pres.get("penaltyRegime"):
        out["penaltyRegime"] = str(pres["penaltyRegime"])   # what a penalty IS here (SI-dependent)
    if pres.get("startSequence"):
        # The event's start-sequence LENGTH (SI 11.1 — e.g. a 3-minute team-racing sequence). The
        # driver derives `trackStartUtcMs = gun - preStartS` from it, so a competition that runs a
        # 5-minute sequence must not silently be served the engine's fallback. Only the duration is
        # the event's choice; the RRS tone patterns + offset shape stay engine capabilities
        # (v3/hooter/schedule.start_sequence_offsets).
        out["startSequence"] = dict(pres["startSequence"])
    return out


def apply_broadcast_config(store, instance_id: str, *,   # noqa: ANN001 — Store
                           meta: dict, team_config: dict | None = None) -> tuple[dict, dict | None]:
    """Overlay the management-plane BROADCAST CONFIG (production instance-meta model) onto the driver
    inputs: the operator sets `{title?, gunUtcMs?, teamConfig?}` via
    `PUT /v3/manage/instances/{id}/broadcast`, and the live driver reads it here — no more hand-passed
    launch parameters. Absent keys leave the derived values untouched."""
    cfg = store.get_broadcast_config(instance_id) or {}
    if cfg.get("title"):
        meta = {**meta, "title": str(cfg["title"])}
    if cfg.get("gunUtcMs") is not None:
        meta = {**meta, "gunUtcMs": int(cfg["gunUtcMs"])}
    if cfg.get("noDownwindSails") is not None:              # TRWC: J/80s race without spinnakers
        meta = {**meta, "noDownwindSails": bool(cfg["noDownwindSails"])}
    if cfg.get("rcVessels") is not None:                    # TRWC: lines = RC vessel + robotic mark
        meta = {**meta, "rcVessels": bool(cfg["rcVessels"])}
    if cfg.get("hullLivery"):                               # per-instance override of the event's
        meta = {**meta, "hullLivery": str(cfg["hullLivery"])}   # fleet paint scheme
    if cfg.get("onboardTiles") is not None:                 # tiles show the live cockpit view (sim)
        meta = {**meta, "onboardTiles": bool(cfg["onboardTiles"])}
    if cfg.get("cockpitCrew") is not None:                  # sim demo: crew + periodic cockpit shots
        meta = {**meta, "cockpitCrew": bool(cfg["cockpitCrew"])}
    return meta, (cfg.get("teamConfig") or team_config)


class _OnAirRace:
    """ONE race on air: its adapter, resolved static feed, scorer, commentator and per-tick publish.

    A race is a built object rather than a stretch of loop body so the producer can **move to the next
    race without restarting** (`broadcast_following`): build the next one, publish its `init`, keep the
    same server and the same SSE connection. The renderer swaps in place (`live/web/live_feed.js`)."""

    def __init__(self, store, srv: LiveFeedServer, *, instance_id: str, group_id: str,  # noqa: ANN001
                 boats: list[str], meta: dict, course: dict, team_config: dict | None = None,
                 course_id: str | None = None, read_base: str | None = None,
                 producer: Callable | None = None, simple: bool = False,
                 clip_url: Callable | None = None, ais=None, center: dict | None = None,  # noqa: ANN001
                 join_ms: int | None = None, teams: list | None = None,
                 backfill: Callable | None = None, on_log: Callable[[str], None] = print) -> None:
        self.store, self.srv, self.ais, self.center = store, srv, ais, center
        self.instance_id, self.boats, self.clip_url = instance_id, boats, clip_url
        if teams is not None:
            # STORE-FREE (REST source): the static feed is already resolved — the producer runs against
            # a backend it has no database connection to (local or gcloud, from the render PC or the
            # cloud). Positions come off the read API; everything else was read at the gun.
            from live_adapter import LiveReadThroughAdapter  # noqa: PLC0415
            from v3_adapter import http_fetch_positions  # noqa: PLC0415
            if not read_base:
                raise ValueError("a pre-resolved presentation needs read_base (positions come over REST)")
            self.adapter = LiveReadThroughAdapter(
                instance_id, group_id,
                fetch_positions=http_fetch_positions(read_base, instance_id),
                fetch_static=lambda: (meta, course, teams))
        else:
            self.adapter = build_v3_live_adapter(
                store, instance_id, group_id, boats,
                meta=meta, course=course, team_config=team_config, course_id=course_id,
                read_base=read_base,
            )
        self.meta, self.course, self.teams = self.adapter.resolve_meta_course_teams()
        self.scorer = StreamingScorer(self.meta, self.course, self.teams, producer=producer)
        #: Has the RACE PLANE ended this race? Until it has, `finished()` falls back to the scorer.
        self._server_closed = False
        self.commentator = _simple_commentator(self.meta, self.teams, self.course) if simple else None
        # Committee signals (recall/postpone/abandon/shorten) refreshed EACH TICK, frontier-filtered so a
        # signal the Race Officer fires mid-race appears on the tick it happens (not retroactively) — the
        # live counterpart to regenerate's read-once. Drives scoring (OCS/shorten via enrich) + RIE beats.
        self._signals_for = getattr(store, "race_signals_for", None) if store is not None else None
        self._seen: set[tuple[str, str]] = set()
        self.published_clips = 0
        media = (media_from_store(store, boats, self.meta["trackStartUtcMs"], clip_url=clip_url)
                 if (clip_url is not None and store is not None) else None)
        if media:
            self.published_clips = sum(len(v) for v in media.values())
        self.srv.publish_init(self.meta, self.course, self.teams, {}, media=media)
        # JOINED IN PROGRESS (this race waited its turn behind another): back-fill the seconds it
        # already sailed. `enrich` is a single-pass whole-feed scorer — without the history the boats
        # start at mark 0 mid-course, so the standings, the leg detection and the finish would all be
        # wrong. Scored, not aired (`seed_frames`): commentary for time that has gone never airs.
        # The frames come from the store, or from the injected `backfill` when there is no store
        # (REST source: `rest_source.RestBackend.frames` over `/v3/view/{i}/tracks`).
        if join_ms is not None:
            t0 = self.meta["trackStartUtcMs"]
            elapsed = max(0, (join_ms - t0) // 1000)
            if elapsed > 0:
                if backfill is not None:
                    frames = backfill(instance_id, boats, t0, t0 + elapsed * 1000)
                elif store is not None:
                    from reads import boat_frames  # noqa: PLC0415 — only on a mid-race join
                    frames = boat_frames(store, boats, track_start_ms=t0,
                                         track_end_ms=t0 + elapsed * 1000)
                else:
                    frames = {}
                if frames:
                    self.scorer.seed_frames(frames)
                    self._seen = {(dev, str(s)) for s in range(elapsed + 1) for dev in boats}
                    on_log(f"[live] {instance_id} joined in progress at T+{elapsed}s — back-filled "
                           f"so the standings are whole")
                else:
                    on_log(f"[live] {instance_id} joined in progress at T+{elapsed}s with NO history "
                           f"available — standings start from the join")

    def note_server_closed(self) -> None:
        """The race plane has ended this race — the operator's stop, or an abandonment. WHETHER is
        the question; an abandonment carries no stop time at all, so there is nothing to record but
        the fact."""
        self._server_closed = True

    def finished(self) -> bool:
        """Is the race over?

        THE SERVER OWNS THIS. A race ends when the Race Officer's record says it ended — the
        broadcast reads that, it does not decide it. This used to answer from the presentation's own
        scorer, which meant the commentary engine calling a winner was what made a race "finished":
        a race the server had stopped could still read as running here, and a scorer that stayed
        silent could leave a finished fleet marked ongoing.

        The scorer remains the FALLBACK, deliberately. A broadcast that cannot reach the read plane
        must still be able to end a race rather than sit on a finished fleet forever — losing the
        server's answer costs authority, not the show.
        """
        return self._server_closed or self.scorer.result() is not None

    def tick(self) -> None:
        """One poll: push every new second into the scorer and publish its delta, then refresh the
        boat-cam header if the clip set grew."""
        batch = self.adapter.poll_once(self._seen)
        for sec in sorted(batch, key=int):
            positions = batch[sec]
            if self._signals_for is not None:
                now_ms = self.meta["trackStartUtcMs"] + int(sec) * 1000
                self.scorer.race_signals = [
                    {"signal": s.signal, "tMs": s.t_ms, "meta": s.meta, "source": s.source}
                    for s in self._signals_for(self.instance_id) if (s.t_ms or 0) <= now_ms]
            delta = self.scorer.push_frame(int(sec), positions)
            centroid = _centroid(positions)
            if centroid is not None and self.center is not None:
                self.center["pos"] = centroid           # AIS box + wind poller follow the boat
            ais_list: list = []
            if self.ais is not None and centroid is not None:
                # Own-fleet AIS suppression needs the store's device list; without one (REST source)
                # nothing is excluded — the overlay simply shows every nearby vessel.
                excl = (_own_mmsis(self.store, self.instance_id, int(time.time() * 1000),
                                   self.ais.store.all_targets()) if self.store is not None else set())
                ais_list = self.ais.targets_near(centroid[0], centroid[1], exclude_mmsi=excl)
            # The event's own sequence length and the committee's signal log, so the phase can name
            # the moment ("the preparatory signal has gone") instead of only the clock.
            prog = program_state(gun_utc_ms=self.meta["gunUtcMs"],
                                 track_start_utc_ms=self.meta["trackStartUtcMs"],
                                 sec=int(sec), finished=self.finished(),
                                 pre_start_s=int((self.meta.get("startSequence") or {}).get("preStartS") or 180),
                                 signals=self.meta.get("raceSignals"))
            evts = (self.commentator.tick(int(sec), delta["standings"], positions)
                    if self.commentator is not None else delta["events"])
            self.srv.publish_delta(int(sec), delta["standings"], evts, positions, ais_list,
                                   program=prog, onboard=delta.get("rieOnboardDirectives"))
        if self.clip_url is not None and self.store is not None:
            media = media_from_store(self.store, self.boats, self.meta["trackStartUtcMs"],
                                     clip_url=self.clip_url)
            n = sum(len(v) for v in media.values())
            if n > self.published_clips:                # clip set grew → refresh the feed header
                self.published_clips = n
                self.srv.publish_init(self.meta, self.course, self.teams, {}, media=media)


class _Waiting:
    """The WAITING presentation — what is ON AIR when no race is.

    A broadcast that has nothing to show must show that it has nothing to show: the venue, the boats
    that are out there, and `meta.idle` so the renderer runs its area-panning look-down camera instead
    of a race pose. Two failure modes this exists to prevent, both seen live:

      * a BLANK page while the operator gets ready (nothing published at all), and
      * a presentation that LIES — a placeholder gun makes the scene claim "RACE 11:30" with standings
        and commentary while the fleet is in fact still sitting on the line waiting for the start.

    No scorer runs here: there is no race to score. Positions still stream, so the fleet moves on
    screen and the camera has the venue to pan over."""

    def __init__(self, srv: LiveFeedServer, *, meta: dict, course: dict, teams: list,
                 fetch_positions: Callable[[], dict], title: str | None = None,
                 on_log: Callable[[str], None] = print) -> None:
        self.srv, self.course, self.teams = srv, course, teams
        self._fetch = fetch_positions
        # A sentinel gun a year out keeps the renderer in its pre-start phase ("waiting"), and `idle`
        # is what switches the camera to the area pan. The clock fields stay COMPLETE — a half-filled
        # meta NaNs the whole show clock.
        t0 = meta.get("trackStartUtcMs") or int(time.time() * 1000)
        self.meta = {**meta, "idle": True, "openEnded": True,
                     "trackStartUtcMs": t0, "gunUtcMs": t0 + 365 * 86_400_000,
                     "trackEndUtcMs": t0 + 86_400_000, "frameStepMs": 1000}
        if title:
            self.meta["title"] = title
        self._t0 = t0
        srv.publish_init(self.meta, course, teams, {})
        on_log("[live] WAITING presentation on air - panning the area until a race starts")

    def tick(self) -> None:
        now_ms = int(time.time() * 1000)
        sec = max(0, (now_ms - self._t0) // 1000)
        positions = self._fetch() or {}
        self.srv.publish_delta(int(sec), None, [], positions, [],
                               program=program_state(gun_utc_ms=self.meta["gunUtcMs"],
                                                     track_start_utc_ms=self._t0, sec=int(sec),
                                                     finished=False))


def broadcast_following(
    store,                                              # noqa: ANN001 — v3 ingest Store
    srv: LiveFeedServer,
    *,
    races: StartedRaces,
    presentation_for: Callable[[str, int], tuple],      # (instance, gunMs) -> (meta, course, boats, cfg)
    group_id: str = "g",
    course_id: str | None = None,
    ais=None,                                           # noqa: ANN001 — optional AISFeed
    center: dict | None = None,
    clip_url: Callable | None = None,
    interval_s: float = 1.0,
    max_polls: int | None = None,
    simple: bool = False,
    read_base: str | None = None,
    producer: Callable | None = None,
    queue: OnAirQueue | None = None,
    backfill: Callable | None = None,                   # (instance, boats, t0, t1) -> frames
    waiting_for: Callable[[], dict] | None = None,      # () -> _Waiting kwargs; None → publish nothing
    #: Mutable cell holding the course the live wind feed writes against. Rewritten as each race goes
    #: on air, because a producer following a regatta has no single course for the session.
    on_air_course: dict | None = None,
    now_ms: Callable[[], int] | None = None,
    sleep: Callable[[float], None] = time.sleep,
    on_log: Callable[[str], None] = print,
) -> None:
    """Broadcast the operator's races, one after another, **without ever restarting**.

    `races` reports each race as its start signal fires; `OnAirQueue` decides what is on air: the
    broadcast stays with the race it is showing until that race has its WINNER (plus the hold on the
    finish) or the operator closes it, and only then puts up the next one — joined in progress, with
    its elapsed track back-filled so its standings are whole. One process, one port, one SSE stream,
    for a whole regatta day: what used to need a relaunch per heat.

    `presentation_for(instance, gun_ms)` supplies that race's `(meta, course, boats, team_config)` —
    the store/config resolution, injected so this loop stays independent of where a race is defined.

    This loop only ever READS the race plane. It never writes a gun, a stop or a result: the broadcast
    presents existing data, and the Race Officer's record must not be a function of what the
    presentation happened to score. Whoever owns the boats (the sim) or the committee (the operator)
    decides a race is over; the broadcast finds out by reading, like every other client."""
    q = queue or OnAirQueue()
    clock = now_ms or (lambda: int(time.time() * 1000))
    on_air: _OnAirRace | None = None
    waiting: _Waiting | None = None
    #: What the waiting picture last showed, so a change in the operator's marshalling can be noticed.
    waiting_key: object | None = None

    def _begin_waiting() -> tuple:
        """The venue instead of a blank — but only once there IS something to show. Before the
        operator has allocated anyone there is no fleet and no line-up, so `waiting_for` answers None
        and we simply try again next poll rather than publishing an empty sea.

        Returns `(waiting, key)`; the key identifies the LINE-UP on screen so the caller can rebuild
        when the operator allocates, un-allocates or swaps a hull."""
        if waiting_for is None:
            return None, None
        kwargs = waiting_for()
        if not kwargs:
            return None, None
        key = tuple(sorted((t.get("teamId"), tuple(b.get("deviceId") for b in (t.get("boats") or [])))
                           for t in (kwargs.get("teams") or [])))
        return _Waiting(srv, on_log=on_log, **kwargs), key

    waiting, waiting_key = _begin_waiting()
    polls = 0
    while max_polls is None or polls < max_polls:
        polls += 1
        for instance, gun in races.poll():
            if q.offer(instance, gun):
                on_log(f"[live] {instance} started while {q.on_air} is ON AIR — QUEUED (the broadcast "
                       f"stays with the race on air until it has a winner); waiting: {q.waiting()}")
        if on_air is not None:
            # The RACE PLANE decides a race is over — the operator's stop, or an abandonment. Hand
            # that to the presentation so its `finished()` reads the record rather than its own
            # scorer's opinion (its scorer stays the fallback for a read plane that is unreachable).
            if races.closed(on_air.instance_id):
                on_air.note_server_closed()
            q.note_result(on_air.finished())      # for the hold on the finish — presentation only
        nxt = q.take_next(over=(on_air is not None and races.closed(on_air.instance_id)))
        if nxt is not None:
            instance, gun = nxt
            # `presentation_for` returns (meta, course, boats, team_config) store-backed, or a 5th
            # element — the already-resolved `teams` — when the source is REST (no store at all).
            resolved = presentation_for(instance, gun)
            meta, course, boats, team_config = resolved[:4]
            teams = resolved[4] if len(resolved) > 4 else None
            if on_air_course is not None:
                # `meta.courseId` is the race's OWN course (`_presentation_for` resolves it from the
                # instance); `course_id` is the operator's pin, which wins when given.
                on_air_course["id"] = course_id or meta.get("courseId")
            on_air = _OnAirRace(store, srv, instance_id=instance, group_id=group_id, boats=boats,
                                meta=meta, course=course, team_config=team_config,
                                course_id=course_id, read_base=read_base, producer=producer,
                                simple=simple, clip_url=clip_url, ais=ais, center=center,
                                join_ms=clock(), teams=teams, backfill=backfill, on_log=on_log)
            waiting = None                               # a race is on air now
            on_log(f"[live] ON AIR: {instance} ({races.name_of(instance) or 'race'}) — "
                   f"{len(boats)} boats, gun {gun}")
        elif (on_air is not None and waiting_for is not None
              and q.done(over=races.closed(on_air.instance_id)) and not q.waiting()):
            # The race on air is finished and nothing is queued. Holding its last frame for ever is
            # what "the broadcast has stopped" looks like — go back to the waiting presentation.
            on_air = None
            if on_air_course is not None and not course_id:
                # Nothing on air ⇒ no course to file weather against. Leaving the finished race's
                # course here would keep writing real readings into a race that is over. A PINNED
                # course (`--course`) is the operator's standing choice and stays.
                on_air_course["id"] = None
            q.release()
            waiting, waiting_key = _begin_waiting()
        if on_air is not None:
            on_air.tick()
        elif waiting is not None:
            # MARSHALLING IS LIVE. The operator is allocating hulls to teams while this picture is up,
            # and a still frame of the line-up as it stood when the picture went up is exactly what
            # makes a boat that has just been given a team keep showing no flag. Re-read it, and
            # republish only when it actually changed — an init per poll would restart the renderer.
            fresh = waiting_for() if waiting_for is not None else None
            key = tuple(sorted((t.get("teamId"), tuple(b.get("deviceId") for b in (t.get("boats") or [])))
                               for t in ((fresh or {}).get("teams") or []))) if fresh else None
            if fresh and key != waiting_key:
                waiting, waiting_key = _Waiting(srv, on_log=on_log, **fresh), key
            waiting.tick()
        elif waiting_for is not None:
            waiting, waiting_key = _begin_waiting()      # nothing to show yet — retry each poll
        if max_polls is None or polls < max_polls:
            sleep(interval_s)


def broadcast_live(
    store,                                              # noqa: ANN001 — v3 ingest Store
    srv: LiveFeedServer,
    *,
    instance_id: str,
    group_id: str,
    boats: list[str],
    meta: dict,
    course: dict,
    team_config: dict | None = None,
    course_id: str | None = None,
    ais=None,                                           # noqa: ANN001 — optional AISFeed
    center: dict | None = None,                         # mutable {"pos": (lat,lon)} the AIS + wind feeds follow
    clip_url: Callable | None = None,                   # segment -> URL: enables store-driven boat-cam media
    interval_s: float = 1.0,
    max_polls: int | None = None,
    simple: bool = False,
    read_base: str | None = None,
    producer: Callable | None = None,                   # RIE producer; None → enrich (default scorer)
    sleep: Callable[[float], None] = time.sleep,
) -> None:
    """Resolve the static feed, publish `init`, then stream `delta`s as live fixes land. Each tick the
    boat centroid is written into `center["pos"]` so the AIS box + the wind poller follow the boat.
    With `clip_url`, the boats' uploaded clips are read from the STORE (`media_from_store`) and the
    feed header is re-published whenever the clip set grows — the live boat-cam tile channel, driven
    by the same `video_segments` production writes. `max_polls` + `sleep` are injectable so a test can
    run a bounded, no-wait loop; production leaves them default (poll forever). Pure over the injected
    `store` — the same path a real Postgres run takes."""
    race = _OnAirRace(store, srv, instance_id=instance_id, group_id=group_id, boats=boats,
                      meta=meta, course=course, team_config=team_config, course_id=course_id,
                      read_base=read_base, producer=producer, simple=simple, clip_url=clip_url,
                      ais=ais, center=center)
    polls = 0
    while max_polls is None or polls < max_polls:
        polls += 1
        race.tick()
        if max_polls is None or polls < max_polls:
            sleep(interval_s)


def run(
    store,                                              # noqa: ANN001
    *,
    instance_id: str,
    group_id: str,
    boats: list[str],
    meta: dict,
    course: dict,
    team_config: dict | None = None,
    course_id: str | None = None,
    port: int = 8765,
    ais_key: str | None = None,
    ais_radius_m: float = 500.0,
    wind_provider: str | None = None,
    simple: bool = False,
    read_base: str | None = None,
    producer: Callable | None = None,                   # RIE producer; None → enrich
    follow: tuple | None = None,                        # (StartedRaces, presentation_for) → follow mode
) -> None:
    """Boot the `LiveFeedServer` (serving the static 3D page + the SSE stream) and broadcast the live
    store forever. Wire an `AISFeed` overlay when `ais_key` is given, and a `WindFeed` (real wind →
    live `WindReading`s) when `wind_provider` is given (needs `course_id`). Both follow the boat via
    one shared `center` holder.

    With `follow=(races, presentation_for)` the producer FOLLOWS the operator's races
    (`broadcast_following`) instead of the one pinned `instance_id`: it stays with the race on air
    until that race has a winner, then moves to the next — same process, same port, same stream."""
    # LIVE: small backlog so a reconnecting viewer (e.g. across Cloud Run's 60-min SSE cap) snaps to NOW
    # instead of replaying ~2 h of history. (run_regenerate keeps the full backlog for VOD late-joiners.)
    srv = LiveFeedServer(host="0.0.0.0", port=port, live_backlog=LiveFeedServer.LIVE_BACKLOG)
    srv.start(background=True)
    print(f"V3 live broadcast on :{srv.port}")
    print(f"Open: http://localhost:{srv.port}/core/web/index.html?feed=live")

    center: dict = {"pos": None}
    ais = None
    if ais_key and store is None:
        print("AIS overlay: OFF - the REST source has no store to persist observations to")
        ais_key = None
    if wind_provider and store is None:
        print("Wind feed: OFF - the REST source has no store to write WindReadings to "
              "(the backend's own wind poller supplies the wind instead)")
        wind_provider = None
    if ais_key:
        from ais_feed import AISFeed
        from store import AisObservation
        ais = AISFeed(ais_key, lambda: center["pos"], radius_m=ais_radius_m,
                      persist=lambda tgt: store.add_ais_observation(AisObservation.from_ais(tgt)))
        ais.start()
        print(f"AIS overlay: ON (aisstream.io, {ais_radius_m:.0f} m radius); observations persisted")
    # The course live wind readings attach to. PINNED mode has one for the session; FOLLOW mode does
    # not — each race states its own — so this is a cell the follow loop rewrites as each race goes on
    # air, and the wind feed resolves at every poll. Without it, the documented follow deployment
    # passed `--wind` and silently got no wind: the feed needed an id up front and there was none.
    on_air_course: dict = {"id": course_id}
    wind = None
    if wind_provider and (course_id or follow is not None):
        from wind_feed import WindFeed
        wind = WindFeed(lambda: center["pos"], store.add_wind_reading,
                        course_id=lambda: on_air_course["id"], provider=wind_provider)
        wind.start()
        print(f"Wind feed: ON ({wind_provider} → live WindReadings on course "
              f"{course_id!r})" if course_id else
              f"Wind feed: ON ({wind_provider} → live WindReadings on each race's own course)")
    elif wind_provider:
        print("Wind feed: OFF - needs a course_id to attach WindReadings to")
    try:
        if follow is not None:
            races, presentation_for, backfill, waiting_for = (*follow, None, None)[:4]
            broadcast_following(store, srv, races=races, presentation_for=presentation_for,
                                group_id=group_id, course_id=course_id, ais=ais, center=center,
                                simple=simple, read_base=read_base, producer=producer,
                                backfill=backfill, waiting_for=waiting_for,
                                on_air_course=on_air_course)
        else:
            broadcast_live(store, srv, instance_id=instance_id, group_id=group_id, boats=boats,
                           meta=meta, course=course, team_config=team_config, course_id=course_id,
                           ais=ais, center=center, simple=simple, read_base=read_base,
                           producer=producer)
        print("stream ended — serving for reconnects. Ctrl-C to stop.")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nstopping")
    finally:
        if ais is not None:
            ais.stop()
        if wind is not None:
            wind.stop()
        srv.stop()


def meta_course_from_store(
    store,                                              # noqa: ANN001
    *,
    instance_id: str,
    course_id: str,
    title: str,
    user: str = "GSYS",
    group_id: str = "g",
    gun_ms: int | None = None,
    now_ms: int | None = None,
    track_start_ms: int | None = None,
    track_end_ms: int | None = None,
) -> tuple[dict, dict, list[str]]:
    """Derive the engine `(meta, course)` + the boats list **from the store** — the production path a
    real Postgres-backed trip takes: `course` geometry from `course_roles` via `resolve_course`
    (drifting marks + fallback), boats from the `entry_occupancy` roster. `gun`/`title` are
    management-plane inputs; the rest is derived. Center = start-gate midpoint when known.

    For LIVE, leave the window unset (track spans now → now+24 h). For **regeneration**, pass
    `track_start_ms`/`track_end_ms` = the `[t0, t1]` window to re-render, and `gun_ms`; course + roster
    are resolved at the gun."""
    from attribution import resolve_roster
    from course import resolve_course
    now = now_ms if now_ms is not None else int(time.time() * 1000)
    start = track_start_ms if track_start_ms is not None else now
    end = track_end_ms if track_end_ms is not None else now + 86_400_000
    gun = gun_ms if gun_ms is not None else start
    course = resolve_course(store, course_id, gun, name=title)
    sl = course.get("startLine")
    if sl:
        course["center"] = {"lat": (sl["port"]["lat"] + sl["starboard"]["lat"]) / 2,
                            "lon": (sl["port"]["lon"] + sl["starboard"]["lon"]) / 2}
    roster = resolve_roster(
        store.entries_covering_instance(instance_id, gun),
        gun, instance_id=instance_id)
    boats = sorted({b for team in roster.get(instance_id, {}).values() for b in team})
    _ctr = course.get("center") or {}
    if (not _ctr or (_ctr.get("lat") == 0.0 and _ctr.get("lon") == 0.0)) and boats:
        # Free-drive instance (no start line / marks, e.g. a car test): `resolve_course` defaults the
        # centre to (0, 0), which would anchor the 3D world off the coast of Africa. Centre the scene on
        # the boats' latest fix instead so the Cesium world + camera follow the craft.
        from reads import boat_position_at  # noqa: PLC0415
        for b in boats:
            fix = boat_position_at(store, b, now)
            if fix is not None:
                course["center"] = {"lat": fix.latitude, "lon": fix.longitude}
                break
    live = track_start_ms is None
    meta = {"title": title, "user": user, "instanceId": instance_id, "groupId": group_id,
            "openEnded": live, "raceType": "team_race_2",
            "trackStartUtcMs": start, "gunUtcMs": gun, "trackEndUtcMs": end, "frameStepMs": 1000}
    return meta, course, boats


def regenerate(
    store,                                              # noqa: ANN001
    srv: LiveFeedServer,
    *,
    instance_id: str,
    boats: list[str],
    meta: dict,
    course: dict,
    team_config: dict | None = None,
    course_id: str | None = None,
    ais_radius_m: float = 500.0,
    speed: float = 1.0,
    max_ticks: int | None = None,
    simple: bool = False,
    sleep: Callable[[float], None] = time.sleep,
    producer: Callable[[dict], dict] | None = None,
) -> None:
    """**Regenerate the complete video from the DB** (docs/plans/youtube-live-demo-hr36.md): rebuild
    the whole feed for the `[trackStart, trackEnd]` window in `meta` from the immutable store
    (`build_feed` → frames from `boat_frames`; teams + wind derived), then re-broadcast it tick-by-tick
    through the same StreamingScorer + LiveFeedServer the live path uses — so OBS can re-record a fresh
    MP4. **AIS is replayed per tick** from the persisted `ais_observations` at that tick's UTC, so the
    surrounding traffic comes back too. Deterministic: same window → same broadcast.

    `speed`/`max_ticks`/`sleep` are injectable (tests run bounded + no-wait); `speed=1` re-records in
    real time."""
    from ais_feed import DEFAULT_TTL_MS, wire_near
    from broadcast_feed import build_feed
    feed = build_feed(store, instance_id, boats, meta=meta, course=course,
                      team_config=team_config, course_id=course_id)
    meta_s, course_s, teams, frames = feed["meta"], feed["course"], feed["teams"], feed["frames"]
    srv.publish_init(meta_s, course_s, teams, {})
    # `producer` defaults to enrich inside StreamingScorer; the RIE live producer (Phase 4) drops in here.
    scorer = StreamingScorer(meta_s, course_s, teams, producer=producer)
    # Carry the RRS committee signals for this instance into the producer feed so a RIE producer can
    # narrate every recall/restart/abnormal event (§4a). `enrich` ignores `raceSignals`, so this is
    # inert under the default producer. Read once — a regenerate is a bounded, immutable window.
    _signals_for = getattr(store, "race_signals_for", None)
    if _signals_for is not None:
        scorer.race_signals = [{"signal": s.signal, "tMs": s.t_ms, "meta": s.meta, "source": s.source}
                               for s in _signals_for(instance_id)]
    commentator = _simple_commentator(meta_s, teams, course_s) if simple else None
    track_start = meta_s["trackStartUtcMs"]
    secs = sorted(int(s) for s in frames)
    if max_ticks is not None:
        secs = secs[:max_ticks]
    dt = 1.0 / max(speed, 0.01)
    last: dict = {"standings": None, "positions": {}}
    for sec in secs:
        positions = frames[str(sec)]
        delta = scorer.push_frame(sec, positions)
        ais_list: list = []
        centroid = _centroid(positions)
        if centroid is not None:
            obs = store.ais_observations_at(track_start + sec * 1000, max_age_ms=DEFAULT_TTL_MS)
            excl = _own_mmsis(store, instance_id, track_start + sec * 1000, obs)
            ais_list = wire_near(obs, centroid[0], centroid[1], radius_m=ais_radius_m, exclude_mmsi=excl)
        evts = (commentator.tick(sec, delta["standings"], positions)
                if commentator is not None else delta["events"])
        srv.publish_delta(sec, delta["standings"], evts, positions, ais_list,
                          onboard=delta.get("rieOnboardDirectives"))
        last = {"standings": delta["standings"], "positions": positions}
        sleep(dt)
    tail = scorer.finish()                              # flush the trailing lag-seconds of commentary
    if not simple and tail.get("events") and secs:      # simple track has no finalisation lag
        srv.publish_delta(secs[-1], last["standings"], tail["events"], last["positions"])
    if scorer.result():
        srv.publish_result(scorer.result())


def run_regenerate(
    store,                                              # noqa: ANN001
    *,
    instance_id: str,
    course_id: str,
    title: str,
    t0: int,
    t1: int,
    gun_ms: int | None = None,
    team_config: dict | None = None,
    port: int = 8765,
    ais_radius_m: float = 500.0,
    speed: float = 1.0,
    simple: bool = False,
    producer: Callable[[dict], dict] | None = None,
) -> None:
    """Boot the LiveFeedServer and regenerate the `[t0, t1]` window of `instance_id` from `store`.
    Point an OBS Browser source at the printed scene URL to re-record the MP4.

    `producer` defaults to enrich inside StreamingScorer; pass a RieLiveProducer to re-render the
    RIE-only commentary/camera (the near-live-replay recovery + archive path)."""
    meta, course, boats = meta_course_from_store(
        store, instance_id=instance_id, course_id=course_id, title=title,
        gun_ms=gun_ms, track_start_ms=t0, track_end_ms=t1)
    # Replay consumes the SAME event + broadcast config as live — one engine, one config (requirement:
    # live/replay/sim parity). apply_broadcast_config loads the operator's teamConfig (team names +
    # flags) so the standings show nations, not raw team ids.
    meta = apply_event_config(store, event_id_for_instance(store, instance_id), meta=meta)
    meta, team_config = apply_broadcast_config(store, instance_id, meta=meta, team_config=team_config)
    srv = LiveFeedServer(host="0.0.0.0", port=port)
    srv.start(background=True)
    print(f"Regenerating {instance_id!r} [{t0}..{t1}] on :{srv.port}")
    print(f"Open / point OBS at: http://localhost:{srv.port}/core/web/index.html?feed=live")
    try:
        regenerate(store, srv, instance_id=instance_id, boats=boats, meta=meta, course=course,
                   team_config=team_config, course_id=course_id, ais_radius_m=ais_radius_m,
                   speed=speed, simple=simple, producer=producer)
        print("regeneration complete — serving for reconnects. Ctrl-C to stop.")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nstopping")
    finally:
        srv.stop()


# --- seeded offline demo (no rig) --------------------------------------------------------------
_ORIGIN_LAT, _ORIGIN_LON = 59.32100, 18.05200          # Riddarfjärden — open water ringed by the city

_M_PER_DEG = 6_371_000.0 * math.pi / 180.0
_INSTANCE, _GROUP, _COURSE = "demo3d", "g-demo", "demo3d-course"


def _ll(north_m: float, east_m: float) -> tuple[float, float]:
    return (_ORIGIN_LAT + north_m / _M_PER_DEG,
            _ORIGIN_LON + east_m / (_M_PER_DEG * math.cos(math.radians(_ORIGIN_LAT))))


def seed_demo_store():
    """Seed an in-memory store with two boats (RED/BLUE) + a wind reading, and return everything the
    driver needs plus a `move_once(t_ms)` that advances the boats — call it ~1 Hz from a thread."""
    from attribution import CourseRole, Entry, WindReading
    from ingest import IngestService
    from models import BoatNavRecord
    from slots import qr_coding
    from store import InMemoryStore

    now = int(time.time() * 1000)
    store = InMemoryStore()
    svc = IngestService(store, qr_coding.DEMO_SECRET)
    for boat, dev in [("2", "dev-red"), ("3", "dev-blue")]:
        fields = {"u": "GSYS", "b": boat, "sc": "Lime", "bt": "J/80", "mt": "Mast",
                  "bo": 8, "ml": -0.25, "h": 1.2, "hd": 0, "iss": now - 1000}
        svc.register(qr_coding.encode_url(fields, qr_coding.DEMO_SECRET), dev, now_ms=now - 1000)
    store.add_entry(Entry("GSYS:boat:2", _INSTANCE, "RED", now - 1000, None))
    store.add_entry(Entry("GSYS:boat:3", _INSTANCE, "BLUE", now - 1000, None))
    store.add_wind_reading(WindReading(_COURSE, now - 1000, 270.0, 8.0, "manual"))
    # static course marks → the seed exercises the SAME resolve_course path the real trip uses.
    sp, ss, wp = _ll(0, -60), _ll(0, 60), _ll(500, 0)
    for role, (lat, lon) in [("start_port", sp), ("start_stbd", ss),
                             ("finish_port", sp), ("finish_stbd", ss), ("M1", wp)]:
        store.add_course_role(CourseRole(_COURSE, role, static_lat=lat, static_lon=lon))

    meta, course, boats = meta_course_from_store(
        store, instance_id=_INSTANCE, course_id=_COURSE, group_id=_GROUP,
        title="V3 live demo — Red v Blue", now_ms=now)
    team_config = {"RED": {"name": "Red", "boats": {"GSYS:boat:2": {"sailId": "2", "color": "#C8102E"}}},
                   "BLUE": {"name": "Blue", "boats": {"GSYS:boat:3": {"sailId": "3", "color": "#0033A0"}}}}

    t0 = time.time()

    def move_once(t_ms: int) -> None:
        # Loop the boats up to the windward mark and back, forever, so the demo stream never runs
        # out of race (previously they sailed straight north off the course and the scene froze).
        elapsed = time.time() - t0
        leg_m, v = 500.0, 1.6                                     # start line → M1; ~3 kt
        for dev, east_m, north0, lag_s in [("dev-red", -12.0, 40.0, 0.0),
                                           ("dev-blue", 12.0, 20.0, 8.0)]:
            d = (v * max(0.0, elapsed - lag_s)) % (2 * leg_m)     # triangle: 0→leg→0→…
            up = d <= leg_m
            north = north0 + (d if up else 2 * leg_m - d)
            lat, lon = _ll(north, east_m)
            svc.ingest_nav(BoatNavRecord.from_wire(
                {"deviceId": dev, "timestamp": t_ms, "gnssTimestamp": t_ms, "sessionId": "demo",
                 "latitude": lat, "longitude": lon, "speed": v,
                 "course": 0.0 if up else 180.0}, dev))

    move_once(now)                                                # seed an initial fix so t=0 has data
    return {"store": store, "boats": boats, "meta": meta, "course": course,
            "team_config": team_config, "course_id": _COURSE, "instance_id": _INSTANCE,
            "group_id": _GROUP, "move_once": move_once}


def main() -> None:
    import argparse
    import os

    # Windows consoles default to cp1252 and crash (UnicodeEncodeError) on the unicode in our status
    # prints (→, —); force UTF-8 so background/redirected runs survive startup logging.
    for _stream in (sys.stdout, sys.stderr):
        if hasattr(_stream, "reconfigure"):
            _stream.reconfigure(encoding="utf-8", errors="replace")

    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--port", type=int, default=8765)
    ap.add_argument("--seed", action="store_true", default=True,
                    help="offline demo: seed a moving two-boat race (default)")
    ap.add_argument("--ais", action="store_true",
                    help="overlay AIS context vessels (needs $AISSTREAM_API_KEY)")
    ap.add_argument("--ais-radius", type=float, default=500.0, dest="ais_radius",
                    help="AIS overlay radius in metres around the boat (default 500; widen for "
                         "demos in low-traffic spots)")
    ap.add_argument("--wind", nargs="?", const="open-meteo", default=None,
                    choices=["open-meteo", "smhi"],
                    help="live wind from a real provider: open-meteo (default, no key) | smhi")
    # Live broadcast of a REAL prod instance (needs $VR_DATABASE_URL). Unlike the default seeded demo,
    # this reads live nav for --instance straight from the Postgres store the phones are uploading to.
    ap.add_argument("--live", action="store_true",
                    help="broadcast a real prod instance live from the DB (with --instance --course)")
    # Regeneration: re-render a past window from the DB (needs $VR_DATABASE_URL).
    ap.add_argument("--regenerate", action="store_true",
                    help="re-render a past window from the DB (with --instance --course --from --to)")
    ap.add_argument("--instance", help="instance id to broadcast/regenerate")
    ap.add_argument("--course", help="course id to broadcast/regenerate")
    ap.add_argument("--follow", action="store_true",
                    help="with --live: FOLLOW the operator's races on --course instead of pinning one "
                         "--instance. The broadcast stays with the race on air until it has a winner, "
                         "then moves to the next — one process for the whole regatta day, no restart "
                         "per heat (the renderer swaps the race in place, it never reloads)")
    ap.add_argument("--owner", help="with --live --follow: only follow this owner's races")
    ap.add_argument("--read-base", dest="read_base",
                    help="poll live positions from this read-API base (e.g. https://apps.viewregatta.com) "
                         "instead of per-tick DB queries — the co-located cached /positions endpoint "
                         "(Phase 0 of docs/plans/live-broadcast-feed-in-gcloud.md). With --follow and "
                         "no $VR_DATABASE_URL the WHOLE producer runs over REST: no database, so the "
                         "same command works against a local backend or gcloud, from the render PC or "
                         "the cloud, over live, replay or simulated races alike")
    ap.add_argument("--read-token", dest="read_token",
                    help="bearer token for --read-base when the races are not public")
    ap.add_argument("--from", dest="t0", type=int, help="window start (UTC ms)")
    ap.add_argument("--to", dest="t1", type=int, help="window end (UTC ms)")
    ap.add_argument("--title", default="Regenerated race", help="broadcast title for regeneration")
    ap.add_argument("--speed", type=float, default=1.0, help="re-render rate ×realtime (1 = real time)")
    ap.add_argument("--rie", action="store_true",
                    help="regenerate with the RIE-only producer (base+tactical narration) instead of enrich")
    ap.add_argument("--simple", action="store_true",
                    help="factual, team-racing-FREE commentary (non-team-race formats, e.g. the GSYS "
                         "sprint): start/leader/gaps/roundings/winner only — no combinations or rule-18")
    args = ap.parse_args()

    if args.regenerate:
        return _main_regenerate(args)
    if args.live:
        return _main_live(args)

    d = seed_demo_store()
    # Drive the boats ~1 Hz so the live read always has a fresh fix (boat_position_at, 5 s window).
    stop = threading.Event()

    def mover() -> None:
        while not stop.is_set():
            d["move_once"](int(time.time() * 1000))
            stop.wait(1.0)
    threading.Thread(target=mover, name="demo-mover", daemon=True).start()

    ais_key = os.environ.get("AISSTREAM_API_KEY") if args.ais else None
    if args.ais and not ais_key:
        print("AIS overlay: OFF - set $AISSTREAM_API_KEY to enable")
    try:
        run(d["store"], instance_id=d["instance_id"], group_id=d["group_id"], boats=d["boats"],
            meta=d["meta"], course=d["course"], team_config=d["team_config"],
            course_id=d["course_id"], port=args.port, ais_key=ais_key,
            ais_radius_m=args.ais_radius, wind_provider=args.wind, simple=args.simple)
    finally:
        stop.set()


def _main_live(args) -> None:                                       # noqa: ANN001 — argparse Namespace
    """`--live` CLI: broadcast a **real prod instance** live from Postgres — the production counterpart
    to the seeded demo. Meta/course/boats are derived from the store (`meta_course_from_store`, no
    window → `openEnded=True`, boats from the roster); `run` then publishes the live store forever
    with the same AIS/wind overlays as the demo path. The phones upload nav into this same store, so the
    scene follows the real track. Needs `$VR_DATABASE_URL` (e.g. the prod store via the Cloud SQL proxy)."""
    import os
    follow_mode = getattr(args, "follow", False)
    read_base = getattr(args, "read_base", None)
    if not (args.instance or follow_mode):
        raise SystemExit("--live needs either --instance <id> or --follow")
    if args.instance and not args.course:
        raise SystemExit("--live --instance needs --course <id> (a pinned race states its course)")
    # `--follow` needs NO course. A race carries its own `courseId`, so following the operator means
    # taking each race's course from the race — asking the deploy to know a course id in advance is
    # asking it to know something the race plane already says, and to be wrong when the day changes.
    dsn = os.environ.get("VR_DATABASE_URL")
    # THE DEPLOYMENT-INDEPENDENT PATH: `--follow --read-base <url>` with no database runs the whole
    # producer over REST (`rest_source.py`) — the same command against a local backend or gcloud, from
    # the render PC or from the cloud, over live, replay or simulated races alike. With a DSN the
    # store-backed path is used instead (co-located: no HTTP hop).
    rest_mode = follow_mode and not dsn and bool(read_base)
    if not dsn and not rest_mode:
        raise SystemExit("--live needs $VR_DATABASE_URL (the Postgres store the phones upload to), "
                         "or --follow --read-base <url> to run entirely over the REST API")
    store = None
    if dsn:
        from postgres_store import PostgresStore, connect
        store = PostgresStore(connect(dsn))

    def _presentation_for(instance: str, gun_ms: int) -> tuple:
        """One race's `(meta, course, boats, team_config)`. The gun is passed to `meta_course_from_store`
        so the roster — and therefore the teams, the standings and the camera — is resolved at the
        OPERATOR'S start signal, not at whatever moment the producer happened to look."""
        inst_row = store.get_race_instance(instance)
        race_course = args.course or getattr(inst_row, "course_id", None)
        m, c, b = meta_course_from_store(store, instance_id=instance, course_id=race_course,
                                         title=args.title, gun_ms=gun_ms)
        # The race's OWN course, carried on its meta so the follow loop can point the live wind feed
        # at it. Resolved here already — it just had nowhere to go.
        m["courseId"] = race_course
        # Config overlays, least- to most-specific: EVENT config (venue/boatClass/presentation for the
        # whole competition) first, then the operator's per-instance broadcast config (title / gun /
        # team display) — the production instance-meta model. Instance keys win.
        m = apply_event_config(store, event_id_for_instance(store, instance), meta=m)
        m, cfg = apply_broadcast_config(store, instance, meta=m)
        pre_s = ((m.get("startSequence") or {}).get("preStartS")) or 180
        m["gunUtcMs"] = gun_ms                          # the operator's gun wins over any stored one
        m["trackStartUtcMs"] = gun_ms - int(pre_s) * 1000
        return m, c, b, cfg

    def _waiting_picture(next_race, course_geo, positions_for):  # noqa: ANN001
        """The WAITING presentation's `_Waiting` kwargs, or None when there is nothing worth showing.

        A producer pointed at a race day spends its first minutes — and every gap between heats — with
        no race on air. Without this it publishes NOTHING, so the OBS browser source sits blank until
        the first gun: no venue, no fleet, none of the area-panning idle camera we built for exactly
        this moment. Only the SIMULATOR supplied one, so the picture was right in rehearsal and blank
        in production — the worst way round.

        What it shows is the NEXT race's fleet on the venue: the boats the operator has allocated but
        not yet started, published team-LESS, because naming teams here would assert a line-up for a
        race that has not begun. `_Waiting` adds `idle` and the sentinel clock; no scorer runs, since
        there is no race to score."""
        picked = next_race()
        if picked is None:
            return None                                  # nothing allocated yet: nothing to show
        instance, boats = picked
        geo = course_geo(instance)
        if not geo or not boats:
            return None
        return {
            "meta": {"title": args.title if args.title != "Regenerated race" else "Waiting for a race",
                     "user": args.owner or "", "instanceId": instance, "groupId": "g",
                     "raceType": "team_race_2"},
            "course": geo,
            # Team-LESS spectators labelled by hull: the renderer builds its boat meshes from `teams`,
            # so without them the sea would be empty.
            "teams": [{"teamId": f"hull-{b.rsplit(':', 1)[-1]}", "name": b.rsplit(":", 1)[-1],
                       "boats": [{"deviceId": b, "sailId": b.rsplit(":", 1)[-1], "visualId": b,
                                  "color": "#8899aa", "sternM": 0, "spectator": True}]}
                      for b in sorted(boats)],
            "fetch_positions": positions_for(instance, sorted(boats)),
        }

    from course import resolve_course as _resolve_course
    from v3_adapter import http_fetch_positions as _http_positions
    from v3_adapter import v3_fetch_positions as _v3_positions

    backfill = None
    waiting_for = None
    if rest_mode:
        # FOLLOW over REST: races, roster, course, wind and the join back-fill all come off the read
        # plane; positions off `--read-base` as before. No store, so no database credentials at all.
        from rest_source import RestBackend, rest_presentation_for
        backend = RestBackend(read_base, token=getattr(args, "read_token", None))
        races = StartedRaces(backend.instances, course_id=args.course, owner=args.owner)
        _presentation_for = rest_presentation_for(backend, course_id=args.course,
                                                  title=(args.title if args.title != "Regenerated race"
                                                         else None))
        backfill = lambda inst, _boats, t0, t1: backend.frames(  # noqa: E731 — a thin adapter
            inst, t_from=t0, t_to=t1, track_start_ms=t0)
        meta, course, boats, team_config = {}, {}, [], None

        def _next_rest():
            """The nearest unstarted race on this course that already has an allocation."""
            now = int(time.time() * 1000)
            for row in backend.instances():
                if row.get("officialStartUtcMs"):
                    continue
                if args.course and row.get("courseId") != args.course:
                    continue
                if args.owner and row.get("ownerUserId") not in (None, args.owner):
                    continue
                inst = row.get("instanceId")
                grid = backend.roster(inst, now) or {}
                fleet = {b for team in grid.values() for b in team}
                if fleet:
                    return inst, fleet
            return None

        waiting_for = lambda: _waiting_picture(               # noqa: E731 — a thin adapter
            _next_rest,
            lambda inst: backend.course(inst, args.course or backend.course_of(inst),
                                        int(time.time() * 1000)),
            lambda inst, _boats: _http_positions(read_base, inst))
        print(f"following races over REST at {read_base} (no database)")
    elif follow_mode:
        # FOLLOW the operator's races on this course: one process for the whole day. `--instance`, if
        # given, is ignored here — which race is on air is the operator's start signal, not a flag.
        races = StartedRaces(lambda: store.list_race_instances(owner_user_id=args.owner),
                             course_id=args.course)
        meta, course, boats, team_config = {}, {}, [], None

        def _next_store():
            """The nearest unstarted race on this course that already has an allocation."""
            from attribution import resolve_roster
            now = int(time.time() * 1000)
            for inst in store.list_race_instances(owner_user_id=args.owner):
                if inst.official_start_utc_ms:
                    continue
                if args.course and inst.course_id != args.course:
                    continue
                grid = resolve_roster(
                    store.entries_covering_instance(inst.id, now),
                    now, instance_id=inst.id).get(inst.id, {})
                fleet = {b for team in grid.values() for b in team}
                if fleet:
                    return inst.id, fleet
            return None

        waiting_for = lambda: _waiting_picture(               # noqa: E731 — a thin adapter
            _next_store,
            lambda inst: _resolve_course(
                store, args.course or getattr(store.get_race_instance(inst), "course_id", None) or "",
                int(time.time() * 1000)),
            lambda _inst, fleet: _v3_positions(store, fleet))
    else:
        meta, course, boats = meta_course_from_store(
            store, instance_id=args.instance, course_id=args.course, title=args.title)
        meta = apply_event_config(store, event_id_for_instance(store, args.instance), meta=meta)
        meta, team_config = apply_broadcast_config(store, args.instance, meta=meta)
    ais_key = os.environ.get("AISSTREAM_API_KEY") if args.ais else None
    if args.ais and not ais_key:
        print("AIS overlay: OFF - set $AISSTREAM_API_KEY to enable")
    producer = None
    if getattr(args, "rie", False):
        # v3/scoring is already on sys.path (via StreamingScorer); add the repo root for v3.rie.
        sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
        from score_race import enrich

        from v3.rie.live_producer import RieLiveProducer
        producer = RieLiveProducer(scorer=enrich)
    run(store, instance_id=args.instance, group_id=meta.get("groupId", "g"), boats=boats,
        meta=meta, course=course, team_config=team_config, course_id=args.course, port=args.port,
        ais_key=ais_key, ais_radius_m=args.ais_radius, wind_provider=args.wind,
        simple=getattr(args, "simple", False), read_base=getattr(args, "read_base", None),
        producer=producer,
        follow=((races, _presentation_for, backfill, waiting_for) if follow_mode else None))


def _main_regenerate(args) -> None:                                 # noqa: ANN001 — argparse Namespace
    """`--regenerate` CLI: re-render a past `[--from, --to]` window of an instance from Postgres."""
    import os
    if not (args.instance and args.course and args.t0 and args.t1):
        raise SystemExit("--regenerate needs --instance --course --from <utcMs> --to <utcMs>")
    dsn = os.environ.get("VR_DATABASE_URL")
    if not dsn:
        raise SystemExit("--regenerate needs $VR_DATABASE_URL (the Postgres store written during the trip)")
    from postgres_store import PostgresStore, connect
    store = PostgresStore(connect(dsn))
    producer = None
    if getattr(args, "rie", False):
        # v3/scoring is already on sys.path (imported via StreamingScorer); add the repo root for v3.rie.
        sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
        from score_race import enrich

        from v3.rie.live_producer import RieLiveProducer
        producer = RieLiveProducer(scorer=enrich)
    run_regenerate(store, instance_id=args.instance, course_id=args.course, title=args.title,
                   t0=args.t0, t1=args.t1, port=args.port, speed=args.speed,
                   simple=getattr(args, "simple", False), producer=producer)


if __name__ == "__main__":
    main()