Reliable event publishing with the transactional outbox pattern

# ======================================================================
# Q: How do you reliably send an event when something changes in the database
# (e.g. "new critical alert")?
#
# With a transactional outbox. If I commit and then publish, and the publish
# fails, the event is lost. If I publish before the commit and the commit fails,
# I send an event for something that did not happen. That is why I write the
# event to an outbox table in the same transaction, and a relay worker sends it
# to the broker.
# Delivery is at-least-once, so consumers must be idempotent (dedup by event
# id). SKIP LOCKED is the classic pattern for a job queue inside PostgreSQL.
# ======================================================================

with session.begin():
    alert = Alert(tenant_id=tid, severity='critical', asset_id=aid)
    session.add(alert)
    session.flush()
    session.add(Outbox(topic='alerts.created',
                       key=str(alert.id),
                       payload={'alert_id': alert.id, 'tenant_id': tid}))

# relay worker (Celery beat every few seconds or a separate process)
def relay_outbox(session, batch=500):
    with session.begin():
        rows = session.scalars(
            select(Outbox).where(Outbox.sent_at.is_(None))
            .order_by(Outbox.id).limit(batch)
            .with_for_update(skip_locked=True)).all()   # many workers without conflicts
        for row in rows:
            broker.publish(row.topic, key=row.key, value=row.payload)
            row.sent_at = func.now()