Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import asyncio | |
| import random | |
| from datetime import datetime, timedelta, timezone | |
| from app.analytics.service import AnalyticsDomainService | |
| from app.core.logger import get_logger | |
| from app.social.database import SocialDatabase | |
| logger = get_logger(__name__) | |
| class AnalyticsSyncWorker: | |
| """Claims durable sync runs and executes bounded provider synchronization.""" | |
| def __init__( | |
| self, | |
| analytics: AnalyticsDomainService, | |
| database: SocialDatabase, | |
| *, | |
| interval_seconds: int, | |
| concurrency: int = 2, | |
| ) -> None: | |
| self.analytics = analytics | |
| self.database = database | |
| self.interval_seconds = max(1, interval_seconds) | |
| self.concurrency = max(1, min(concurrency, 8)) | |
| self._task: asyncio.Task[None] | None = None | |
| self._stop = asyncio.Event() | |
| async def start(self) -> None: | |
| if self._task is not None or not self.analytics.ready: | |
| return | |
| self._stop.clear() | |
| self._task = asyncio.create_task(self._run(), name="analytics-sync-worker") | |
| async def stop(self) -> None: | |
| self._stop.set() | |
| if self._task is not None: | |
| self._task.cancel() | |
| await asyncio.gather(self._task, return_exceptions=True) | |
| self._task = None | |
| async def run_once(self) -> None: | |
| if not self.analytics.ready: | |
| return | |
| async with self.database.worker_boundary(): | |
| runs = await self.analytics.repository.claim_due_syncs(limit=self.concurrency) | |
| await asyncio.gather(*(self._execute(run) for run in runs)) | |
| async def _execute(self, run) -> None: | |
| try: | |
| latest = await self.analytics.repository.get_sync(run.workspace_id, run.id) | |
| if latest.status == "cancelled": | |
| return | |
| await self.analytics.sync_once(run) | |
| except asyncio.CancelledError: | |
| raise | |
| except Exception as exc: | |
| retryable = run.attempt_count < 3 | |
| if retryable: | |
| delay = min(900, 30 * (2 ** max(0, run.attempt_count - 1))) + random.randint(0, 10) | |
| await self.analytics.repository.update_sync( | |
| run.id, | |
| status="queued", | |
| error_code=getattr(exc, "code", "ANALYTICS_SYNC_RETRY"), | |
| error_message="Analytics synchronization will be retried.", | |
| next_attempt_at=datetime.now(timezone.utc) + timedelta(seconds=delay), | |
| ) | |
| else: | |
| await self.analytics.repository.update_sync( | |
| run.id, | |
| status="failed", | |
| error_code=getattr(exc, "code", "ANALYTICS_SYNC_FAILED"), | |
| error_message="Analytics synchronization failed.", | |
| ) | |
| await self.analytics.audit.record( | |
| workspace_id=run.workspace_id, | |
| event_type="analytics.sync_failed", | |
| metadata={"sync_run_id": run.id, "error_code": getattr(exc, "code", None)}, | |
| ) | |
| logger.warning( | |
| "analytics sync execution failed", | |
| extra={"sync_run_id": run.id, "attempt": run.attempt_count}, | |
| ) | |
| async def _run(self) -> None: | |
| while not self._stop.is_set(): | |
| try: | |
| await self.run_once() | |
| except asyncio.CancelledError: | |
| raise | |
| except Exception: | |
| logger.exception("analytics sync worker iteration failed") | |
| try: | |
| await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds) | |
| except TimeoutError: | |
| pass | |