78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
|
"""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"
|
||
|
|
)
|