"""The live-tracking store and its rules (AV-022).

What is stored (its own SQLite file, its own migration table -- never the
Journey database):

* ``live_journeys`` -- what a public page may show of a Journey: its title
  and route coordinates, published by the owner's panel.
* ``live_links`` -- one row per issued link: kind (``view``/``push``), the
  SHA-256 of its token (the token itself is never stored), an optional PBKDF2
  password hash, when it was issued and revoked.
* ``live_positions`` -- what a push link sent, stamped with the server's time.
* ``live_push_log`` / ``live_unlock_failures`` -- the last day of sends and
  the last minutes of wrong passwords, for the limits below.

Rules: one active link per kind per Journey (issuing a new one revokes the
old); a push link sends at most once per ``min_push_interval_s`` and
``daily_push_limit`` times per rolling 24 h -- past either, a clear 429 with
when it is possible again, never a silent drop; positions older than
``retention_days`` are deleted, and the owner can delete them all at any
time. A password unlocks a link once per device: the page gets a signed,
expiring grant scoped to that one link (an HttpOnly cookie, set by the HTTP
layer), which dies with the link.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import math
import re
import secrets
import sqlite3
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional, Sequence

KINDS = ("view", "push")
TOKEN_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}")
JOURNEY_ID_MAX = 200
TITLE_MAX = 256
MAX_ROUTE_POINTS = 60000
PASSWORD_MIN, PASSWORD_MAX = 4, 128
PBKDF2_ITERATIONS = 600_000
POSITIONS_ON_VIEW = 200
ADMIN_POSITIONS = 50
# The own server's satellite layer (EOX Sentinel-2 cloudless 2016, CC BY 4.0)
# as the view page's basemap -- EOX's required attribution.
SATELLITE_ATTRIBUTION = (
    '<a href="https://s2maps.eu" target="_blank">Sentinel-2 cloudless – s2maps.eu</a> by EOX IT Services GmbH '
    "(Contains modified Copernicus Sentinel data 2016)"
)


@dataclass(frozen=True)
class LiveTrackingSettings:
    min_push_interval_s: int = 30
    daily_push_limit: int = 500
    retention_days: int = 90
    view_poll_interval_s: int = 30
    unlock_attempts: int = 10
    unlock_window_s: int = 15 * 60
    grant_ttl_s: int = 30 * 24 * 3600


class LiveTrackingError(Exception):
    """A refusal the page shows as is: ``message`` is for the person,
    ``code`` for the code, ``extra`` carries e.g. ``retry_after_s``."""

    def __init__(self, status: int, code: str, message: str, **extra: Any):
        super().__init__(message)
        self.status, self.code, self.message, self.extra = status, code, message, extra

    def document(self) -> dict[str, Any]:
        return {"error": self.code, "message": self.message, **self.extra}


NOT_FOUND = (404, "LINK_NOT_FOUND", "Ten link jest nieważny: został unieważniony albo jest błędny.")


def iso(epoch: float) -> str:
    return datetime.fromtimestamp(epoch, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def token_hash(token: str) -> str:
    return hashlib.sha256(token.encode("ascii")).hexdigest()


def _b64(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")


def hash_password(password: str) -> str:
    """PBKDF2-HMAC-SHA256 (OWASP's 600 000 iterations): in every Python
    build, unlike scrypt (absent where Python links LibreSSL)."""
    salt = secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS)
    return f"pbkdf2_sha256${PBKDF2_ITERATIONS}${_b64(salt)}${_b64(digest)}"


def check_password(password: str, stored: str) -> bool:
    try:
        scheme, iterations, salt, digest = stored.split("$")
        if scheme != "pbkdf2_sha256":
            return False
        candidate = hashlib.pbkdf2_hmac(
            "sha256", password.encode("utf-8"), _unb64(salt), int(iterations)
        )
        return hmac.compare_digest(candidate, _unb64(digest))
    except (ValueError, TypeError):
        return False


def _unb64(text: str) -> bytes:
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


SCHEMA = """
CREATE TABLE IF NOT EXISTS live_schema_migrations (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS live_journeys (
    journey_id TEXT PRIMARY KEY,
    title TEXT NOT NULL,
    route_revision_id TEXT,
    route_json TEXT NOT NULL,
    published_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS live_links (
    link_id TEXT PRIMARY KEY,
    journey_id TEXT NOT NULL,
    kind TEXT NOT NULL CHECK (kind IN ('view', 'push')),
    token_hash TEXT NOT NULL UNIQUE,
    password_hash TEXT,
    created_at REAL NOT NULL,
    revoked_at REAL
);
CREATE INDEX IF NOT EXISTS live_links_by_journey ON live_links (journey_id, kind);
CREATE TABLE IF NOT EXISTS live_positions (
    position_id INTEGER PRIMARY KEY AUTOINCREMENT,
    journey_id TEXT NOT NULL,
    link_id TEXT NOT NULL,
    latitude REAL NOT NULL,
    longitude REAL NOT NULL,
    accuracy_m REAL,
    altitude_m REAL,
    device_time REAL,
    received_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS live_positions_by_journey ON live_positions (journey_id, received_at);
CREATE TABLE IF NOT EXISTS live_push_log (link_id TEXT NOT NULL, at REAL NOT NULL);
CREATE INDEX IF NOT EXISTS live_push_log_by_link ON live_push_log (link_id, at);
CREATE TABLE IF NOT EXISTS live_unlock_failures (link_id TEXT NOT NULL, at REAL NOT NULL);
"""


class LiveTrackingService:
    def __init__(
        self,
        database: Path | str,
        grant_secret: bytes,
        *,
        settings: Optional[LiveTrackingSettings] = None,
        now: Callable[[], float] = time.time,
    ):
        if len(grant_secret) < 32:
            raise ValueError("the grant secret must be at least 32 bytes")
        self.database = str(database)
        self.settings = settings or LiveTrackingSettings()
        self._grant_secret = grant_secret
        self._now = now
        self._lock = threading.Lock()
        if self.database != ":memory:":
            Path(self.database).parent.mkdir(parents=True, exist_ok=True)
        self._memory = (
            sqlite3.connect(":memory:", check_same_thread=False)
            if self.database == ":memory:"
            else None
        )
        with self._connect() as connection:
            connection.executescript(SCHEMA)
            connection.execute(
                "INSERT OR IGNORE INTO live_schema_migrations VALUES (1, ?)", (iso(self._now()),)
            )

    # --- storage ----------------------------------------------------------
    def _connect(self) -> sqlite3.Connection:
        if self._memory is not None:
            self._memory.row_factory = sqlite3.Row
            return self._memory
        connection = sqlite3.connect(self.database, timeout=10)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA journal_mode=WAL")
        return connection

    def _transaction(self, work: Callable[[sqlite3.Connection], Any]) -> Any:
        with self._lock:
            connection = self._connect()
            try:
                with connection:
                    return work(connection)
            finally:
                if self._memory is None:
                    connection.close()

    def _purge(self, connection: sqlite3.Connection, now: float) -> None:
        connection.execute(
            "DELETE FROM live_positions WHERE received_at < ?",
            (now - self.settings.retention_days * 86400,),
        )
        connection.execute("DELETE FROM live_push_log WHERE at < ?", (now - 86400,))
        connection.execute(
            "DELETE FROM live_unlock_failures WHERE at < ?", (now - self.settings.unlock_window_s,)
        )

    # --- owner (admin) ----------------------------------------------------
    def publish(
        self,
        journey_id: str,
        title: str,
        route_revision_id: Optional[str],
        latlngs: Sequence[Sequence[float]],
    ) -> dict[str, Any]:
        """What the view page may show: the title and the route's points."""
        journey_id = _journey_id(journey_id)
        title = str(title or "").strip()[:TITLE_MAX]
        if not title:
            raise LiveTrackingError(400, "TITLE_REQUIRED", "Wyprawa nie ma nazwy.")
        if not isinstance(latlngs, (list, tuple)) or not 2 <= len(latlngs) <= MAX_ROUTE_POINTS:
            raise LiveTrackingError(400, "ROUTE_REQUIRED", "Wyprawa nie ma trasy do pokazania.")
        route = [[_coordinate(p[0], 90), _coordinate(p[1], 180)] for p in latlngs]
        now = self._now()

        def work(connection):
            connection.execute(
                """INSERT INTO live_journeys VALUES (?, ?, ?, ?, ?)
                   ON CONFLICT (journey_id) DO UPDATE SET title = excluded.title,
                   route_revision_id = excluded.route_revision_id, route_json = excluded.route_json,
                   published_at = excluded.published_at""",
                (
                    journey_id,
                    title,
                    route_revision_id,
                    json.dumps(route, separators=(",", ":")),
                    now,
                ),
            )

        self._transaction(work)
        return {
            "journey_id": journey_id,
            "route_revision_id": route_revision_id,
            "published_at": iso(now),
        }

    def issue_link(
        self, journey_id: str, kind: str, password: Optional[str] = None
    ) -> dict[str, Any]:
        """A new link of this kind; the previous one of the kind stops
        working. The token is returned here and nowhere else, ever."""
        journey_id, kind = _journey_id(journey_id), _kind(kind)
        password_hash = None
        if password:
            if not PASSWORD_MIN <= len(password) <= PASSWORD_MAX:
                raise LiveTrackingError(
                    400,
                    "PASSWORD_LENGTH",
                    f"Hasło musi mieć od {PASSWORD_MIN} do {PASSWORD_MAX} znaków.",
                )
            password_hash = hash_password(password)
        token, link_id, now = secrets.token_urlsafe(32), secrets.token_hex(12), self._now()

        def work(connection):
            if (
                connection.execute(
                    "SELECT 1 FROM live_journeys WHERE journey_id = ?", (journey_id,)
                ).fetchone()
                is None
            ):
                raise LiveTrackingError(
                    409, "NOT_PUBLISHED", "Trasa tej wyprawy nie została jeszcze przesłana."
                )
            connection.execute(
                "UPDATE live_links SET revoked_at = ? WHERE journey_id = ? AND kind = ? AND revoked_at IS NULL",
                (now, journey_id, kind),
            )
            connection.execute(
                "INSERT INTO live_links VALUES (?, ?, ?, ?, ?, ?, NULL)",
                (link_id, journey_id, kind, token_hash(token), password_hash, now),
            )

        self._transaction(work)
        return {
            "kind": kind,
            "token": token,
            "link_id": link_id,
            "created_at": iso(now),
            "has_password": bool(password_hash),
        }

    def revoke_link(self, journey_id: str, kind: str) -> bool:
        journey_id, kind = _journey_id(journey_id), _kind(kind)
        now = self._now()
        return self._transaction(
            lambda c: (
                c.execute(
                    "UPDATE live_links SET revoked_at = ? WHERE journey_id = ? AND kind = ? AND revoked_at IS NULL",
                    (now, journey_id, kind),
                ).rowcount
                > 0
            )
        )

    def status(self, journey_id: str) -> dict[str, Any]:
        journey_id = _journey_id(journey_id)
        now = self._now()

        def work(connection):
            self._purge(connection, now)
            journey = connection.execute(
                "SELECT * FROM live_journeys WHERE journey_id = ?", (journey_id,)
            ).fetchone()
            links = {}
            for kind in KINDS:
                row = connection.execute(
                    "SELECT * FROM live_links WHERE journey_id = ? AND kind = ? AND revoked_at IS NULL",
                    (journey_id, kind),
                ).fetchone()
                links[kind] = (
                    {
                        "link_id": row["link_id"],
                        "created_at": iso(row["created_at"]),
                        "has_password": bool(row["password_hash"]),
                    }
                    if row
                    else None
                )
            rows = connection.execute(
                "SELECT * FROM live_positions WHERE journey_id = ? ORDER BY received_at DESC, position_id DESC LIMIT ?",
                (journey_id, ADMIN_POSITIONS),
            ).fetchall()
            count = connection.execute(
                "SELECT COUNT(*) FROM live_positions WHERE journey_id = ?", (journey_id,)
            ).fetchone()[0]
            return {
                "journey_id": journey_id,
                "published": journey is not None,
                "route_revision_id": journey["route_revision_id"] if journey else None,
                "links": links,
                "positions": [_position(row) for row in rows],
                "position_count": count,
                "rules": self.rules(),
            }

        return self._transaction(work)

    def rules(self) -> dict[str, int]:
        s = self.settings
        return {
            "min_push_interval_s": s.min_push_interval_s,
            "daily_push_limit": s.daily_push_limit,
            "retention_days": s.retention_days,
            "view_poll_interval_s": s.view_poll_interval_s,
        }

    def delete_positions(self, journey_id: str) -> int:
        journey_id = _journey_id(journey_id)
        return self._transaction(
            lambda c: (
                c.execute("DELETE FROM live_positions WHERE journey_id = ?", (journey_id,)).rowcount
            )
        )

    def delete_journey(self, journey_id: str) -> None:
        """Everything about the Journey: links, positions, the route."""
        journey_id = _journey_id(journey_id)

        def work(connection):
            link_ids = [
                r[0]
                for r in connection.execute(
                    "SELECT link_id FROM live_links WHERE journey_id = ?", (journey_id,)
                )
            ]
            for link_id in link_ids:
                connection.execute("DELETE FROM live_push_log WHERE link_id = ?", (link_id,))
                connection.execute("DELETE FROM live_unlock_failures WHERE link_id = ?", (link_id,))
            for table in ("live_positions", "live_links", "live_journeys"):
                connection.execute(f"DELETE FROM {table} WHERE journey_id = ?", (journey_id,))

        self._transaction(work)

    # --- public (by token) ------------------------------------------------
    def _link(self, connection: sqlite3.Connection, kind: str, token: str) -> sqlite3.Row:
        if not isinstance(token, str) or not TOKEN_PATTERN.fullmatch(token):
            raise LiveTrackingError(*NOT_FOUND)
        row = connection.execute(
            """SELECT l.*, j.title FROM live_links l JOIN live_journeys j ON j.journey_id = l.journey_id
               WHERE l.token_hash = ? AND l.kind = ? AND l.revoked_at IS NULL""",
            (token_hash(token), kind),
        ).fetchone()
        if row is None:
            raise LiveTrackingError(*NOT_FOUND)
        return row

    def _grant_for(self, link_id: str, expires: int) -> str:
        signature = hmac.new(
            self._grant_secret, f"{link_id}|{expires}".encode(), hashlib.sha256
        ).digest()
        return f"{expires}.{_b64(signature)}"

    def _authorize(self, link: sqlite3.Row, grant: Optional[str], now: float) -> None:
        if not link["password_hash"]:
            return
        try:
            expires = int(str(grant).split(".", 1)[0])
        except ValueError:
            expires = 0
        if (
            not grant
            or expires < now
            or not hmac.compare_digest(grant, self._grant_for(link["link_id"], expires))
        ):
            raise LiveTrackingError(401, "PASSWORD_REQUIRED", "Ten link jest chroniony hasłem.")

    def unlock(self, kind: str, token: str, password: str) -> dict[str, Any]:
        """The password once; returns the grant the page keeps (cookie)."""
        kind, now = _kind(kind), self._now()

        def work(connection):
            self._purge(connection, now)
            link = self._link(connection, kind, token)
            if not link["password_hash"]:
                return {"link_id": link["link_id"], "grant": None, "max_age": 0}
            failures = connection.execute(
                "SELECT COUNT(*), MIN(at) FROM live_unlock_failures WHERE link_id = ?",
                (link["link_id"],),
            ).fetchone()
            if failures[0] >= self.settings.unlock_attempts:
                retry = int(math.ceil(failures[1] + self.settings.unlock_window_s - now))
                raise LiveTrackingError(
                    429,
                    "TOO_MANY_ATTEMPTS",
                    f"Za dużo błędnych prób hasła. Spróbuj ponownie za {_minutes(retry)}.",
                    retry_after_s=max(1, retry),
                )
            if not check_password(str(password or ""), link["password_hash"]):
                connection.execute(
                    "INSERT INTO live_unlock_failures VALUES (?, ?)", (link["link_id"], now)
                )
                return LiveTrackingError(401, "WRONG_PASSWORD", "Nieprawidłowe hasło.")
            expires = int(now) + self.settings.grant_ttl_s
            return {
                "link_id": link["link_id"],
                "grant": self._grant_for(link["link_id"], expires),
                "max_age": self.settings.grant_ttl_s,
            }

        result = self._transaction(work)
        if isinstance(result, LiveTrackingError):  # the failure is recorded first, then refused
            raise result
        return result

    def link_id_for(self, kind: str, token: str) -> str:
        return self._transaction(lambda c: self._link(c, _kind(kind), token)["link_id"])

    def view_document(self, token: str, grant: Optional[str] = None) -> dict[str, Any]:
        now = self._now()

        def work(connection):
            self._purge(connection, now)
            link = self._link(connection, "view", token)
            self._authorize(link, grant, now)
            journey = connection.execute(
                "SELECT route_json FROM live_journeys WHERE journey_id = ?", (link["journey_id"],)
            ).fetchone()
            rows = connection.execute(
                "SELECT * FROM live_positions WHERE journey_id = ? ORDER BY received_at DESC, position_id DESC LIMIT ?",
                (link["journey_id"], POSITIONS_ON_VIEW),
            ).fetchall()
            positions = [_position(row, public=True) for row in reversed(rows)]
            return {
                "title": link["title"],
                "route": json.loads(journey["route_json"]),
                "positions": positions,
                "latest": positions[-1] if positions else None,
                "poll_interval_s": self.settings.view_poll_interval_s,
                "server_time": iso(now),
            }

        return self._transaction(work)

    def push_status(self, token: str, grant: Optional[str] = None) -> dict[str, Any]:
        now = self._now()

        def work(connection):
            self._purge(connection, now)
            link = self._link(connection, "push", token)
            self._authorize(link, grant, now)
            return self._push_state(connection, link, now)

        return self._transaction(work)

    def _push_state(
        self, connection: sqlite3.Connection, link: sqlite3.Row, now: float
    ) -> dict[str, Any]:
        sends = [
            r[0]
            for r in connection.execute(
                "SELECT at FROM live_push_log WHERE link_id = ? AND at >= ? ORDER BY at",
                (link["link_id"], now - 86400),
            )
        ]
        last = connection.execute(
            "SELECT * FROM live_positions WHERE link_id = ? ORDER BY received_at DESC, position_id DESC LIMIT 1",
            (link["link_id"],),
        ).fetchone()
        next_allowed = now
        if sends:
            next_allowed = max(next_allowed, sends[-1] + self.settings.min_push_interval_s)
        if len(sends) >= self.settings.daily_push_limit:
            next_allowed = max(
                next_allowed, sends[len(sends) - self.settings.daily_push_limit] + 86400
            )
        return {
            "title": link["title"],
            "last_position": _position(last, public=True) if last else None,
            "sent_last_24h": len(sends),
            "next_allowed_at": iso(next_allowed),
            "next_allowed_in_s": int(math.ceil(next_allowed - now)),
            "rules": self.rules(),
            "server_time": iso(now),
        }

    def push(self, token: str, payload: Any, grant: Optional[str] = None) -> dict[str, Any]:
        latitude, longitude, accuracy, altitude, device_time = _push_payload(payload)
        now = self._now()

        def work(connection):
            self._purge(connection, now)
            link = self._link(connection, "push", token)
            self._authorize(link, grant, now)
            state = self._push_state(connection, link, now)
            wait = state["next_allowed_in_s"]
            if wait > 0:
                if state["sent_last_24h"] >= self.settings.daily_push_limit:
                    raise LiveTrackingError(
                        429,
                        "DAILY_LIMIT",
                        f"Dzienny limit {self.settings.daily_push_limit} wysłań z tego linku jest wyczerpany. "
                        f"Następne wysłanie będzie możliwe za {_minutes(wait)}.",
                        retry_after_s=wait,
                        next_allowed_at=state["next_allowed_at"],
                    )
                raise LiveTrackingError(
                    429,
                    "TOO_SOON",
                    f"Za wcześnie: pozycję można wysyłać nie częściej niż co {self.settings.min_push_interval_s} s. "
                    f"Spróbuj ponownie za {wait} s.",
                    retry_after_s=wait,
                    next_allowed_at=state["next_allowed_at"],
                )
            connection.execute("INSERT INTO live_push_log VALUES (?, ?)", (link["link_id"], now))
            connection.execute(
                """INSERT INTO live_positions (journey_id, link_id, latitude, longitude, accuracy_m,
                   altitude_m, device_time, received_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    link["journey_id"],
                    link["link_id"],
                    latitude,
                    longitude,
                    accuracy,
                    altitude,
                    device_time,
                    now,
                ),
            )
            return self._push_state(connection, link, now)

        return self._transaction(work)


def _journey_id(value: Any) -> str:
    text = str(value or "").strip()
    if not text or len(text) > JOURNEY_ID_MAX:
        raise LiveTrackingError(400, "JOURNEY_ID", "Nieprawidłowy identyfikator wyprawy.")
    return text


def _kind(value: Any) -> str:
    if value not in KINDS:
        raise LiveTrackingError(404, "KIND", "Nieznany rodzaj linku.")
    return value


def _coordinate(value: Any, limit: float) -> float:
    number = float(value)
    if not math.isfinite(number) or abs(number) > limit:
        raise LiveTrackingError(400, "COORDINATE", "Współrzędne poza zakresem.")
    return round(number, 6)


def _optional_number(value: Any, low: float, high: float) -> Optional[float]:
    if value is None:
        return None
    number = float(value)
    if not math.isfinite(number) or not low <= number <= high:
        raise ValueError
    return number


def _push_payload(payload: Any):
    try:
        if not isinstance(payload, dict):
            raise ValueError
        latitude = _coordinate(payload["latitude"], 90)
        longitude = _coordinate(payload["longitude"], 180)
        accuracy = _optional_number(payload.get("accuracy_m"), 0, 1_000_000)
        altitude = _optional_number(payload.get("altitude_m"), -1000, 10000)
        device_time = _optional_number(payload.get("device_time_ms"), 0, 4_102_444_800_000)
    except (KeyError, TypeError, ValueError, LiveTrackingError):
        raise LiveTrackingError(
            400, "POSITION", "Nieprawidłowa pozycja: brak albo błędne współrzędne."
        ) from None
    return (
        latitude,
        longitude,
        accuracy,
        altitude,
        (device_time / 1000 if device_time is not None else None),
    )


def _position(row: sqlite3.Row, public: bool = False) -> dict[str, Any]:
    document = {
        "latitude": row["latitude"],
        "longitude": row["longitude"],
        "accuracy_m": row["accuracy_m"],
        "altitude_m": row["altitude_m"],
        "received_at": iso(row["received_at"]),
        "device_time": iso(row["device_time"]) if row["device_time"] is not None else None,
    }
    if not public:
        document["position_id"] = row["position_id"]
    return document


def _minutes(seconds: int) -> str:
    if seconds < 90:
        return f"{max(1, seconds)} s"
    minutes = int(math.ceil(seconds / 60))
    if minutes < 90:
        return f"{minutes} min"
    hours, rest = divmod(minutes, 60)
    return f"{hours} h {rest} min" if rest else f"{hours} h"