72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
|
"""Create event_outbox table for transactional outbox pattern.
|
||
|
|
|
||
|
|
Revision ID: 0040_outbox
|
||
|
|
Revises: 0039_contact_normalize
|
||
|
|
Create Date: 2026-07-25
|
||
|
|
|
||
|
|
Stores domain events in a durable table so they survive process crashes,
|
||
|
|
restarts, and multi-replica deployments. A background worker polls the
|
||
|
|
outbox and publishes events to the in-process event bus.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Union
|
||
|
|
|
||
|
|
import sqlalchemy as sa
|
||
|
|
from alembic import op
|
||
|
|
|
||
|
|
# revision identifiers
|
||
|
|
revision: str = "0040_outbox"
|
||
|
|
down_revision: Union[str, None] = "0039_contact_normalize"
|
||
|
|
branch_labels: Union[str, None] = None
|
||
|
|
depends_on: Union[str, None] = None
|
||
|
|
|
||
|
|
|
||
|
|
def upgrade() -> None:
|
||
|
|
conn = op.get_bind()
|
||
|
|
|
||
|
|
# Ensure pgcrypto extension for gen_random_uuid()
|
||
|
|
conn.execute(sa.text("CREATE EXTENSION IF NOT EXISTS pgcrypto"))
|
||
|
|
|
||
|
|
conn.execute(
|
||
|
|
sa.text(
|
||
|
|
"""
|
||
|
|
CREATE TABLE IF NOT EXISTS event_outbox (
|
||
|
|
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
|
|
tenant_id UUID NOT NULL,
|
||
|
|
event_name VARCHAR(255) NOT NULL,
|
||
|
|
payload JSONB NOT NULL,
|
||
|
|
status VARCHAR(20) NOT NULL DEFAULT 'pending',
|
||
|
|
attempts INT NOT NULL DEFAULT 0,
|
||
|
|
max_attempts INT NOT NULL DEFAULT 5,
|
||
|
|
next_retry_at TIMESTAMPTZ,
|
||
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
|
|
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
|
|
published_at TIMESTAMPTZ
|
||
|
|
)
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
# Index for the worker query: WHERE status = 'pending' ORDER BY next_retry_at
|
||
|
|
conn.execute(
|
||
|
|
sa.text(
|
||
|
|
"CREATE INDEX IF NOT EXISTS ix_outbox_status "
|
||
|
|
"ON event_outbox (status, next_retry_at)"
|
||
|
|
)
|
||
|
|
)
|
||
|
|
conn.execute(
|
||
|
|
sa.text(
|
||
|
|
"CREATE INDEX IF NOT EXISTS ix_outbox_tenant "
|
||
|
|
"ON event_outbox (tenant_id)"
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def downgrade() -> None:
|
||
|
|
conn = op.get_bind()
|
||
|
|
conn.execute(sa.text("DROP INDEX IF EXISTS ix_outbox_tenant"))
|
||
|
|
conn.execute(sa.text("DROP INDEX IF EXISTS ix_outbox_status"))
|
||
|
|
conn.execute(sa.text("DROP TABLE IF EXISTS event_outbox"))
|