Transactional notifications

A "welcome email" that fires on signup shouldn't send if the signup rolled back, and shouldn't be lost if it committed. That's the outbox guarantee again — plus a small kit that routes each relayed event to the right notification and hands it to a sender.

The runnable version lives at examples/recipes/notifications/ (mock — no broker or SMTP needed).

Stage the notification¶

The producer stages an event exactly like the outbox recipe — the notification is just the integration event's purpose:

class UserRegistered(BaseModel):
    email: str


NOTIFY_EVENTS = OutboxSpec(
    name="notify-events",
    codec=PydanticModelCodec(UserRegistered),
    destination=OutboxDestination.queue(route="notifications", channel="notifications"),
)
NOTIFICATIONS = QueueSpec(
    name="notifications", codec=PydanticModelCodec(UserRegistered)
)

Route events to notifications¶

A NotificationRouter registers a mapper per event type, then freeze()s into an immutable FrozenNotificationRouter the consumer resolves against. Registration and resolution are separate — the routing table is fixed once frozen, so it can't change under a running consumer. The mapper receives the integration event, so it reads the payload:

# Map each integration event type to the notifications it should produce, then freeze:
# registration happens once at wiring time; the consumer holds an immutable resolver.
router = (
    NotificationRouter()
    .register(
        "user.registered",
        lambda event: [
            EmailNotification(
                to=event.payload.email, subject="Welcome", body="Thanks for joining!"
            )
        ],
    )
    .freeze()
)

EmailNotification, PushNotification, and WebhookNotification are the shipped shapes.

Provide senders¶

NotificationSenders is a protocol — any object with send_email / send_push / send_webhook satisfies it. Real senders wrap SMTP, FCM, or an HTTP client; here it just records:

class RecordingSenders:
    """A NotificationSenders implementation — here it just records what it sent."""

    def __init__(self) -> None:
        self.emails: list[EmailNotification] = []

    async def send_email(self, notification: EmailNotification) -> None:
        self.emails.append(notification)

    async def send_push(self, notification: object) -> None: ...
    async def send_webhook(self, notification: object) -> None: ...

Consume and dispatch¶

OutboxRelay(...).to_queue(...) publishes the staged events to the queue; a QueueConsumer then drains that queue, routing each message through the frozen router to the matching sender. Going through the consumer (rather than a hand-rolled receive/ack loop) gives inbox dedup — an at-least-once redelivery won't re-send — and poison parking, for free. In production wire the consumer as a background step with notification_consumer_lifecycle_step; the example drains once:

async def deliver_notifications(
    ctx: ExecutionContext,
    senders: RecordingSenders,
) -> int:
    # Relay staged events to the queue, then let a QueueConsumer drain it. The consumer
    # deduplicates redeliveries through the inbox (so an at-least-once queue cannot
    # double-send) and parks poison messages — instead of a hand-rolled receive/ack loop
    # that would re-send on every redelivery. In production this is a background
    # lifecycle step (notification_consumer_lifecycle_step); here we drain once.
    await OutboxRelay(outbox_spec=NOTIFY_EVENTS).to_queue(ctx, NOTIFICATIONS)

    consumer = QueueConsumer(
        queue="notifications",
        queue_spec=NOTIFICATIONS,
        handler=notification_queue_consumer_handler(
            router=router,
            senders=senders,  # pyright: ignore[reportArgumentType]
        ),
        inbox_spec=NOTIFY_INBOX,
        tx_route="mock",
    )

    # A finite idle timeout ends the drain once the queue goes quiet.
    await consumer.run(ctx, timeout=timedelta(milliseconds=250))

    # Report notifications actually dispatched (one message can map to several), not the
    # number of queue messages processed.
    return len(senders.emails)

Notes¶

The dedup key is the relayed event's deterministic id (forze_event_id header, else the broker message.id), so a redelivered message is processed once even though delivery is at-least-once. key is never used — it is the ordering (grouping) key, shared by every event of one aggregate.
notification_queue_consumer_handler adapts process_notification_message (which takes a QueueMessage, not a raw payload) to the consumer's handler signature.
Unmapped event types are skipped by default (skip_unmapped=True).
The producer and consumer are decoupled: they can be different processes, and the consumer is just a queue worker.