M3: channels/threads + cron delivery + search + outbox sync
Gateway plugin: - ChannelDirectory (SQLite): channels + threads, friendly-name resolution - FTS5 search bridge over state.db (scope all/chat, LIKE fallback) - Outbox (monotonic cursor, 72h retention) for reconnect catch-up - adapter: channel.* handlers, search, sync, cron target parsing - ws_server: M3 frame routing + hello.ack sync cursor App (KMP): - ChannelStore (directory cache) + lane-aware ChatStore (per chat/thread) - GatewayClient.sendFrame; IrisController channel/search/nav actions - ChatScreen: channel drawer, thread toggle, topic switcher, search overlay
This commit is contained in:
ARIA committed
1 parent
218c50d688
commit
9b511fdd28
13 files changed
+2035
-259
No files matched your search
+154
-5
@@ -1,10 +1,159 @@
|
||||
"""SQLite offline outbox + monotonic sync cursor.
|
||||
"""SQLite offline outbox + monotonic sync cursor (M3).
|
||||
|
||||
Undelivered frames are appended per ``chat_id`` with a monotonic cursor so a
|
||||
reconnecting app can ``sync {cursor}`` the delta without re-reading full
|
||||
history. Retention prunes old entries (``outbox_retention_hours``).
|
||||
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
|
||||
Reference in new issue
Block a user