diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 45751a6..8b35f94 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -3,6 +3,7 @@ from __future__ import annotations from celery import Celery from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider +from govoplan_core.core.calendar import CAPABILITY_CALENDAR_OUTBOX, CalendarOutboxProvider from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids from govoplan_core.core.modules import ModuleContext from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH, NotificationDispatchProvider @@ -27,10 +28,18 @@ celery.conf.update( "govoplan.campaigns.append_sent": {"queue": "append_sent"}, "govoplan.notifications.deliver": {"queue": "notifications"}, "govoplan.notifications.deliver_pending": {"queue": "notifications"}, + "govoplan.calendar.dispatch_outbox": {"queue": "calendar"}, }, worker_prefetch_multiplier=1, task_acks_late=True, task_reject_on_worker_lost=True, + beat_schedule={ + "calendar-outbox-every-minute": { + "task": "govoplan.calendar.dispatch_outbox", + "schedule": 60.0, + "args": (None, 100), + }, + }, ) @@ -67,6 +76,16 @@ def _notification_dispatch() -> NotificationDispatchProvider: return capability +def _calendar_outbox() -> CalendarOutboxProvider | None: + registry = _platform_registry() + if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX): + return None + capability = registry.require_capability(CAPABILITY_CALENDAR_OUTBOX) + if not isinstance(capability, CalendarOutboxProvider): + raise RuntimeError("Calendar outbox capability is invalid") + return capability + + @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0) def send_email(self, job_id: str): """Send one explicitly queued campaign job. @@ -111,3 +130,18 @@ def deliver_pending_notifications(self, tenant_id: str | None = None, limit: int with get_database().SessionLocal() as session: return dict(_notification_dispatch().deliver_pending(session, tenant_id=tenant_id, limit=limit)) + + +@celery.task(name="govoplan.calendar.dispatch_outbox", bind=True, max_retries=0) +def dispatch_calendar_outbox(self, tenant_id: str | None = None, limit: int = 50): + """Drain durable Calendar operations; retry timing lives in the database.""" + + from govoplan_core.db.session import get_database + + with get_database().SessionLocal() as session: + provider = _calendar_outbox() + if provider is None: + return {"processed": 0, "succeeded": 0, "retrying": 0, "failed": 0, "operations": []} + result = dict(provider.dispatch_due(session, tenant_id=tenant_id, limit=limit)) + session.commit() + return result diff --git a/src/govoplan_core/core/calendar.py b/src/govoplan_core/core/calendar.py new file mode 100644 index 0000000..4adf44d --- /dev/null +++ b/src/govoplan_core/core/calendar.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Protocol, runtime_checkable + + +CAPABILITY_CALENDAR_SCHEDULING = "calendar.scheduling" +CAPABILITY_CALENDAR_OUTBOX = "calendar.outbox" +CALENDAR_AVAILABILITY_READ_SCOPE = "calendar:availability:read" +CALENDAR_EVENT_WRITE_SCOPE = "calendar:event:write" + + +class CalendarCapabilityError(ValueError): + """Stable error raised by Calendar capability implementations.""" + + +@dataclass(frozen=True, slots=True) +class CalendarEventRequest: + summary: str + start_at: datetime + calendar_id: str | None = None + description: str | None = None + location: str | None = None + status: str = "CONFIRMED" + transparency: str = "OPAQUE" + classification: str = "PUBLIC" + end_at: datetime | None = None + timezone: str | None = None + attendees: tuple[Mapping[str, object], ...] = () + categories: tuple[str, ...] = () + related_to: tuple[Mapping[str, object], ...] = () + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class CalendarEventRef: + id: str + calendar_id: str + uid: str + external_state: str = "local" + outbox_operation_id: str | None = None + + +@runtime_checkable +class CalendarSchedulingProvider(Protocol): + def list_freebusy( + self, + session: object, + *, + tenant_id: str, + start_at: datetime, + end_at: datetime, + calendar_ids: Sequence[str] | None = None, + ) -> Sequence[Mapping[str, object]]: + ... + + def create_event( + self, + session: object, + *, + tenant_id: str, + user_id: str | None, + request: CalendarEventRequest, + ) -> CalendarEventRef: + ... + + +@runtime_checkable +class CalendarOutboxProvider(Protocol): + """Stable worker boundary for Calendar-owned external side effects.""" + + def dispatch_due( + self, + session: object, + *, + tenant_id: str | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + ... + +def calendar_scheduling_provider(registry: object | None) -> CalendarSchedulingProvider | None: + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(CAPABILITY_CALENDAR_SCHEDULING): + return None + capability = registry.capability(CAPABILITY_CALENDAR_SCHEDULING) + return capability if isinstance(capability, CalendarSchedulingProvider) else None + + +def calendar_outbox_provider(registry: object | None) -> CalendarOutboxProvider | None: + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX): + return None + capability = registry.capability(CAPABILITY_CALENDAR_OUTBOX) + return capability if isinstance(capability, CalendarOutboxProvider) else None diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index 70992ec..d806b70 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -49,7 +49,12 @@ class Settings(BaseSettings): auth_session_hours: int = Field(default=12, alias="AUTH_SESSION_HOURS") master_key_b64: str | None = Field(default=None, alias="MASTER_KEY_B64") - celery_queues: str = Field(default="send_email,append_sent,notifications,default", alias="CELERY_QUEUES") + celery_queues: str = Field(default="send_email,append_sent,notifications,calendar,default", alias="CELERY_QUEUES") + calendar_outbox_terminal_retention_days: int = Field( + default=90, + ge=0, + alias="CALENDAR_OUTBOX_TERMINAL_RETENTION_DAYS", + ) mock_mailbox_dir: str = Field(default="runtime/mock-mailbox", alias="MOCK_MAILBOX_DIR") # Development bootstrap only. Do not use this in production. diff --git a/tests/test_calendar_outbox_worker.py b/tests/test_calendar_outbox_worker.py new file mode 100644 index 0000000..bd1d810 --- /dev/null +++ b/tests/test_calendar_outbox_worker.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from govoplan_core.celery_app import celery, dispatch_calendar_outbox + + +class CalendarOutboxWorkerTests(unittest.TestCase): + def test_worker_commits_provider_outcome(self) -> None: + session = MagicMock() + database = MagicMock() + database.SessionLocal.return_value.__enter__.return_value = session + provider = MagicMock() + provider.dispatch_due.return_value = { + "processed": 1, + "succeeded": 1, + "retrying": 0, + "failed": 0, + "operations": [{"id": "operation-1", "status": "succeeded"}], + } + + with ( + patch("govoplan_core.celery_app._calendar_outbox", return_value=provider), + patch("govoplan_core.db.session.get_database", return_value=database), + ): + result = dispatch_calendar_outbox.run("tenant-1", 25) + + provider.dispatch_due.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) + session.commit.assert_called_once_with() + self.assertEqual(result["succeeded"], 1) + + def test_calendar_worker_route_and_periodic_recovery_are_registered(self) -> None: + self.assertEqual( + celery.conf.task_routes["govoplan.calendar.dispatch_outbox"], + {"queue": "calendar"}, + ) + schedule = celery.conf.beat_schedule["calendar-outbox-every-minute"] + self.assertEqual(schedule["task"], "govoplan.calendar.dispatch_outbox") + self.assertEqual(schedule["schedule"], 60.0) + + +if __name__ == "__main__": + unittest.main()