38 lines
1.6 KiB
Python
38 lines
1.6 KiB
Python
"""Transactional-outbox delivery orchestration.
|
|
|
|
State transitions persist an outbox row in the same SQLite transaction. This
|
|
dispatcher is the sole path that acknowledges a row after a delivery adapter
|
|
accepts it, which gives consumers a stable event ID for idempotent handling.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.cad_agent.ports import EventPublisher, TaskRepository
|
|
|
|
|
|
class OutboxDispatcher:
|
|
def __init__(self, repository: TaskRepository, publisher: EventPublisher) -> None:
|
|
self.repository = repository
|
|
self.publisher = publisher
|
|
|
|
async def dispatch_pending(self, *, limit: int = 100, task_id: str | None = None) -> list[dict[str, Any]]:
|
|
"""Deliver pending events in order and acknowledge only on success.
|
|
|
|
A process interruption after ``publish`` but before acknowledgement is
|
|
deliberately delivered again. Consumers use the stable ``event_id``
|
|
to collapse that at-least-once replay.
|
|
"""
|
|
delivered: list[dict[str, Any]] = []
|
|
for item in self.repository.pending_outbox(limit, task_id=task_id):
|
|
event_id = item.get("event_id")
|
|
task_id = item.get("task_id")
|
|
if not isinstance(event_id, int) or not isinstance(task_id, str):
|
|
raise ValueError("Outbox row has no stable event identity")
|
|
envelope = {**item, "event_id": event_id, "task_id": task_id}
|
|
await self.publisher.publish(envelope)
|
|
self.repository.mark_outbox_published(event_id)
|
|
delivered.append(envelope)
|
|
return delivered
|