# SPDX-License-Identifier: GPL-3.0-or-later
"""Background AIS history recorder: live feed -> voyage tracker -> store.

The live decode/bind pipeline calls :meth:`HistoryRecorder.record` once per bound
position fix. That call only enqueues (thread-safe, non-blocking) and returns, so
the GUI/feed thread is never held up. A single daemon worker thread owns the
:class:`VoyageTracker` and the store's SQLite/PostGIS connection (both are
thread-bound), drains the queue, and writes in batches flushed on a ~1 s timer.

Three cumulative stages gate what is persisted (see history_store):
  * stage 1  register + events only
  * stage 2  + voyages, geometry STORED as a thinned LineString accumulated in RAM
  * stage 3  + per-fix Points; the voyage line is then DERIVED (never stored twice)

The bounded queue is the RAM guarantee: in a dense port scene the worker batches
thousands of inserts/s while the queue caps at ``max_queue``; excess is dropped
(counted) rather than growing memory without bound.
"""
import math
import queue
import threading
import time

from .voyage_tracker import VoyageTracker


def _meters(lat, lon, lat2, lon2):
    """Rough planar metres between two lon/lat (equirectangular, fine at fix scale)."""
    k = 111320.0
    dx = (lon2 - lon) * k * math.cos(math.radians((lat + lat2) * 0.5))
    dy = (lat2 - lat) * k
    return math.hypot(dx, dy)


