feat(calendar): schedule durable outbox recovery
This commit is contained in:
@@ -3,6 +3,7 @@ from __future__ import annotations
|
|||||||
from celery import Celery
|
from celery import Celery
|
||||||
|
|
||||||
from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider
|
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.module_management import load_startup_enabled_modules, startup_candidate_module_ids
|
||||||
from govoplan_core.core.modules import ModuleContext
|
from govoplan_core.core.modules import ModuleContext
|
||||||
from govoplan_core.core.notifications import CAPABILITY_NOTIFICATIONS_DISPATCH, NotificationDispatchProvider
|
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.campaigns.append_sent": {"queue": "append_sent"},
|
||||||
"govoplan.notifications.deliver": {"queue": "notifications"},
|
"govoplan.notifications.deliver": {"queue": "notifications"},
|
||||||
"govoplan.notifications.deliver_pending": {"queue": "notifications"},
|
"govoplan.notifications.deliver_pending": {"queue": "notifications"},
|
||||||
|
"govoplan.calendar.dispatch_outbox": {"queue": "calendar"},
|
||||||
},
|
},
|
||||||
worker_prefetch_multiplier=1,
|
worker_prefetch_multiplier=1,
|
||||||
task_acks_late=True,
|
task_acks_late=True,
|
||||||
task_reject_on_worker_lost=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
|
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)
|
@celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0)
|
||||||
def send_email(self, job_id: str):
|
def send_email(self, job_id: str):
|
||||||
"""Send one explicitly queued campaign job.
|
"""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:
|
with get_database().SessionLocal() as session:
|
||||||
return dict(_notification_dispatch().deliver_pending(session, tenant_id=tenant_id, limit=limit))
|
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
|
||||||
|
|||||||
98
src/govoplan_core/core/calendar.py
Normal file
98
src/govoplan_core/core/calendar.py
Normal file
@@ -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
|
||||||
@@ -49,7 +49,12 @@ class Settings(BaseSettings):
|
|||||||
auth_session_hours: int = Field(default=12, alias="AUTH_SESSION_HOURS")
|
auth_session_hours: int = Field(default=12, alias="AUTH_SESSION_HOURS")
|
||||||
|
|
||||||
master_key_b64: str | None = Field(default=None, alias="MASTER_KEY_B64")
|
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")
|
mock_mailbox_dir: str = Field(default="runtime/mock-mailbox", alias="MOCK_MAILBOX_DIR")
|
||||||
|
|
||||||
# Development bootstrap only. Do not use this in production.
|
# Development bootstrap only. Do not use this in production.
|
||||||
|
|||||||
48
tests/test_calendar_outbox_worker.py
Normal file
48
tests/test_calendar_outbox_worker.py
Normal file
@@ -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()
|
||||||
Reference in New Issue
Block a user