import base64 import json import sqlite3 from edge.alert import SendResult from edge.events import Event, EventType, Severity class OfflineQueue: def __init__(self, db_path: str, client, max_rows: int = 2000, max_send_attempts: int = 20): self.client = client self.max_rows = max_rows # SD-card guard: evict oldest beyond this self.max_send_attempts = max_send_attempts # drop a poison event after this many tries self.db = sqlite3.connect(db_path) self.db.execute( "CREATE TABLE IF NOT EXISTS queue (" "id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, snaps TEXT, " "attempts INTEGER DEFAULT 0)" ) # Migrate an older DB that predates the attempts column. try: self.db.execute("ALTER TABLE queue ADD COLUMN attempts INTEGER DEFAULT 0") except sqlite3.OperationalError: pass # column already exists self.db.commit() def _rehydrate(self, payload: dict, snaps: list[bytes]) -> tuple[Event, list[bytes]]: # Preserve the stored event_uuid so a re-sent event keeps its idempotency key and the # server dedups it. Rows queued before this field existed simply get a fresh uuid (they # predate dedup, so a first-and-only delivery is correct for them). uuid_kw = {"event_uuid": payload["event_uuid"]} if payload.get("event_uuid") else {} 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, **uuid_kw) 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))) # Bound the queue so a multi-day outage — or a permanent misconfig that queues every # event forever — cannot fill the SD card and hard-crash the Pi. Keep the newest # max_rows; a fresh fall alert matters more than a stale one. self.db.execute( "DELETE FROM queue WHERE id NOT IN " "(SELECT id FROM queue ORDER BY id DESC LIMIT ?)", (self.max_rows,)) 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, attempts FROM queue ORDER BY id").fetchall() sent = 0 for rid, payload_json, snaps_json, attempts in rows: try: payload = json.loads(payload_json) snaps = [base64.b64decode(s) for s in json.loads(snaps_json)] event, snapshots = self._rehydrate(payload, snaps) except Exception: # An undecodable row — a torn SD-card write, or schema drift from an older edge # build whose payload lacks a key _rehydrate indexes — can NEVER be delivered. # flush() breaks on the first undelivered row to preserve order, so leaving this # row in place would wedge every queued safety event BEHIND it forever. Drop the # single unrecoverable row and continue; losing it beats freezing the whole queue. self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) self.db.commit() continue # A row queued before event_uuid existed has none stored; _rehydrate minted a fresh # one just now. Persist it so every retry of this row reuses the SAME key — otherwise # each flush sends a different uuid and a timeout-retried delivery duplicates on the # server (the P1-9 bug, still open for transitional rows). if not payload.get("event_uuid"): payload["event_uuid"] = event.event_uuid self.db.execute("UPDATE queue SET payload = ? WHERE id = ?", (json.dumps(payload), rid)) self.db.commit() result = self.client.send(event, snapshots) if result: # SendResult.DELIVERED (only truthy result) self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) self.db.commit() sent += 1 continue # Not delivered. Only a DEFINITE rejection (4xx) counts toward the poison drop cap — # a transient outage (network error / 429 / 5xx) must retry forever so a real fall # alert survives an overnight outage; max_rows already bounds the disk. This is the # fix for the old bool send(), which counted every offline failure toward the cap and # silently deleted the oldest queued safety events after ~20 flushes of downtime. if result is SendResult.REJECTED: attempts += 1 if attempts >= self.max_send_attempts: self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) self.db.commit() continue # poison dropped — let the events behind it drain self.db.execute("UPDATE queue SET attempts = ? WHERE id = ?", (attempts, rid)) self.db.commit() break # preserve order; retry the whole queue next flush return sent def pending(self) -> int: return self.db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]