feat(core): dispatch bounded postbox routes
This commit is contained in:
@@ -22,6 +22,10 @@ from govoplan_core.core.events import (
|
||||
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
|
||||
from govoplan_core.core.postbox import (
|
||||
CAPABILITY_POSTBOX_ROUTING,
|
||||
PostboxRoutingProvider,
|
||||
)
|
||||
from govoplan_core.core.workflows import (
|
||||
CAPABILITY_WORKFLOW_RUNTIME_WORKER,
|
||||
WorkflowRuntimeWorker,
|
||||
@@ -52,6 +56,7 @@ celery.conf.update(
|
||||
"govoplan.dataflow.purge_runs": {"queue": "dataflow"},
|
||||
"govoplan.dataflow.dispatch_triggers": {"queue": "dataflow"},
|
||||
"govoplan.workflow.reconcile": {"queue": "workflow"},
|
||||
"govoplan.postbox.dispatch_routes": {"queue": "postbox"},
|
||||
"govoplan.events.dispatch_outbox": {"queue": "events"},
|
||||
"govoplan.events.purge_outbox": {"queue": "events"},
|
||||
},
|
||||
@@ -84,6 +89,11 @@ celery.conf.update(
|
||||
"schedule": 5.0,
|
||||
"args": (50,),
|
||||
},
|
||||
"postbox-routes-every-minute": {
|
||||
"task": "govoplan.postbox.dispatch_routes",
|
||||
"schedule": 60.0,
|
||||
"args": (None, 50),
|
||||
},
|
||||
"platform-events-every-ten-seconds": {
|
||||
"task": "govoplan.events.dispatch_outbox",
|
||||
"schedule": 10.0,
|
||||
@@ -181,6 +191,18 @@ def _workflow_runtime_worker(
|
||||
return capability
|
||||
|
||||
|
||||
def _postbox_routing_provider(
|
||||
registry: PlatformRegistry | None = None,
|
||||
) -> PostboxRoutingProvider | None:
|
||||
registry = registry or _platform_registry()
|
||||
if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING):
|
||||
return None
|
||||
capability = registry.require_capability(CAPABILITY_POSTBOX_ROUTING)
|
||||
if not isinstance(capability, PostboxRoutingProvider):
|
||||
raise RuntimeError("Postbox routing capability is invalid")
|
||||
return capability
|
||||
|
||||
|
||||
def _platform_event_outbox(
|
||||
registry: PlatformRegistry | None = None,
|
||||
) -> PlatformEventOutbox | None:
|
||||
@@ -358,6 +380,43 @@ def reconcile_workflow_instances(self, limit: int = 50):
|
||||
return result
|
||||
|
||||
|
||||
@celery.task(
|
||||
name="govoplan.postbox.dispatch_routes",
|
||||
bind=True,
|
||||
max_retries=0,
|
||||
)
|
||||
def dispatch_postbox_routes(
|
||||
self,
|
||||
tenant_id: str | None = None,
|
||||
limit: int = 50,
|
||||
):
|
||||
"""Deliver due Postbox vacancy escalations from durable route rows."""
|
||||
|
||||
from govoplan_core.db.session import get_database
|
||||
|
||||
with get_database().SessionLocal() as session:
|
||||
provider = _postbox_routing_provider()
|
||||
if provider is None:
|
||||
return {
|
||||
"selected": 0,
|
||||
"delivered": 0,
|
||||
"vacant": 0,
|
||||
"rescheduled": 0,
|
||||
"cancelled": 0,
|
||||
"failed": 0,
|
||||
"route_ids": [],
|
||||
}
|
||||
result = dict(
|
||||
provider.dispatch_due_routes(
|
||||
session,
|
||||
tenant_id=tenant_id,
|
||||
limit=limit,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return result
|
||||
|
||||
|
||||
@celery.task(
|
||||
name="govoplan.events.dispatch_outbox",
|
||||
bind=True,
|
||||
|
||||
@@ -130,6 +130,13 @@ class OrganizationRelationTypeRef:
|
||||
status: OrganizationStatus = "active"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class OrganizationHierarchyCatalogRef:
|
||||
tenant_id: str
|
||||
structures: tuple[OrganizationStructureRef, ...] = ()
|
||||
relation_types: tuple[OrganizationRelationTypeRef, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class OrganizationHierarchyEdgeRef:
|
||||
id: str
|
||||
@@ -231,6 +238,12 @@ class OrganizationDirectory(Protocol):
|
||||
|
||||
@runtime_checkable
|
||||
class OrganizationHierarchyDirectory(Protocol):
|
||||
def hierarchy_catalog(
|
||||
self,
|
||||
tenant_id: str,
|
||||
) -> OrganizationHierarchyCatalogRef:
|
||||
...
|
||||
|
||||
def get_unit_type(
|
||||
self,
|
||||
tenant_id: str,
|
||||
@@ -355,6 +368,7 @@ __all__ = [
|
||||
"OrganizationFunctionTypeRef",
|
||||
"OrganizationFunctionTypeResolution",
|
||||
"OrganizationHierarchyDirection",
|
||||
"OrganizationHierarchyCatalogRef",
|
||||
"OrganizationHierarchyDirectory",
|
||||
"OrganizationHierarchyEdgeRef",
|
||||
"OrganizationHierarchyMatchRef",
|
||||
|
||||
@@ -12,6 +12,7 @@ CAPABILITY_POSTBOX_ACCESS = "postbox.access"
|
||||
CAPABILITY_POSTBOX_MESSAGES = "postbox.messages"
|
||||
CAPABILITY_POSTBOX_DELIVERY = "postbox.delivery"
|
||||
CAPABILITY_POSTBOX_EVIDENCE = "postbox.evidence"
|
||||
CAPABILITY_POSTBOX_ROUTING = "postbox.routing"
|
||||
|
||||
PostboxAction = Literal["discover", "read", "send", "acknowledge", "administer"]
|
||||
PostboxMessageListState = Literal["all", "unread", "read", "acknowledged"]
|
||||
@@ -317,6 +318,18 @@ class PostboxEvidenceProvider(Protocol):
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class PostboxRoutingProvider(Protocol):
|
||||
def dispatch_due_routes(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str | None = None,
|
||||
limit: int = 50,
|
||||
) -> Mapping[str, object]:
|
||||
...
|
||||
|
||||
|
||||
def _postbox_provider(
|
||||
registry: object | None,
|
||||
*,
|
||||
@@ -384,3 +397,14 @@ def postbox_evidence_provider(
|
||||
provider_type=PostboxEvidenceProvider,
|
||||
)
|
||||
return provider if isinstance(provider, PostboxEvidenceProvider) else None
|
||||
|
||||
|
||||
def postbox_routing_provider(
|
||||
registry: object | None,
|
||||
) -> PostboxRoutingProvider | None:
|
||||
provider = _postbox_provider(
|
||||
registry,
|
||||
capability_name=CAPABILITY_POSTBOX_ROUTING,
|
||||
provider_type=PostboxRoutingProvider,
|
||||
)
|
||||
return provider if isinstance(provider, PostboxRoutingProvider) else None
|
||||
|
||||
Reference in New Issue
Block a user