class HistoryRecorder:
    def __init__(self, store_factory, stage=1, node_id="local",
                 plugin_version="", logic_version="", slow_kn=2.0, debounce=3,
                 coverage_gap_s=3600.0, flush_interval=1.0, sweep_interval=60.0,
                 max_queue=100000, min_move_m=15.0, max_vertices=10000,
                 receiver=(None, None), reconnect_min_s=5.0,
                 reconnect_max_s=60.0):
        """store_factory() is invoked ON the worker thread (connections are
        thread-bound). stage in {1,2,3}. min_move_m thins stage-2 line vertices."""
        self.stage = int(stage)
        self.node_id = node_id
        self.plugin_version = plugin_version
        self.logic_version = logic_version
        self.min_move_m = min_move_m
        self.max_vertices = max_vertices
        self.flush_interval = flush_interval
        self.sweep_interval = sweep_interval
        self.receiver = receiver
        self.reconnect_min_s = reconnect_min_s
        self.reconnect_max_s = reconnect_max_s
        self._factory = store_factory
        self._tracker_kw = dict(node_id=node_id, slow_kn=slow_kn,
                                debounce=debounce, coverage_gap_s=coverage_gap_s)
        self._q = queue.Queue(maxsize=max_queue)
        self._stop = threading.Event()
        self._thread = None
        self._err = None
        self._reconnecting = False
        # stats (display only; single-writer-per-field, GIL-atomic ints)
        self.n_in = 0
        self.n_dropped = 0
        self.n_written = 0
        self.n_reconnects = 0
        self.n_discarded = 0             # trivial/trackless voyages discarded
        self._dangling_closed = False    # close orphaned open voyages once, at start

    # ---- lifecycle ------------------------------------------------------- #
    def start(self):
        if self._thread is not None:
            return
        self._stop.clear()
        self._thread = threading.Thread(target=self._run, name="ais-history",
                                        daemon=True)
        self._thread.start()

    def stop(self, timeout=10.0):
        """Signal shutdown and wait for the worker to finalise open voyages,
        flush, and close the store. Safe to call from the GUI thread."""
        self._stop.set()
        t = self._thread
        if t is not None:
            t.join(timeout)
        self._thread = None

    @property
    def error(self):
        return self._err

    @property
    def reconnecting(self):
        return self._reconnecting

    # ---- producer side (GUI/feed thread) -------------------------------- #
    def record(self, mmsi, t, lon, lat, sog=None, cog=None, heading=None,
               nav_status=None, static=None, routeid=None, route_km=None,
               flow=None):
        """Enqueue one bound position fix. Non-blocking; drops (and counts) when
        the queue is full so a burst can never grow memory unbounded."""
        self.n_in += 1
        rec = (mmsi, t, lon, lat, sog, cog, heading, nav_status, static,
               routeid, route_km, flow)
        try:
            self._q.put_nowait(rec)
        except queue.Full:
            self.n_dropped += 1

    # ---- worker thread --------------------------------------------------- #
    def _connect(self):
        """Build + provision a store (idempotent). Used for the first connect and
        every reconnect after a dropped backend connection."""
        store = self._factory()
        store.ensure_schema()
        store.ensure_indexes(self.stage)
        store.ensure_views(self.stage)
        store.register_node(self.node_id, self.plugin_version,
                            self.logic_version, self.receiver[0],
                            self.receiver[1])
        # Once, at first connect (not on reconnects), finalise voyages left open by
        # a previous session — they can never continue and otherwise pile up.
        if not self._dangling_closed:
            try:
                store.close_dangling_voyages(self.node_id)
            except Exception:                                      # noqa: BLE001  # nosec B110
                pass
            self._dangling_closed = True
        return store

    def _run(self):
        tracker = VoyageTracker(**self._tracker_kw)
        vlines = {}          # voyage_id -> [(lon,lat), ...]  (stage-2 line accum)
        cur_voyage = {}      # mmsi -> open voyage_id (to stamp events)
        static_sig = {}      # mmsi -> signature of last-written static
        store = None
        backoff = 0.0
        retry_at = 0.0
        last_flush = last_sweep = time.time()
        while True:
            stopping = self._stop.is_set()

            # (re)establish the backend connection, backing off on repeated
            # failure. A dropped connection must never kill the recorder — it
            # keeps buffering (bounded queue) and retries.
            if store is None:
                now = time.time()
                if now < retry_at:
                    if stopping and self._q.empty():
                        break
                    time.sleep(0.25)
                    continue
                try:
                    store = self._connect()
                    self._err = None
                    backoff = 0.0
                    last_flush = last_sweep = time.time()
                except Exception as exc:                             # noqa: BLE001
                    self._err = exc
                    self._reconnecting = True
                    backoff = min(backoff * 2 if backoff else self.reconnect_min_s,
                              self.reconnect_max_s)
                    retry_at = time.time() + backoff
                    if stopping and self._q.empty():
                        break
                    continue
                self._reconnecting = False

            try:
                rec = self._q.get(timeout=0.2)
            except queue.Empty:
                rec = None
            try:
                if rec is not None:
                    self._process(rec, tracker, store, vlines, cur_voyage,
                                  static_sig)
                now = time.time()
                if now - last_flush >= self.flush_interval:
                    store.flush()
                    last_flush = now
                if now - last_sweep >= self.sweep_interval:
                    for ev in tracker.sweep(now):
                        self._apply(ev, store, vlines, cur_voyage)
                    last_sweep = now
            except Exception as exc:                                 # noqa: BLE001
                # backend error mid-op (e.g. connection dropped): drop the store
                # and reconnect on the next iteration; keep the recorder alive.
                self._err = exc
                self._reconnecting = True
                self.n_reconnects += 1
                try:
                    store.close()
                except Exception:                                    # noqa: BLE001  # nosec B110
                    pass
                store = None
                backoff = min(backoff * 2 if backoff else self.reconnect_min_s,
                              self.reconnect_max_s)
                retry_at = time.time() + backoff
                continue
            if stopping and rec is None and self._q.empty():
                break

        # finalise: close every still-open voyage so its line/row is written
        if store is not None:
            try:
                for mmsi, vid in list(tracker.open_voyages().items()):
                    self._apply({"kind": "voyage_end", "mmsi": mmsi,
                                 "voyage_id": vid, "t": time.time(),
                                 "reason": "shutdown"},
                                store, vlines, cur_voyage)
                store.flush()
            except Exception:                                        # noqa: BLE001  # nosec B110
                pass
            try:
                store.close()
            except Exception:                                        # noqa: BLE001  # nosec B110
                pass

    def _process(self, rec, tracker, store, vlines, cur_voyage, static_sig):
        (mmsi, t, lon, lat, sog, cog, heading, nav_status, static,
         routeid, route_km, flow) = rec
        # stage 1: keep the vessel register fresh (only write on a real change)
        if self.stage >= 1 and static:
            sig = (static.get("name"), static.get("callsign"), static.get("imo"),
                   static.get("ship_type"), static.get("destination"),
                   static.get("eta"), static.get("draught"))
            if static_sig.get(mmsi) != sig:
                static_sig[mmsi] = sig
                st = dict(static); st["t"] = t
                store.upsert_vessel(mmsi, st, self.node_id)
        # feed the segmentation state machine, then persist its events
        for ev in tracker.update(mmsi, t, lon, lat, sog=sog, cog=cog,
                                 heading=heading, nav_status=nav_status,
                                 routeid=routeid, route_km=route_km, flow=flow):
            if ev["kind"] == "voyage_start":
                ev["_static"] = static
            self._apply(ev, store, vlines, cur_voyage)

    def _apply(self, ev, store, vlines, cur_voyage):
        kind = ev["kind"]
        mmsi = ev["mmsi"]
        if kind == "voyage_start":
            cur_voyage[mmsi] = ev["voyage_id"]
            vlines[ev["voyage_id"]] = []
            if self.stage >= 2:
                s = ev.get("_static") or {}
                store.begin_voyage(ev["voyage_id"], mmsi, ev["t"], s, self.node_id)
        elif kind == "voyage_end":
            line = vlines.pop(ev["voyage_id"], None)
            if self.stage == 2:
                if line and len(line) >= 2:   # a real track -> store the line
                    store.end_voyage(ev["voyage_id"], ev["t"], ev["reason"], line)
                else:                         # 1-point / trackless -> discard
                    store.discard_voyage(ev["voyage_id"])
                    self.n_discarded += 1
            elif self.stage >= 3:             # points present -> line derived
                store.end_voyage(ev["voyage_id"], ev["t"], ev["reason"], None)
            cur_voyage.pop(mmsi, None)
        elif kind == "fix":
            if self.stage >= 3:
                store.add_fix(ev)
                self.n_written += 1
            elif self.stage == 2:
                self._accum_vertex(vlines.get(ev["voyage_id"]), ev["lon"], ev["lat"])
        elif kind in ("coverage_in", "coverage_out", "nav_change"):
            if self.stage >= 1:
                out = {"mmsi": mmsi, "t": ev["t"], "kind": kind,
                       "voyage_id": cur_voyage.get(mmsi), "nav_status": None,
                       "detail": None}
                if kind == "nav_change":
                    out["nav_status"] = ev["new"]
                    out["detail"] = "%s->%s" % (ev["old"], ev["new"])
                store.add_event(out, self.node_id)
                self.n_written += 1

    def _accum_vertex(self, vl, lon, lat):
        """Append a thinned vertex to a stage-2 voyage line (drops sub-threshold
        moves; caps per-voyage vertices to bound RAM)."""
        if vl is None or len(vl) >= self.max_vertices:
            return
        if vl and _meters(vl[-1][1], vl[-1][0], lat, lon) < self.min_move_m:
            return
        vl.append((lon, lat))
