#!/usr/bin/env python3
"""ninlight recovery/migration station.

Serves firmware images to lamps that look for a rescue station while booting (mode
"recovery": ninlight_<target>_ota.img) or to old Klyqa lamps looking for their original
update server (mode "migration": qcx_firmware_<product-id>_<version>_ota_light.img).
Deliberately minimal: only files matching one of these two exact patterns ever leave the
machine, and only the pattern for the active mode is served.

Standard library only - the Pi gets no extra packages.
"""
from __future__ import annotations

import argparse
import contextlib
import json
import os
import re
import select
import socket
import sys
import tempfile
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

# products.py lives next to this file. sys.path[0] already covers that when server.py is
# run the normal way (as a script or via `python3 server.py ...`), but a test that loads
# this module through importlib.util.spec_from_file_location (see test_server.py's
# test_log_error_is_not_silenced_by_log_message) does not get that for free.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import products

TARGETS = ("evk", "e27_rgb", "e14_rgb", "gu10_rgb", "e27_cw", "e14_cw", "gu10_cw", "g95_cw")
PATH_RE = re.compile(r"^/ninlight_(" + "|".join(TARGETS) + r")_ota\.img$")
TRAILER_SIZE = 512
TRAILER_MAGIC = b"NINL"
VERSION_OFF = 21
VERSION_LEN = 16
RECENT_MAX = 8
CHUNK = 32 * 1024


def now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def trailer_version(path: Path) -> str:
    """Firmware version from the image trailer, or '?' if the file carries none."""
    try:
        with path.open("rb") as f:
            if path.stat().st_size < TRAILER_SIZE:
                return "?"
            f.seek(-TRAILER_SIZE, os.SEEK_END)
            tail = f.read(TRAILER_SIZE)
    except OSError:
        return "?"
    if tail[0:4] != TRAILER_MAGIC:
        return "?"
    return tail[VERSION_OFF:VERSION_OFF + VERSION_LEN].rstrip(b"\x00").decode("ascii", "replace")


