"""Crash-safe SQLite outbox for telemetry and camera events."""

from __future__ import annotations

import json
import math
import sqlite3
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence

OUTBOX_STATUS_SCHEMA_VERSION = 1
_MAX_PUBLIC_COUNT = 1_000_000_000
_MAX_PUBLIC_BYTES = 10_000_000_000_000
_MAX_PUBLIC_AGE_SECONDS = 315_576_000.0
_MAX_PUBLIC_TIMESTAMP = 100_000_000_000.0


class OutboxCapacityError(RuntimeError):
    """Raised instead of silently dropping an event when the queue is full."""


def _positive_quota(value: Any, name: str, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError("{} must be a positive integer".format(name))
    if value > maximum:
        raise ValueError("{} exceeds its supported maximum".format(name))
    return value


def _bounded_count(value: Any, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return 0
    try:
        parsed = int(value)
    except (OverflowError, TypeError, ValueError):
        return 0
    return min(max(parsed, 0), maximum)


def _bounded_optional_number(value: Any, maximum: float) -> Optional[float]:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    parsed = float(value)
    if not math.isfinite(parsed) or parsed < 0.0:
        return None
    return min(parsed, maximum)


def _clock_time(clock: Any) -> float:
    value = clock()
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError("outbox clock returned an invalid value")
    parsed = float(value)
    if not math.isfinite(parsed) or not 0.0 <= parsed <= _MAX_PUBLIC_TIMESTAMP:
        raise ValueError("outbox clock returned an invalid value")
    return parsed


@dataclass(frozen=True)
class OutboxEvent:
    """One leased outbox event."""

    event_id: str
    kind: str
    observed_at: str
    payload: Mapping[str, Any]
    blob: bytes
    attempts: int


@dataclass(frozen=True)
class OutboxReceipt:
    """Durable proof that one event was acknowledged by the remote service."""

    event_id: str
    acknowledged_at: float
    status_code: int
    dataset_asset_id: Optional[int]


class DurableOutbox:
    """Persist events until a remote service acknowledges them.

    Each operation opens its own SQLite connection. This keeps the object safe
    to use from camera, robot and uploader threads without sharing cursors.
    """

    def __init__(
        self,
        path: Path,
        *,
        max_events: int = 10_000,
        max_bytes: int = 512 * 1024 * 1024,
        max_dead_letters: int = 10_000,
        max_dead_letter_bytes: int = 512 * 1024 * 1024,
        max_receipts: int = 100_000,
        clock=time.time,
    ) -> None:
        self.path = Path(path)
        self.max_events = _positive_quota(max_events, "max_events", _MAX_PUBLIC_COUNT)
        self.max_bytes = _positive_quota(max_bytes, "max_bytes", _MAX_PUBLIC_BYTES)
        self.max_dead_letters = _positive_quota(
            max_dead_letters, "max_dead_letters", _MAX_PUBLIC_COUNT
        )
        self.max_dead_letter_bytes = _positive_quota(
            max_dead_letter_bytes, "max_dead_letter_bytes", _MAX_PUBLIC_BYTES
        )
        self.max_receipts = _positive_quota(
            max_receipts, "max_receipts", _MAX_PUBLIC_COUNT
        )
        self.clock = clock
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(
            str(self.path),
            timeout=15.0,
            isolation_level=None,
        )
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA busy_timeout = 15000")
        connection.execute("PRAGMA journal_mode = WAL")
        connection.execute("PRAGMA synchronous = FULL")
        return connection

    def _initialize(self) -> None:
        with self._connect() as connection:
            connection.executescript("""
                CREATE TABLE IF NOT EXISTS outbox_events (
                    event_id TEXT PRIMARY KEY,
                    kind TEXT NOT NULL,
                    observed_at TEXT NOT NULL,
                    payload_json TEXT NOT NULL,
                    blob BLOB NOT NULL DEFAULT X'',
                    attempts INTEGER NOT NULL DEFAULT 0,
                    available_at REAL NOT NULL,
                    lease_until REAL,
                    lease_started_at REAL,
                    created_at REAL NOT NULL,
                    last_error TEXT NOT NULL DEFAULT ''
                );
                CREATE INDEX IF NOT EXISTS outbox_due_idx
                    ON outbox_events (available_at, lease_until, created_at);
                CREATE TABLE IF NOT EXISTS outbox_dead_letters (
                    event_id TEXT PRIMARY KEY,
                    kind TEXT NOT NULL,
                    observed_at TEXT NOT NULL,
                    payload_json TEXT NOT NULL,
                    blob BLOB NOT NULL DEFAULT X'',
                    attempts INTEGER NOT NULL,
                    created_at REAL NOT NULL,
                    failed_at REAL NOT NULL,
                    last_error TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS outbox_receipts (
                    event_id TEXT PRIMARY KEY,
                    acknowledged_at REAL NOT NULL,
                    status_code INTEGER NOT NULL,
                    dataset_asset_id INTEGER
                );
                CREATE TABLE IF NOT EXISTS outbox_delivery_status (
                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
                    last_acknowledged_at REAL
                );
                """)
            connection.execute("BEGIN IMMEDIATE")
            try:
                columns = {
                    str(row["name"])
                    for row in connection.execute(
                        "PRAGMA table_info(outbox_events)"
                    ).fetchall()
                }
                if "lease_started_at" not in columns:
                    connection.execute(
                        "ALTER TABLE outbox_events " "ADD COLUMN lease_started_at REAL"
                    )
                connection.commit()
            except Exception:
                connection.rollback()
                raise

    def enqueue(
        self,
        event_id: str,
        kind: str,
        observed_at: str,
        payload: Mapping[str, Any],
        *,
        blob: bytes = b"",
    ) -> bool:
        """Atomically add an event.

        Returns ``False`` when the same event id is already queued, making
        producer retries idempotent.
        """
        serialized = json.dumps(
            payload,
            ensure_ascii=False,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        )
        blob = bytes(blob)
        now = _clock_time(self.clock)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            duplicate = connection.execute(
                """
                SELECT 1 FROM outbox_events WHERE event_id = ?
                UNION ALL
                SELECT 1 FROM outbox_dead_letters WHERE event_id = ?
                UNION ALL
                SELECT 1 FROM outbox_receipts WHERE event_id = ?
                LIMIT 1
                """,
                (event_id, event_id, event_id),
            ).fetchone()
            if duplicate is not None:
                connection.commit()
                return False

            usage = connection.execute("""
                SELECT COUNT(*) AS count,
                       COALESCE(
                           SUM(LENGTH(payload_json) + LENGTH(blob)), 0
                       ) AS bytes
                FROM outbox_events
                """).fetchone()
            projected_count = int(usage["count"]) + 1
            projected_bytes = int(usage["bytes"]) + len(serialized.encode("utf-8"))
            projected_bytes += len(blob)
            if projected_count > self.max_events or projected_bytes > self.max_bytes:
                connection.rollback()
                raise OutboxCapacityError(
                    "Durable telemetry outbox is full; event was not discarded"
                )

            connection.execute(
                """
                INSERT INTO outbox_events (
                    event_id, kind, observed_at, payload_json, blob,
                    available_at, created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    event_id,
                    kind,
                    observed_at,
                    serialized,
                    sqlite3.Binary(blob),
                    now,
                    now,
                ),
            )
            connection.commit()
            return True

    def claim(self, limit: int = 25, lease_seconds: float = 60.0) -> List[OutboxEvent]:
        """Lease due events so concurrent upload workers cannot double-send."""
        now = _clock_time(self.clock)
        if (
            isinstance(lease_seconds, bool)
            or not isinstance(lease_seconds, (int, float))
            or not math.isfinite(float(lease_seconds))
            or float(lease_seconds) <= 0.0
        ):
            raise ValueError("lease_seconds must be finite and positive")
        lease_until = now + float(lease_seconds)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            rows = connection.execute(
                """
                SELECT event_id, kind, observed_at, payload_json, blob,
                       attempts
                FROM outbox_events
                WHERE available_at <= ?
                  AND (lease_until IS NULL OR lease_until <= ?)
                ORDER BY created_at, event_id
                LIMIT ?
                """,
                (now, now, max(1, int(limit))),
            ).fetchall()
            ids = [row["event_id"] for row in rows]
            if ids:
                placeholders = ",".join("?" for _ in ids)
                connection.execute(
                    "UPDATE outbox_events SET lease_until = ?, "
                    "lease_started_at = ? "
                    "WHERE event_id IN ({})".format(placeholders),
                    (lease_until, now, *ids),
                )
            connection.commit()
        return [
            OutboxEvent(
                event_id=str(row["event_id"]),
                kind=str(row["kind"]),
                observed_at=str(row["observed_at"]),
                payload=json.loads(row["payload_json"]),
                blob=bytes(row["blob"]),
                attempts=int(row["attempts"]),
            )
            for row in rows
        ]

    def acknowledge(
        self,
        event_id: str,
        *,
        status_code: int = 0,
        dataset_asset_id: Optional[int] = None,
        retain_receipt: bool = False,
    ) -> None:
        """Remove an event, optionally replacing it with a durable receipt.

        Retained receipts are acknowledgement tombstones intended for bounded,
        deterministic capture-frame ids. High-rate random telemetry must use
        the default so its acknowledgements do not grow the database forever.
        """

        status_code = int(status_code)
        if not isinstance(retain_receipt, bool):
            raise ValueError("retain_receipt must be a boolean")
        if dataset_asset_id is not None:
            if isinstance(dataset_asset_id, bool):
                raise ValueError("dataset_asset_id must be an integer")
            dataset_asset_id = int(dataset_asset_id)
        acknowledged_at = _clock_time(self.clock)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            pending = connection.execute(
                "SELECT 1 FROM outbox_events WHERE event_id = ?",
                (event_id,),
            ).fetchone()
            if pending is None:
                connection.commit()
                return
            if retain_receipt:
                receipt_count = connection.execute(
                    "SELECT COUNT(*) AS count FROM outbox_receipts"
                ).fetchone()
                existing_receipt = connection.execute(
                    "SELECT 1 FROM outbox_receipts WHERE event_id = ?",
                    (event_id,),
                ).fetchone()
                projected_receipts = int(receipt_count["count"]) + (
                    0 if existing_receipt is not None else 1
                )
                if projected_receipts > self.max_receipts:
                    connection.rollback()
                    raise OutboxCapacityError(
                        "Durable acknowledgement receipt quota is full; "
                        "event remains queued"
                    )
                connection.execute(
                    """
                    INSERT INTO outbox_receipts (
                        event_id, acknowledged_at, status_code,
                        dataset_asset_id
                    )
                    SELECT event_id, ?, ?, ?
                    FROM outbox_events
                    WHERE event_id = ?
                    """,
                    (
                        acknowledged_at,
                        status_code,
                        dataset_asset_id,
                        event_id,
                    ),
                )
            connection.execute(
                """
                INSERT OR REPLACE INTO outbox_delivery_status (
                    singleton, last_acknowledged_at
                ) VALUES (1, ?)
                """,
                (acknowledged_at,),
            )
            connection.execute(
                "DELETE FROM outbox_events WHERE event_id = ?",
                (event_id,),
            )
            connection.commit()

    def retry(self, event_id: str, error: str, delay_seconds: float) -> None:
        """Release an event for a later retry."""
        if (
            isinstance(delay_seconds, bool)
            or not isinstance(delay_seconds, (int, float))
            or not math.isfinite(float(delay_seconds))
        ):
            raise ValueError("delay_seconds must be finite")
        next_attempt = _clock_time(self.clock) + max(0.0, float(delay_seconds))
        next_attempt = min(next_attempt, _MAX_PUBLIC_TIMESTAMP)
        with self._connect() as connection:
            connection.execute(
                """
                UPDATE outbox_events
                SET attempts = attempts + 1,
                    available_at = ?,
                    lease_until = NULL,
                    lease_started_at = NULL,
                    last_error = ?
                WHERE event_id = ?
                """,
                (next_attempt, str(error)[:1000], event_id),
            )

    def dead_letter(self, event_id: str, error: str) -> None:
        """Durably quarantine a non-retryable event for inspection."""
        now = _clock_time(self.clock)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            pending = connection.execute(
                """
                SELECT LENGTH(payload_json) + LENGTH(blob) AS bytes
                FROM outbox_events WHERE event_id = ?
                """,
                (event_id,),
            ).fetchone()
            if pending is None:
                connection.commit()
                return
            usage = connection.execute("""
                SELECT COUNT(*) AS count,
                       COALESCE(
                           SUM(LENGTH(payload_json) + LENGTH(blob)), 0
                       ) AS bytes
                FROM outbox_dead_letters
                """).fetchone()
            projected_count = int(usage["count"]) + 1
            projected_bytes = int(usage["bytes"]) + int(pending["bytes"])
            if (
                projected_count > self.max_dead_letters
                or projected_bytes > self.max_dead_letter_bytes
            ):
                connection.rollback()
                raise OutboxCapacityError(
                    "Durable dead-letter quota is full; event remains queued"
                )
            connection.execute(
                """
                INSERT INTO outbox_dead_letters (
                    event_id, kind, observed_at, payload_json, blob,
                    attempts, created_at, failed_at, last_error
                )
                SELECT event_id, kind, observed_at, payload_json, blob,
                       attempts + 1, created_at, ?, ?
                FROM outbox_events
                WHERE event_id = ?
                """,
                (now, str(error)[:1000], event_id),
            )
            connection.execute(
                "DELETE FROM outbox_events WHERE event_id = ?",
                (event_id,),
            )
            connection.commit()

    def stats(self) -> Mapping[str, Any]:
        """Return one atomic, bounded queue/retention status snapshot."""

        now = _clock_time(self.clock)
        with self._connect() as connection:
            connection.execute("BEGIN")
            row = connection.execute(
                """
                SELECT
                    (SELECT COUNT(*) FROM outbox_events) AS pending_count,
                    (SELECT COALESCE(
                        SUM(LENGTH(payload_json) + LENGTH(blob)), 0
                     ) FROM outbox_events) AS queue_bytes,
                    (SELECT COUNT(*) FROM outbox_events
                     WHERE lease_until IS NOT NULL AND lease_until > ?)
                        AS leased_count,
                    (SELECT MIN(lease_started_at) FROM outbox_events
                     WHERE lease_until IS NOT NULL AND lease_until > ?)
                        AS oldest_lease_started_at,
                    (SELECT COALESCE(SUM(attempts), 0)
                     FROM outbox_events) AS retry_count,
                    (SELECT COUNT(*) FROM outbox_dead_letters)
                        AS dead_letter_count,
                    (SELECT COALESCE(
                        SUM(LENGTH(payload_json) + LENGTH(blob)), 0
                     ) FROM outbox_dead_letters) AS dead_letter_bytes,
                    (SELECT COUNT(*) FROM outbox_receipts) AS receipt_count,
                    (SELECT last_acknowledged_at
                     FROM outbox_delivery_status WHERE singleton = 1)
                        AS last_acknowledged_at
                """,
                (now, now),
            ).fetchone()
            connection.commit()
        if row is None:
            raise RuntimeError("outbox statistics are unavailable")

        pending_count = _bounded_count(row["pending_count"], _MAX_PUBLIC_COUNT)
        queue_bytes = _bounded_count(row["queue_bytes"], _MAX_PUBLIC_BYTES)
        leased_count = _bounded_count(row["leased_count"], _MAX_PUBLIC_COUNT)
        retry_count = _bounded_count(row["retry_count"], _MAX_PUBLIC_COUNT)
        dead_letter_count = _bounded_count(row["dead_letter_count"], _MAX_PUBLIC_COUNT)
        dead_letter_bytes = _bounded_count(row["dead_letter_bytes"], _MAX_PUBLIC_BYTES)
        receipt_count = _bounded_count(row["receipt_count"], _MAX_PUBLIC_COUNT)
        oldest_started = _bounded_optional_number(
            row["oldest_lease_started_at"], _MAX_PUBLIC_TIMESTAMP
        )
        last_ack = _bounded_optional_number(
            row["last_acknowledged_at"], _MAX_PUBLIC_TIMESTAMP
        )
        oldest_age = None
        if oldest_started is not None:
            oldest_age = min(max(now - oldest_started, 0.0), _MAX_PUBLIC_AGE_SECONDS)
        last_ack_age = None
        if last_ack is not None:
            last_ack_age = min(max(now - last_ack, 0.0), _MAX_PUBLIC_AGE_SECONDS)
        return {
            "schema_version": OUTBOX_STATUS_SCHEMA_VERSION,
            "pending_count": pending_count,
            "queue_bytes": queue_bytes,
            "leased_count": leased_count,
            "oldest_lease_age_seconds": oldest_age,
            "retry_count": retry_count,
            "dead_letter_count": dead_letter_count,
            "dead_letter_bytes": dead_letter_bytes,
            "receipt_count": receipt_count,
            "last_acknowledged_at": last_ack,
            "last_acknowledgement_age_seconds": last_ack_age,
            "capacity_ok": bool(
                pending_count < self.max_events
                and queue_bytes < self.max_bytes
                and dead_letter_count < self.max_dead_letters
                and dead_letter_bytes < self.max_dead_letter_bytes
                and receipt_count < self.max_receipts
            ),
        }

    def pending_count(self) -> int:
        """Return the number of events awaiting delivery."""
        with self._connect() as connection:
            row = connection.execute(
                "SELECT COUNT(*) AS count FROM outbox_events"
            ).fetchone()
        return int(row["count"])

    def dead_letter_count(self) -> int:
        """Return the number of quarantined events."""
        with self._connect() as connection:
            row = connection.execute(
                "SELECT COUNT(*) AS count FROM outbox_dead_letters"
            ).fetchone()
        return int(row["count"])

    def receipt(self, event_id: str) -> Optional[OutboxReceipt]:
        """Return the durable acknowledgement receipt for an event, if any."""

        with self._connect() as connection:
            row = connection.execute(
                """
                SELECT event_id, acknowledged_at, status_code, dataset_asset_id
                FROM outbox_receipts
                WHERE event_id = ?
                """,
                (event_id,),
            ).fetchone()
        if row is None:
            return None
        asset_id = row["dataset_asset_id"]
        return OutboxReceipt(
            event_id=str(row["event_id"]),
            acknowledged_at=float(row["acknowledged_at"]),
            status_code=int(row["status_code"]),
            dataset_asset_id=None if asset_id is None else int(asset_id),
        )

    def acknowledged_count(self) -> int:
        """Return the number of durable acknowledgement tombstones."""

        with self._connect() as connection:
            row = connection.execute(
                "SELECT COUNT(*) AS count FROM outbox_receipts"
            ).fetchone()
        return int(row["count"])

    def event_state(self, event_id: str) -> Optional[str]:
        """Return the durable state for one deterministic event id."""

        with self._connect() as connection:
            row = connection.execute(
                """
                SELECT 'pending' AS state
                FROM outbox_events WHERE event_id = ?
                UNION ALL
                SELECT 'dead_lettered' AS state
                FROM outbox_dead_letters WHERE event_id = ?
                UNION ALL
                SELECT 'acknowledged' AS state
                FROM outbox_receipts WHERE event_id = ?
                LIMIT 1
                """,
                (event_id, event_id, event_id),
            ).fetchone()
        return None if row is None else str(row["state"])

    def event_states(self, event_ids: Sequence[str]) -> Mapping[str, str]:
        """Return states for a bounded group of deterministic event ids."""

        identifiers = tuple(dict.fromkeys(str(value) for value in event_ids))
        if len(identifiers) > 500:
            raise ValueError("at most 500 event ids may be inspected at once")
        if not identifiers:
            return {}
        placeholders = ",".join("?" for _value in identifiers)
        query = """
            SELECT event_id, 'pending' AS state
            FROM outbox_events WHERE event_id IN ({0})
            UNION ALL
            SELECT event_id, 'dead_lettered' AS state
            FROM outbox_dead_letters WHERE event_id IN ({0})
            UNION ALL
            SELECT event_id, 'acknowledged' AS state
            FROM outbox_receipts WHERE event_id IN ({0})
        """.format(placeholders)
        with self._connect() as connection:
            rows = connection.execute(
                query,
                identifiers + identifiers + identifiers,
            ).fetchall()
        states: Dict[str, str] = {}
        for row in rows:
            states[str(row["event_id"])] = str(row["state"])
        return states

    def event_ids(self) -> Iterable[str]:
        """Return queued ids in delivery order; intended for diagnostics."""
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT event_id FROM outbox_events " "ORDER BY created_at, event_id"
            ).fetchall()
        return tuple(str(row["event_id"]) for row in rows)
