Phase 5: Outbox DLQ, Monitoring, Consumer-Registry

- Migration 0092: DLQ columns (error_message, failed_at) + consumer_inbox RLS fix - outbox.py: DLQ logic, replay functions, stats, consumer registry - app/routes/outbox.py: 5 API endpoints (stats, failed, replay, replay-all, consumer-registry) - outbox_deliveries tracking per consumer handler - 18/18 tests passing
2026-08-02 23:25:54 +02:00
parent 24cb10a7a2
commit 07a99975ec
12 changed files with 1017 additions and 79 deletions
@@ -0,0 +1,77 @@
+"""Add DLQ columns to event_outbox and fix consumer_inbox RLS policy.
+
+Phase 5: Dead-Letter-Queue support.
+- Adds error_message TEXT and failed_at TIMESTAMPTZ to event_outbox
+- Adds partial index for failed events
+- Fixes consumer_inbox RLS policy (previous 0085 policy referenced
+  tenant_id column which does not exist on consumer_inbox; the
+  correct policy uses the event_id FK to event_outbox.tenant_id)
+
+Revision ID: 0092
+Revises: 0091
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0092"
+down_revision = "0091"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. Add DLQ columns to event_outbox
+    op.execute(
+        "ALTER TABLE IF EXISTS event_outbox "
+        "ADD COLUMN IF NOT EXISTS error_message TEXT"
+    )
+    op.execute(
+        "ALTER TABLE IF EXISTS event_outbox "
+        "ADD COLUMN IF NOT EXISTS failed_at TIMESTAMPTZ"
+    )
+
+    # 2. Partial index for efficient failed-event queries
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_outbox_failed "
+        "ON event_outbox (status, failed_at) WHERE status = 'failed'"
+    )
+
+    # 3. Fix consumer_inbox RLS policy
+    #    Migration 0085 created a policy using tenant_id, but consumer_inbox
+    #    has no tenant_id column.  Drop the broken policy and create one
+    #    that follows the same pattern as outbox_deliveries (0075): use the
+    #    event_id FK to check event_outbox.tenant_id.
+    op.execute(
+        "DROP POLICY IF EXISTS consumer_inbox_tenant_isolation ON consumer_inbox"
+    )
+    op.execute("DROP POLICY IF EXISTS tenant_isolation ON consumer_inbox")
+    op.execute("ALTER TABLE consumer_inbox ENABLE ROW LEVEL SECURITY")
+    op.execute("ALTER TABLE consumer_inbox FORCE ROW LEVEL SECURITY")
+    op.execute(
+        "CREATE POLICY consumer_inbox_tenant_isolation ON consumer_inbox "
+        "FOR ALL TO crm_api, crm_worker "
+        "USING (EXISTS (SELECT 1 FROM event_outbox "
+        "WHERE event_outbox.id = consumer_inbox.event_id "
+        "AND event_outbox.tenant_id = "
+        "NULLIF(current_setting('app.current_tenant_id', true), '')::uuid)) "
+        "WITH CHECK (EXISTS (SELECT 1 FROM event_outbox "
+        "WHERE event_outbox.id = consumer_inbox.event_id "
+        "AND event_outbox.tenant_id = "
+        "NULLIF(current_setting('app.current_tenant_id', true), '')::uuid))"
+    )
+    # Ensure grants are in place
+    op.execute(
+        "GRANT SELECT, INSERT, UPDATE, DELETE ON consumer_inbox TO crm_api, crm_worker"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_outbox_failed")
+    op.execute("ALTER TABLE event_outbox DROP COLUMN IF EXISTS failed_at")
+    op.execute("ALTER TABLE event_outbox DROP COLUMN IF EXISTS error_message")
+    # Restore the broken policy state (consumer_inbox RLS remains enabled)
+    op.execute(
+        "DROP POLICY IF EXISTS consumer_inbox_tenant_isolation ON consumer_inbox"
+    )