diff --git a/channel-service/app/models/event.py b/channel-service/app/models/event.py index e046c0e..74ea93f 100644 --- a/channel-service/app/models/event.py +++ b/channel-service/app/models/event.py @@ -32,6 +32,8 @@ class OutboundEvent: timestamp: datetime = field( default_factory=lambda: datetime.now(timezone.utc) ) + status: EventStatus = field(default=EventStatus.PENDING) + attempt_count: int = field(default=0) def to_payload(self) -> dict[str, Any]: return { diff --git a/channel-service/app/repositories/event_store.py b/channel-service/app/repositories/event_store.py new file mode 100644 index 0000000..f40270d --- /dev/null +++ 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 diff --git a/channel-service/tests/test_event_store.py b/channel-service/tests/test_event_store.py new file mode 100644 index 0000000..907f18d --- /dev/null +++ b/channel-service/tests/test_event_store.py @@ -0,0 +1,42 @@ +import os +import tempfile + +import pytest + +from app.models.event import EventType, OutboundEvent +from app.repositories.event_store import EventStore + + +@pytest.fixture +def store(): + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + store = EventStore(path) + yield store + os.unlink(path) + + +def test_save_and_fetch_pending(store): + event = OutboundEvent( + event_type=EventType.HEARTBEAT, + account_id="bot-001", + tenant_code="rescue", + data={"online_status": "online"}, + ) + store.save(event) + pending = store.fetch_pending(limit=10) + assert len(pending) == 1 + assert pending[0].event_id == event.event_id + + +def test_mark_delivered(store): + event = OutboundEvent( + event_type=EventType.HEARTBEAT, + account_id="bot-001", + tenant_code="rescue", + data={}, + ) + store.save(event) + store.mark_delivered(event.event_id) + pending = store.fetch_pending(limit=10) + assert len(pending) == 0