iris_x_hermes/gateway-plugin/outbox.py

"""SQLite offline outbox + monotonic sync cursor (M3).

Undelivered frames (sent while no device is live) are appended with a
**monotonic** cursor so a reconnecting app can ``sync {cursor}`` the delta
without re-reading full history. The cursor is a separate high-water counter
that only ever increases -- pruning old rows never resets it, so a late
reconnect can't be handed a cursor lower than one it already saw.

Retention prunes rows older than ``outbox_retention_hours`` (default 72h). A
device offline longer than the window misses those frames; it recovers full
context via ``history`` (M5 wires push so the device is woken to sync).

Storage: ``get_hermes_home()/"android"/outbox.db``.

Milestone M3 (built), extended in M5 (push integration).
"""

import json
import logging
import sqlite3
import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)

DEFAULT_RETENTION_HOURS = 72
_REPLAY_LIMIT = 1000
_PRUNE_INTERVAL_S = 3600.0


class Outbox:
    """Persistent outbox under ``get_hermes_home()/"android"``.

    Thread-safe (single connection + lock); operations are small and fast
    enough to run inline on the gateway's asyncio loop (mirrors
    ``DeviceRegistry`` / ``ChannelDirectory``).
    """

    def __init__(self, db_path: Path, retention_hours: int = DEFAULT_RETENTION_HOURS):
        self._db_path = Path(db_path)
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._retention_hours = max(1, int(retention_hours))
        self._lock = threading.Lock()
        self._last_prune = 0.0
        self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        with self._lock:
            self._conn.execute("PRAGMA journal_mode=WAL")
            self._conn.execute(
                """
                CREATE TABLE IF NOT EXISTS outbox (
                    cursor  INTEGER PRIMARY KEY,
                    chat_id TEXT,
                    frame   TEXT NOT NULL,
                    created REAL NOT NULL DEFAULT 0
                )
                """
            )
            self._conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_outbox_created ON outbox (created)"
            )
            self._conn.execute(
                """
                CREATE TABLE IF NOT EXISTS counters (
                    name  TEXT PRIMARY KEY,
                    value INTEGER NOT NULL DEFAULT 0
                )
                """
            )
            self._conn.commit()

    # ── append / cursor ───────────────────────────────────────────────────

    def append(self, chat_id: Optional[str], frame_json: str) -> int:
        """Append a frame; returns the (monotonic) cursor assigned to it."""
        now = time.time()
        with self._lock:
            self._conn.execute(
                "INSERT INTO counters (name, value) VALUES ('cursor', 1) "
                "ON CONFLICT(name) DO UPDATE SET value = value + 1"
            )
            row = self._conn.execute(
                "SELECT value FROM counters WHERE name = 'cursor'"
            ).fetchone()
            cursor = int(row["value"]) if row else 1
            self._conn.execute(
                "INSERT INTO outbox (cursor, chat_id, frame, created) "
                "VALUES (?, ?, ?, ?)",
                (cursor, chat_id, frame_json, now),
            )
            self._conn.commit()
        self._maybe_prune()
        return cursor

    def latest_cursor(self) -> int:
        """The high-water cursor (0 when nothing has been appended)."""
        with self._lock:
            row = self._conn.execute(
                "SELECT value FROM counters WHERE name = 'cursor'"
            ).fetchone()
        return int(row["value"]) if row else 0

    # ── replay ────────────────────────────────────────────────────────────

    def replay(self, cursor: int, limit: int = _REPLAY_LIMIT) -> List[Dict[str, Any]]:
        """Frames with ``cursor > `cursor```, oldest first.

        Each entry: ``{cursor, chat_id, frame}`` where ``frame`` is the parsed
        frame dict (the caller re-serializes / forwards it to the device).
        """
        cursor = max(0, int(cursor or 0))
        limit = max(1, min(int(limit or _REPLAY_LIMIT), _REPLAY_LIMIT))
        with self._lock:
            rows = self._conn.execute(
                "SELECT cursor, chat_id, frame FROM outbox "
                "WHERE cursor > ? ORDER BY cursor ASC LIMIT ?",
                (cursor, limit),
            ).fetchall()
        out: List[Dict[str, Any]] = []
        for r in rows:
            try:
                frame = json.loads(r["frame"])
            except (json.JSONDecodeError, TypeError):
                continue
            if not isinstance(frame, dict):
                continue
            out.append(
                {"cursor": int(r["cursor"]), "chat_id": r["chat_id"], "frame": frame}
            )
        return out

    # ── retention ─────────────────────────────────────────────────────────

    def _maybe_prune(self) -> None:
        now = time.time()
        if now - self._last_prune < _PRUNE_INTERVAL_S:
            return
        self._last_prune = now
        cutoff = now - self._retention_hours * 3600
        with self._lock:
            try:
                self._conn.execute("DELETE FROM outbox WHERE created < ?", (cutoff,))
                self._conn.commit()
            except sqlite3.Error as e:
                logger.debug("android outbox: prune failed: %s", e)

    def prune(self) -> None:
        """Force a retention prune (ignores the interval throttle)."""
        self._last_prune = 0.0
        self._maybe_prune()

    def close(self) -> None:
        with self._lock:
            try:
                self._conn.close()
            except Exception:
                pass