From 0e6f7a3072cd6ab08ac04fd04f2578f7f19a7b54 Mon Sep 17 00:00:00 2001 From: marsal Date: Fri, 10 Jul 2026 12:08:24 +0800 Subject: [PATCH] feat(event): add single-thread event publisher with retry and cleanup --- .../app/repositories/event_store.py | 9 +++ .../app/services/event_publisher.py | 76 +++++++++++++++++++ channel-service/requirements.txt | 1 + channel-service/tests/test_event_publisher.py | 57 ++++++++++++++ 4 files changed, 143 insertions(+) create mode 100644 channel-service/app/services/event_publisher.py create mode 100644 channel-service/tests/test_event_publisher.py diff --git a/channel-service/app/repositories/event_store.py b/channel-service/app/repositories/event_store.py index f40270d..6eac066 100644 --- a/channel-service/app/repositories/event_store.py +++ b/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: diff --git a/channel-service/app/services/event_publisher.py b/channel-service/app/services/event_publisher.py new file mode 100644 index 0000000..b163dfa --- /dev/null +++ b/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) diff --git a/channel-service/requirements.txt b/channel-service/requirements.txt index f2f029d..8a73209 100644 --- a/channel-service/requirements.txt +++ b/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 diff --git a/channel-service/tests/test_event_publisher.py b/channel-service/tests/test_event_publisher.py new file mode 100644 index 0000000..30e1672 --- /dev/null +++ b/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