admin / Syanpse-Vanguard

"""Local disk buffer (SQLite ring) so no logs are lost during a server/network
outage. Each event gets a monotonic autoincrement id; the sender uses the max id
in a batch as the bundle seq and trims rows <= the server-acked last_seq. Replay
after reconnect is therefore idempotent server-side."""
from __future__ import annotations

import json
import os
import sqlite3
import threading
import time


class DiskBuffer:
    def __init__(self, path: str, max_rows: int = 500_000):
        os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
        self._db = sqlite3.connect(path, check_same_thread=False)
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute(
            "CREATE TABLE IF NOT EXISTS buf ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, line TEXT)"
        )
        self._db.commit()
        self._max_rows = max_rows
        self._lock = threading.Lock()

    def append(self, event: dict) -> None:
        line = json.dumps(event, default=str)
        with self._lock:
            self._db.execute("INSERT INTO buf (ts, line) VALUES (?, ?)",
                             (time.time(), line))
            self._db.commit()
            self._enforce_cap()

    def _enforce_cap(self) -> None:
        row = self._db.execute("SELECT COUNT(*) FROM buf").fetchone()
        if row and row[0] > self._max_rows:
            # Drop the oldest overflow — bounded disk use; newest data wins.
            excess = row[0] - self._max_rows
            self._db.execute(
                "DELETE FROM buf WHERE id IN "
                "(SELECT id FROM buf ORDER BY id ASC LIMIT ?)", (excess,))
            self._db.commit()

    def next_batch(self, limit: int) -> tuple[int, list[str]]:
        """Return (max_id, lines). max_id becomes the bundle seq."""
        with self._lock:
            rows = self._db.execute(
                "SELECT id, line FROM buf ORDER BY id ASC LIMIT ?", (limit,)
            ).fetchall()
        if not rows:
            return 0, []
        return rows[-1][0], [r[1] for r in rows]

    def trim(self, last_id: int) -> None:
        with self._lock:
            self._db.execute("DELETE FROM buf WHERE id <= ?", (last_id,))
            self._db.commit()

    def pending(self) -> int:
        return self._db.execute("SELECT COUNT(*) FROM buf").fetchone()[0]