930490d1b4
Two approved data-model changes. P1-9 — ingest idempotency. A POST can succeed server-side but time out client-side; the edge then re-queues and re-sends the same event, so the cloud stored duplicate alerts. The edge Event now carries a stable event_uuid (uuid4 hex) generated once at creation and preserved through the offline queue and every retry. Cloud: events.event_uuid CHAR(32) NULL UNIQUE; EventValidator accepts/validates it; IngestController returns the already-stored event (200, dedup:true) on a repeat uuid, race-safe against the UNIQUE key (catches the insert-race PDOException and returns the winner). P3-2b — site-scoped access control. site_scope was stored in the session but never enforced, so any authenticated user could read every site's worker snapshots (PIPA). Added devices.site; ingest denormalizes it onto events.site (server-derived from the authenticated device_id, never client-supplied). App\Security\SiteScope centralizes the rule (NULL scope = global operator). The event list filters in SQL (page(...,$site)); event detail + acknowledge and the snapshot proxy guard per-row and return 404 (not 403) cross-site so existence never leaks. Schema migration ALTERs and per-device/user provisioning UPDATEs are documented at the bottom of schema.sql. Tests: edge queue round-trip keeps the uuid; repeat-uuid dedups to one row; list scoped to caller site; detail cross-site 404; snapshot cross-site 404 / same-site 200; event_uuid validation. phpunit 97 passing (4 skipped: PDO), pytest 158 passing (2 skipped). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019Xmn8KZWqiQUguEMRtn6rW
89 lines
4.4 KiB
Python
89 lines
4.4 KiB
Python
import base64
|
|
import json
|
|
import sqlite3
|
|
|
|
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
|
|
# cloud 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:
|
|
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
|
|
continue
|
|
# Failed. A bool send() can't tell a transient outage from a permanently-rejected
|
|
# (poison) event, so count attempts: below the cap, stop and preserve order (an
|
|
# outage recovers and the whole queue drains next flush); at the cap, DROP this
|
|
# row and continue so a poison head can never wedge the queue forever.
|
|
attempts += 1
|
|
if attempts >= self.max_send_attempts:
|
|
self.db.execute("DELETE FROM queue WHERE id = ?", (rid,))
|
|
self.db.commit()
|
|
continue
|
|
self.db.execute("UPDATE queue SET attempts = ? WHERE id = ?", (attempts, rid))
|
|
self.db.commit()
|
|
break # transient — preserve order, retry the whole queue next time
|
|
return sent
|
|
|
|
def pending(self) -> int:
|
|
return self.db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
|