feat(event): add SQLite outbound event store

2026-07-10 12:06:19 +08:00
parent b2e5cc187c
commit 41cb52daa1
3 changed files with 173 additions and 0 deletions
--- a/channel-service/app/repositories/event_store.py
+++ b/channel-service/app/repositories/event_store.py
@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import json
+import sqlite3
+import threading
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from app.models.event import EventStatus, EventType, OutboundEvent
+
+
+class EventStore:
+    def __init__(self, db_path: str):
+        self.db_path = db_path
+        self._local = threading.local()
+        self._init_db()
+
+    def _now(self) -> datetime:
+        return datetime.now(timezone.utc).replace(tzinfo=None)
+
+    def _connection(self) -> sqlite3.Connection:
+        if not hasattr(self._local, "conn") or self._local.conn is None:
+            self._local.conn = sqlite3.connect(self.db_path, check_same_thread=False)
+            self._local.conn.row_factory = sqlite3.Row
+        return self._local.conn
+
+    def _init_db(self) -> None:
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS outbound_events (
+                    id TEXT PRIMARY KEY,
+                    event_type TEXT NOT NULL,
+                    payload TEXT NOT NULL,
+                    status TEXT NOT NULL DEFAULT 'pending',
+                    attempt_count INTEGER NOT NULL DEFAULT 0,
+                    next_retry_at TIMESTAMP NOT NULL,
+                    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                    last_error TEXT
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_status_next_retry ON outbound_events(status, next_retry_at)"
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_created_at ON outbound_events(created_at)"
+            )
+
+    def save(self, event: OutboundEvent) -> None:
+        now = self._now().isoformat()
+        with self._connection() as conn:
+            conn.execute(
+                """
+                INSERT INTO outbound_events
+                (id, event_type, payload, status, attempt_count, next_retry_at, created_at, updated_at, last_error)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    event.event_id,
+                    event.event_type.value,
+                    json.dumps(event.to_payload(), default=str),
+                    EventStatus.PENDING.value,
+                    0,
+                    now,
+                    now,
+                    now,
+                    None,
+                ),
+            )
+
+    def fetch_pending(self, limit: int = 100) -> list[OutboundEvent]:
+        now = self._now().isoformat()
+        cursor = self._connection().execute(
+            """
+            SELECT id, event_type, payload, status, attempt_count
+            FROM outbound_events
+            WHERE status IN ('pending', 'retrying') AND next_retry_at <= ?
+            ORDER BY next_retry_at ASC
+            LIMIT ?
+            """,
+            (now, limit),
+        )
+        return [self._row_to_event(row) for row in cursor.fetchall()]
+
+    def _row_to_event(self, row: sqlite3.Row) -> OutboundEvent:
+        payload = json.loads(row["payload"])
+        return OutboundEvent(
+            event_id=payload["event_id"],
+            event_type=EventType(payload["event_type"]),
+            account_id=payload["account_id"],
+            tenant_code=payload["tenant_code"],
+            data=payload["data"],
+            status=EventStatus(row["status"]),
+            attempt_count=row["attempt_count"],
+        )
+
+    def mark_delivered(self, event_id: str) -> None:
+        now = self._now().isoformat()
+        with self._connection() as conn:
+            conn.execute(
+                "UPDATE outbound_events SET status = 'delivered', updated_at = ? WHERE id = ?",
+                (now, event_id),
+            )
+
+    def mark_retry(self, event_id: str, status: EventStatus, attempt_count: int, next_retry_at: datetime, last_error: Optional[str]) -> None:
+        now = self._now().isoformat()
+        with self._connection() as conn:
+            conn.execute(
+                """
+                UPDATE outbound_events
+                SET status = ?, attempt_count = ?, next_retry_at = ?, updated_at = ?, last_error = ?
+                WHERE id = ?
+                """,
+                (status.value, attempt_count, next_retry_at.isoformat(), now, last_error, event_id),
+            )
+
+    def cleanup(self, retention_days: int) -> int:
+        threshold = (self._now() - timedelta(days=retention_days)).isoformat()
+        with self._connection() as conn:
+            cursor = conn.execute(
+                """
+                DELETE FROM outbound_events
+                WHERE status IN ('delivered', 'dead_letter') AND created_at < ?
+                """,
+                (threshold,),
+            )
+            return cursor.rowcount