"""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 # Row cap (docs/08 §8.3): a very busy offline period must not grow the outbox # without bound; the oldest rows beyond the cap are pruned and the app is # told (generic notification, throttled in the adapter). DEFAULT_MAX_ROWS = 5000 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, max_rows: int = DEFAULT_MAX_ROWS, ): 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._max_rows = max(1, int(max_rows)) self._lock = threading.Lock() self._last_prune = 0.0 self._overflow_pruned = 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._enforce_row_cap() self._conn.commit() self._maybe_prune() return cursor def take_overflow_pruned(self) -> int: """Rows pruned by the row cap since the last call (and reset to 0). The adapter turns a non-zero count into a (throttled) generic notification so the app knows older frames are gone. """ with self._lock: n = self._overflow_pruned self._overflow_pruned = 0 return n def _enforce_row_cap(self) -> None: """Drop the oldest rows beyond ``max_rows`` (caller holds the lock).""" row = self._conn.execute("SELECT COUNT(*) AS n FROM outbox").fetchone() n = int(row["n"]) if row else 0 excess = n - self._max_rows if excess <= 0: return self._conn.execute( "DELETE FROM outbox WHERE cursor IN (" " SELECT cursor FROM outbox ORDER BY cursor ASC LIMIT ?)", (excess,), ) self._overflow_pruned += excess logger.info("android outbox: row cap pruned %s oldest row(s)", excess) 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