class Station:
    def __init__(self, image_dir: Path, state_path: Path, stall_after: int | None,
                 mode: str = "recovery", control_path: Path | None = None) -> None:
        self.image_dir = image_dir
        self.state_path = state_path
        self.stall_after = stall_after
        self.mode = mode
        self.control_path = control_path or Path("/run/ninlight-station/control.json")
        self.since = now_iso()
        self.served = 0
        self.recent: list[dict] = []
        # Zuletzt erkanntes Modell im Migrationsmodus, auch wenn (noch) nichts ausgeliefert
        # wurde - das ist es, was die Anzeige gross darstellt (Spec 5.2).
        self.last_seen: dict | None = None
        self._manifest_mtime: float | None = None
        self._manifest_version = "unbekannt"
        # Cumulative bytes already delivered to each client while --stall-after is active,
        # keyed by client IP. A lamp whose download stalls re-opens a fresh connection and
        # resumes with Range right where it left off, so a per-request counter would hand
        # every reconnect a brand-new --stall-after allowance and the stall would never
        # actually stick (see reserve_stall_bytes below).
        self.stall_sent: dict[str, int] = {}
        # ThreadingHTTPServer runs one handler thread per connection, and several lamps
        # can be downloading at once. Every mutation of served/recent/stall_sent and the
        # snapshot that gets serialized to disk must happen under this same lock, or
        # concurrent requests lose increments and stomp on each other's status.json writes.
        self.lock = threading.RLock()
        self.write_state()

    def image_path(self, target: str) -> Path:
        return self.image_dir / f"ninlight_{target}_ota.img"

    def migrate_image_path(self, target: str, variant: str) -> Path:
        return self.image_dir / "migrate" / f"{target}_{variant}.img"

    def manifest_version(self) -> str:
        """Version aus <image_dir>/migrate/manifest.json, neu eingelesen bei jeder
        Aenderung der Datei (mtime-Vergleich). Fehlt sie oder ist sie kaputt, ist die
        Version "unbekannt" - die Migrationsimages selbst tragen keinen ninlight-Trailer,
        aus dem der Server sie lesen koennte (Spec Abschnitt 4)."""
        path = self.image_dir / "migrate" / "manifest.json"
        try:
            mtime = path.stat().st_mtime
        except OSError:
            self._manifest_mtime = None
            self._manifest_version = "unbekannt"
            return self._manifest_version
        if mtime != self._manifest_mtime:
            try:
                data = json.loads(path.read_text())
                self._manifest_version = data.get("ninlight_version", "unbekannt")
            except (OSError, ValueError):
                self._manifest_version = "unbekannt"
            self._manifest_mtime = mtime
        return self._manifest_version

    def read_control(self):
        """Phase und Freigabe. Fehlt die Datei oder ist sie unlesbar/kaputt, gilt "auto" -
        eine fehlende Steuerdatei darf eine Migration nie blockieren, sondern hoechstens
        bequemer machen als beabsichtigt. Bei jeder Anfrage frisch gelesen, damit
        "ninlight-station arm" ohne Neustart des Dienstes wirkt."""
        try:
            d = json.loads(self.control_path.read_text())
        except (OSError, ValueError):
            return "auto", None
        phase = d.get("phase")
        if phase not in ("auto", "listen", "armed"):
            return "auto", None
        return phase, d.get("armed_target")

    def write_control(self, phase: str, armed_target: str | None) -> None:
        with self.lock:
            try:
                self.control_path.parent.mkdir(parents=True, exist_ok=True)
                fd, tmp = tempfile.mkstemp(dir=self.control_path.parent, prefix=".control-")
                try:
                    with os.fdopen(fd, "w") as f:
                        json.dump({"phase": phase, "armed_target": armed_target}, f)
                    # Dasselbe Muster wie write_state() unten, aus demselben Grund: der
                    # Dienst laeuft als root, mkstemp() legt 0600 an, und ohne das
                    # explizite chmod() koennte das CLI die Steuerdatei nicht lesen.
                    os.chmod(tmp, 0o644)
                    os.replace(tmp, self.control_path)
                except BaseException:
                    with contextlib.suppress(OSError):
                        os.unlink(tmp)
                    raise
            except OSError as e:
                print(f"warning: cannot write {self.control_path}: {e}", flush=True)

    def reserve_stall_bytes(self, ip: str, n: int) -> int:
        """How many of the next `n` bytes this client may still receive under --stall-after.

        Tracks bytes cumulatively per client IP, across requests, so a stalled connection
        that the lamp abandons and resumes via Range does not get a fresh budget just
        because it is a new TCP connection. Only called when stall_after is set.
        """
        with self.lock:
            sent_so_far = self.stall_sent.get(ip, 0)
            allowed = max(0, min(n, self.stall_after - sent_so_far))
            self.stall_sent[ip] = sent_so_far + allowed
            return allowed

    def record(self, ip: str, target: str, version: str, nbytes: int, status: str) -> None:
        with self.lock:
            if status == "ok":
                self.served += 1
            self.recent.append({"ts": now_iso(), "ip": ip, "target": target,
                                "version": version, "bytes": nbytes, "status": status})
            del self.recent[:-RECENT_MAX]
            self.write_state()

    def record_migration(self, ip: str, target: str | None, variant: str | None,
                          product_id: str, lamp_version: str, product_name: str | None,
                          version: str, nbytes: int, status: str) -> None:
        """Protokolliert eine Migrationsanfrage. Anders als record() oben wird hier auch
        dann ein Eintrag angelegt (Status "listen"/"unknown"/"notfound"), wenn nichts
        ausgeliefert wurde - das ist der ganze Zweck der Horchphase. last_seen wird nur bei
        einer erkannten Produkt-ID gesetzt (target is not None); eine unbekannte Produkt-ID
        hat kein Modell, das die Anzeige zeigen koennte."""
        with self.lock:
            if status == "ok":
                self.served += 1
            entry = {"ts": now_iso(), "ip": ip, "target": target, "variant": variant,
                     "product_id": product_id, "lamp_version": lamp_version,
                     "version": version, "bytes": nbytes, "status": status}
            self.recent.append(entry)
            del self.recent[:-RECENT_MAX]
            if target is not None:
                self.last_seen = {"product_id": product_id, "target": target,
                                   "variant": variant, "product_name": product_name,
                                   "lamp_version": lamp_version, "ip": ip, "ts": entry["ts"]}
            self.write_state()

    def write_state(self) -> None:
        with self.lock:
            payload = {"since": self.since, "mode": self.mode, "served": self.served,
                       "recent": list(self.recent)}
            if self.mode == "migration":
                phase, armed_target = self.read_control()
                payload["phase"] = phase
                payload["armed_target"] = armed_target
                payload["last_seen"] = self.last_seen
            try:
                self.state_path.parent.mkdir(parents=True, exist_ok=True)
                fd, tmp = tempfile.mkstemp(dir=self.state_path.parent, prefix=".status-")
                try:
                    with os.fdopen(fd, "w") as f:
                        json.dump(payload, f, indent=2)
                    # The service runs as root and mkstemp() creates the file 0600, so
                    # without this an operator reading status.json as a normal user (see
                    # tools/station/ninlight-station) gets "permission denied" - and since
                    # os.replace() preserves the mode it arrives with, that wrong mode
                    # would otherwise survive onto the live file on every single write,
                    # not just the first. os.chmod() ignores the process umask (unlike the
                    # mode argument to mkstemp/open), so a restrictive umask on the Pi
                    # can't quietly bring the 0600 back. Set it before the rename so
                    # there's never a window where the path holds a 0600 file.
                    os.chmod(tmp, 0o644)
                    os.replace(tmp, self.state_path)
                except BaseException:
                    with contextlib.suppress(OSError):
                        os.unlink(tmp)
                    raise
            except OSError as e:
                print(f"warning: cannot write {self.state_path}: {e}", flush=True)


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = "ninlight-station"

    @property
    def station(self) -> Station:
        return self.server.station  # type: ignore[attr-defined]

    def log_message(self, fmt: str, *args) -> None:  # noqa: D102 - quieter default log
        pass

    def log_error(self, fmt: str, *args) -> None:
        # BaseHTTPRequestHandler.log_error() calls log_message() by default, which we
        # silence above in favour of our own log() calls - but that also silences it for
        # requests that never reach do_GET/do_POST (a malformed request line, a client
        # timeout mid-request). The station is the only diagnostic tool available during a
        # rescue, so those cases must still produce a line instead of nothing.
        try:
            client = self.client_address[0]
        except (AttributeError, IndexError):
            client = "?"
        print(f"{now_iso()} {client} ERROR {fmt % args}", flush=True)

    def log(self, status: int, nbytes: int, started: float) -> None:
        print(f"{now_iso()} {self.client_address[0]} {self.path} {status} "
              f"{nbytes}B {time.monotonic() - started:.1f}s", flush=True)

    def do_POST(self) -> None:
        started = time.monotonic()
        self.reject(405, "method not allowed")
        self.log(405, 0, started)

    do_PUT = do_DELETE = do_HEAD = do_POST

    def reject(self, status: int, text: str) -> None:
        body = (text + "\n").encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        started = time.monotonic()
        if self.station.mode == "migration":
            self._do_get_migration(started)
        else:
            self._do_get_recovery(started)

    def _do_get_recovery(self, started: float) -> None:
        m = PATH_RE.match(self.path)
        if not m:
            self.reject(404, "not found")
            self.log(404, 0, started)
            return
        target = m.group(1)
        path = self.station.image_path(target)
        if not path.is_file():
            # Record before responding: a client that sees the 404 complete may check
            # status.json right away, so the entry must already be on disk by then.
            self.station.record(self.client_address[0], target, "?", 0, "notfound")
            self.reject(404, "no image for this device")
            self.log(404, 0, started)
            return

        version = trailer_version(path)
        sent, status = self._serve_file(path, started)
        # Record only now that the transfer has actually finished (or failed): "ok" must
        # never be written before the last byte is confirmed sent, or status.json can
        # claim success for a device that never received the end of the image.
        self.station.record(self.client_address[0], target, version, sent, status)

    def _do_get_migration(self, started: float) -> None:
        parsed = products.parse_request(self.path)
        if parsed is None:
            # Ein fremder Pfad, kein qcx_firmware_*-Muster - kein Protokolleintrag in
            # status.json, das waere nur Rauschen (Brief Schritt 3.4).
            self.reject(404, "not found")
            self.log(404, 0, started)
            return
        pid, lamp_version, product = parsed
        ip = self.client_address[0]
        if product is None:
            self.station.record_migration(ip, None, None, pid, lamp_version, None, "?", 0,
                                           "unknown")
            self.reject(404, "unknown product")
            self.log(404, 0, started)
            return

        target, variant, name = product.target, product.variant, product.name
        version = self.station.manifest_version()
        phase, armed_target = self.station.read_control()

        if phase == "listen" or (phase == "armed" and armed_target != target):
            # last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase:
            # die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht.
            self.station.record_migration(ip, target, variant, pid, lamp_version, name,
                                           version, 0, "listen")
            self.reject(404, "listening only")
            self.log(404, 0, started)
            return

        path = self.station.migrate_image_path(target, variant)
        if not path.is_file():
            self.station.record_migration(ip, target, variant, pid, lamp_version, name,
                                           version, 0, "notfound")
            self.reject(404, "no image for this device")
            self.log(404, 0, started)
            return

        sent, status = self._serve_file(path, started)
        if phase == "armed" and status == "ok":
            # Erst nach dem vollstaendigen Senden zurueck auf "listen" - _serve_file() ist
            # an dieser Stelle bereits fertig. Bricht der Transfer ab (status == "aborted"),
            # bleibt die Freigabe stehen, damit die Lampe es beim naechsten Boot noch
            # einmal versuchen kann.
            self.station.write_control("listen", None)
        self.station.record_migration(ip, target, variant, pid, lamp_version, name, version,
                                       sent, status)

    def _serve_file(self, path: Path, started: float) -> tuple[int, str]:
        """Sendet `path` mit Range-Unterstuetzung; liefert (gesendete Bytes, Status).
        Gemeinsam fuer Rettungs- und Migrationsroute - Uebertragungsmechanik (Range,
        --stall-after, Abbruchbehandlung) ist identisch, nur was mit dem Ergebnis
        geschieht, unterscheidet sich."""
        size = path.stat().st_size
        start = 0
        rng = self.headers.get("Range", "")
        if rng.startswith("bytes="):
            first = rng.split("=", 1)[1].split("-", 1)[0]
            if first.isdigit() and int(first) < size:
                start = int(first)

        self.send_response(206 if start else 200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Length", str(size - start))
        if start:
            self.send_header("Content-Range", f"bytes {start}-{size - 1}/{size}")
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("Connection", "close")
        self.end_headers()

        sent = 0
        status = "ok"
        remaining = size - start
        client_ip = self.client_address[0]
        try:
            with path.open("rb") as f:
                f.seek(start)
                while sent < remaining:
                    read_size = min(CHUNK, remaining - sent)
                    if self.station.stall_after is not None:
                        # Reserve against this client's cumulative budget, not just this
                        # request's byte count - see Station.reserve_stall_bytes.
                        read_size = self.station.reserve_stall_bytes(client_ip, read_size)
                        if read_size == 0:
                            # test aid: hold the connection open and send nothing more, so
                            # the firmware's stall handling can be exercised on real
                            # hardware. We still notice if the peer gives up and closes its
                            # side, so a test (or an impatient client) that hangs up is
                            # recorded as "aborted" instead of leaking the thread forever.
                            while True:
                                readable, _, _ = select.select([self.connection], [], [], 1.0)
                                if readable:
                                    try:
                                        peek = self.connection.recv(1, socket.MSG_PEEK)
                                    except OSError:
                                        peek = b""
                                    if not peek:
                                        raise ConnectionResetError("peer closed during stall")
                    chunk = f.read(read_size)
                    if not chunk:
                        break
                    self.wfile.write(chunk)
                    sent += len(chunk)
        except (BrokenPipeError, ConnectionResetError, TimeoutError):
            status = "aborted"
        self.log(206 if start else 200, sent, started)
        return sent, status


def resolve_mode(explicit_mode: str | None, mode_file: Path | None) -> str:
    """Bestimmt den Betriebsmodus. Vorrang: ein ausdrueckliches --mode schlaegt
    --mode-file, --mode-file schlaegt die Vorgabe "recovery". Eine fehlende oder
    unlesbare/unerwartete Modusdatei bedeutet ebenfalls "recovery" - Task 4 baut die
    systemd-Unit darauf auf, dass der Modus ohne Neustart-Argumentaenderung aus einer
    Datei kommt (/run/ninlight-station/mode), und ein fehlerhafter Zustand dieser Datei
    darf den Dienst nicht zum Absturz bringen oder unbeabsichtigt scharf schalten."""
    if explicit_mode is not None:
        return explicit_mode
    if mode_file is not None:
        try:
            text = mode_file.read_text().strip()
        except OSError:
            return "recovery"
        if text in ("recovery", "migration"):
            return text
        return "recovery"
    return "recovery"


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="ninlight recovery/migration station HTTP server")
    p.add_argument("--image-dir", required=True, type=Path)
    p.add_argument("--bind", default="192.168.8.1")
    p.add_argument("--port", type=int, default=80)
    p.add_argument("--state", type=Path, default=Path("/run/ninlight-station/status.json"))
    p.add_argument("--mode", choices=("recovery", "migration"), default=None,
                   help="ausdruecklicher Modus - schlaegt --mode-file (Vorgabe: siehe dort)")
    p.add_argument("--mode-file", type=Path, default=None,
                   help="Datei mit 'recovery' oder 'migration'; wird ohne --mode verwendet, "
                        "Vorgabe bei fehlender/kaputter Datei ist 'recovery'")
    p.add_argument("--control", type=Path, default=Path("/run/ninlight-station/control.json"),
                   help="Steuerdatei fuer Phase/Freigabe im Migrationsmodus")
    p.add_argument("--stall-after", type=int, default=None,
                   help="test aid: stop sending after N bytes but keep the connection open")
    a = p.parse_args(argv)

    if not a.image_dir.is_dir():
        print(f"error: image directory {a.image_dir} does not exist", file=sys.stderr)
        return 2

    mode = resolve_mode(a.mode, a.mode_file)
    station = Station(a.image_dir, a.state, a.stall_after, mode=mode, control_path=a.control)
    httpd = ThreadingHTTPServer((a.bind, a.port), Handler)
    httpd.daemon_threads = True
    httpd.station = station  # type: ignore[attr-defined]
    images = sorted(p.name for p in a.image_dir.glob("ninlight_*_ota.img"))
    print(f"{now_iso()} listening on {a.bind}:{a.port}, mode={mode}, {len(images)} image(s): "
          f"{', '.join(images) or 'none'}", flush=True)
    if a.stall_after is not None:
        print(f"{now_iso()} WARNING: --stall-after {a.stall_after} is a test aid", flush=True)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    raise SystemExit(main())