feat(event): add single-thread event publisher with retry and cleanup

channel-service/app/repositories/event_store.py

@@ -96,6 +96,15 @@ class EventStore:
             attempt_count=row["attempt_count"],
         )
 
+    def get(self, event_id: str) -> Optional[OutboundEvent]:
+        row = self._connection().execute(
+            "SELECT id, event_type, payload, status, attempt_count FROM outbound_events WHERE id = ?",
+            (event_id,),
+        ).fetchone()
+        if row is None:
+            return None
+        return self._row_to_event(row)
+
     def mark_delivered(self, event_id: str) -> None:
         now = self._now().isoformat()
         with self._connection() as conn:

channel-service/app/services/event_publisher.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from app.clients.mci_server_client import MciServerClient
+from app.models.event import EventStatus, OutboundEvent
+from app.repositories.event_store import EventStore
+
+
+class EventPublisher:
+    def __init__(
+        self,
+        store: EventStore,
+        client: MciServerClient,
+        max_retry: int = 5,
+        base_delay_seconds: int = 5,
+        interval_seconds: float = 5.0,
+        batch_size: int = 100,
+    ):
+        self.store = store
+        self.client = client
+        self.max_retry = min(max_retry, 10)
+        self.base_delay_seconds = base_delay_seconds
+        self.interval_seconds = interval_seconds
+        self.batch_size = batch_size
+        self._task: Optional[asyncio.Task] = None
+        self._stop_event = asyncio.Event()
+
+    def start(self) -> None:
+        self._task = asyncio.create_task(self._loop())
+
+    async def stop(self) -> None:
+        self._stop_event.set()
+        if self._task:
+            try:
+                await asyncio.wait_for(self._task, timeout=5.0)
+            except asyncio.TimeoutError:
+                self._task.cancel()
+
+    async def _loop(self) -> None:
+        cleanup_counter = 0
+        cleanup_interval = max(1, int(24 * 3600 / self.interval_seconds))
+        while not self._stop_event.is_set():
+            try:
+                await self._tick()
+            except Exception:
+                pass
+            cleanup_counter += 1
+            if cleanup_counter >= cleanup_interval:
+                try:
+                    await self.cleanup_old_events(10)
+                except Exception:
+                    pass
+                cleanup_counter = 0
+            try:
+                await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds)
+            except asyncio.TimeoutError:
+                pass
+
+    async def _tick(self) -> None:
+        events = self.store.fetch_pending(limit=self.batch_size)
+        for event in events:
+            try:
+                await self.client.post_event(event)
+                self.store.mark_delivered(event.event_id)
+            except Exception as e:
+                next_attempt = event.attempt_count + 1
+                delay = self.base_delay_seconds * (2 ** (next_attempt - 1))
+                next_retry_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(seconds=delay)
+                status = EventStatus.DEAD_LETTER if next_attempt >= self.max_retry else EventStatus.RETRYING
+                self.store.mark_retry(event.event_id, status, next_attempt, next_retry_at, str(e)[:500])
+
+    async def cleanup_old_events(self, retention_days: int) -> int:
+        return self.store.cleanup(retention_days)

channel-service/requirements.txt

@@ -5,3 +5,4 @@ pydantic-settings==2.7.1
 httpx==0.28.1
 pytest==8.3.4
 pytest-asyncio==0.25.2
+respx==0.22.0

channel-service/tests/test_event_publisher.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+import respx
+from httpx import Response
+
+from app.clients.mci_server_client import MciServerClient
+from app.models.event import EventStatus, EventType, OutboundEvent
+from app.repositories.event_store import EventStore
+from app.services.event_publisher import EventPublisher
+
+
+@pytest.fixture
+def publisher(tmp_path):
+    db_path = tmp_path / "events.db"
+    store = EventStore(str(db_path))
+    client = MciServerClient("http://mock-mci-server/msg-platform")
+    pub = EventPublisher(store, client, max_retry=2, base_delay_seconds=1, interval_seconds=0.1)
+    yield pub
+
+
+@respx.mock
+def test_publish_delivers_event(publisher):
+    route = respx.post("http://mock-mci-server/msg-platform/api/v1/channel-account/event").mock(
+        return_value=Response(200)
+    )
+    event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {"online_status": "online"})
+    publisher.store.save(event)
+
+    async def run():
+        await publisher._tick()
+
+    asyncio.run(run())
+    assert route.called
+    pending = publisher.store.fetch_pending(limit=10)
+    assert len(pending) == 0
+
+
+@respx.mock
+def test_publish_retries_then_dead_letter(publisher):
+    route = respx.post("http://mock-mci-server/msg-platform/api/v1/channel-account/event").mock(
+        return_value=Response(500)
+    )
+    event = OutboundEvent(EventType.HEARTBEAT, "bot-001", "rescue", {})
+    publisher.store.save(event)
+
+    async def run():
+        await publisher._tick()
+
+    asyncio.run(run())
+    assert route.call_count == 1
+    updated = publisher.store.get(event.event_id)
+    assert updated is not None
+    assert updated.status == EventStatus.RETRYING
+    assert updated.attempt_count == 1