feat(edge): OfflineQueue (SQLite persist + ordered resend)

edge/queue.py

@@ -0,0 +1,57 @@
+import base64
+import json
+import sqlite3
+
+from edge.events import Event, EventType, Severity
+
+
+class OfflineQueue:
+    def __init__(self, db_path: str, client):
+        self.client = client
+        self.db = sqlite3.connect(db_path)
+        self.db.execute(
+            "CREATE TABLE IF NOT EXISTS queue ("
+            "id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, snaps TEXT)"
+        )
+        self.db.commit()
+
+    def _rehydrate(self, payload: dict, snaps: list[bytes]) -> tuple[Event, list[bytes]]:
+        e = Event(EventType(payload["event_type"]), Severity(payload["severity"]),
+                  payload["camera_id"], payload["track_id"], payload["zone_id"],
+                  payload["occurred_at"], payload["duration_s"], payload["metrics"],
+                  tuple(payload["spatial"]) if payload.get("spatial") else None)
+        return e, snaps
+
+    def enqueue(self, event: Event, snapshots: list[bytes]) -> None:
+        payload = event.to_payload("")  # device_id는 client가 다시 채움
+        payload["event_type"] = event.event_type.value
+        payload["severity"] = event.severity.value
+        payload["spatial"] = list(event.spatial) if event.spatial else None
+        snaps = [base64.b64encode(s).decode() for s in snapshots]
+        self.db.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)",
+                        (json.dumps(payload), json.dumps(snaps)))
+        self.db.commit()
+
+    def try_send(self, event: Event, snapshots: list[bytes]) -> bool:
+        if self.client.send(event, snapshots):
+            return True
+        self.enqueue(event, snapshots)
+        return False
+
+    def flush(self) -> int:
+        rows = self.db.execute("SELECT id, payload, snaps FROM queue ORDER BY id").fetchall()
+        sent = 0
+        for rid, payload_json, snaps_json in rows:
+            payload = json.loads(payload_json)
+            snaps = [base64.b64decode(s) for s in json.loads(snaps_json)]
+            event, snapshots = self._rehydrate(payload, snaps)
+            if self.client.send(event, snapshots):
+                self.db.execute("DELETE FROM queue WHERE id = ?", (rid,))
+                self.db.commit()
+                sent += 1
+            else:
+                break  # 여전히 오프라인 — 순서 보존
+        return sent
+
+    def pending(self) -> int:
+        return self.db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]

tests/test_queue.py

@@ -0,0 +1,36 @@
+from edge.events import Event, EventType, Severity
+from edge.queue import OfflineQueue
+
+def _event(n=1):
+    return Event(EventType.FALL, Severity.HIGH, "cam-01", n, None, float(n), 3.0, {}, None)
+
+class _Client:
+    def __init__(self, fail=False):
+        self.fail = fail
+        self.sent = []
+    def send(self, event, snapshots):
+        if self.fail:
+            return False
+        self.sent.append(event.track_id)
+        return True
+
+def test_try_send_success_not_queued(tmp_path):
+    q = OfflineQueue(str(tmp_path / "q.db"), _Client())
+    assert q.try_send(_event(1), [b"x"]) is True
+    assert q.pending() == 0
+
+def test_failure_is_queued(tmp_path):
+    q = OfflineQueue(str(tmp_path / "q.db"), _Client(fail=True))
+    assert q.try_send(_event(1), [b"x"]) is False
+    assert q.pending() == 1
+
+def test_flush_resends_when_recovered(tmp_path):
+    client = _Client(fail=True)
+    q = OfflineQueue(str(tmp_path / "q.db"), client)
+    q.try_send(_event(1), [b"x"])
+    q.try_send(_event(2), [b"y"])
+    assert q.pending() == 2
+    client.fail = False
+    assert q.flush() == 2
+    assert q.pending() == 0
+    assert client.sent == [1, 2]