Leopoldadmin/leocrm
1
0
Fork 0
Code Issues 363 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

phase6: standardized event envelope (aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version) + outbox_deliveries table

Browse Source
This commit is contained in:
Agent Zero
2026-07-29 22:50:27 +02:00
parent 0fb0ca9925
commit 54c275580f
2 changed files with 113 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alembic/versions/0075_outbox_envelope.py
+80
View File
@@ -0,0 +1,80 @@
"""Add outbox_deliveries table and envelope columns to event_outbox.
Standardized Event-Envelope:
- event_id (already exists as id)
- event_type (already exists as event_name)
- tenant_id (already exists)
- aggregate_type (NEW)
- aggregate_id (NEW)
- occurred_at (NEW)
- correlation_id (NEW)
- schema_version (NEW, default 1)
- payload (already exists)
outbox_deliveries tracks per-consumer delivery status.
An event is only 'published' when all mandatory deliveries succeed.
Revision ID: 0075
Revises: 0074
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID as PGUUID
revision = "0075"
down_revision = "0074"
branch_labels = None
depends_on = None
def upgrade() -> None:
# 1. Add envelope columns to event_outbox
op.add_column("event_outbox", sa.Column("aggregate_type", sa.String(100), nullable=True))
op.add_column("event_outbox", sa.Column("aggregate_id", PGUUID(as_uuid=True), nullable=True))
op.add_column("event_outbox", sa.Column("occurred_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()"), nullable=False))
op.add_column("event_outbox", sa.Column("correlation_id", PGUUID(as_uuid=True), nullable=True))
op.add_column("event_outbox", sa.Column("schema_version", sa.Integer, nullable=False, server_default=sa.text("1")))
op.execute("CREATE INDEX IF NOT EXISTS ix_event_outbox_aggregate ON event_outbox (tenant_id, aggregate_type, aggregate_id)")
op.execute("CREATE INDEX IF NOT EXISTS ix_event_outbox_correlation ON event_outbox (correlation_id)")
# 2. Create outbox_deliveries table
op.create_table(
"outbox_deliveries",
sa.Column("id", PGUUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
sa.Column("event_id", PGUUID(as_uuid=True), sa.ForeignKey("event_outbox.id", ondelete="CASCADE"), nullable=False),
sa.Column("consumer_name", sa.String(150), nullable=False),
sa.Column("status", sa.String(30), nullable=False, server_default="pending"),
sa.Column("attempt_count", sa.Integer, nullable=False, server_default=sa.text("0")),
sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_error", sa.Text, nullable=True),
sa.Column("processed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()"), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()"), nullable=False),
sa.UniqueConstraint("event_id", "consumer_name", name="uq_outbox_deliveries_event_consumer"),
)
op.create_index("ix_outbox_deliveries_event", "outbox_deliveries", ["event_id"])
op.create_index("ix_outbox_deliveries_status", "outbox_deliveries", ["status", "next_attempt_at"])
# RLS + Grants
op.execute("ALTER TABLE outbox_deliveries ENABLE ROW LEVEL SECURITY")
op.execute(
"CREATE POLICY outbox_deliveries_tenant_isolation ON outbox_deliveries "
"FOR ALL "
"USING (EXISTS (SELECT 1 FROM event_outbox WHERE event_outbox.id = outbox_deliveries.event_id AND event_outbox.tenant_id = current_setting('app.current_tenant_id', true)::uuid)) "
"WITH CHECK (EXISTS (SELECT 1 FROM event_outbox WHERE event_outbox.id = outbox_deliveries.event_id AND event_outbox.tenant_id = current_setting('app.current_tenant_id', true)::uuid))"
)
op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON outbox_deliveries TO crm_api, crm_worker")
def downgrade() -> None:
op.execute("DROP POLICY IF EXISTS outbox_deliveries_tenant_isolation ON outbox_deliveries")
op.drop_table("outbox_deliveries")
op.execute("DROP INDEX IF EXISTS ix_event_outbox_correlation")
op.execute("DROP INDEX IF EXISTS ix_event_outbox_aggregate")
op.drop_column("event_outbox", "schema_version")
op.drop_column("event_outbox", "correlation_id")
op.drop_column("event_outbox", "occurred_at")
op.drop_column("event_outbox", "aggregate_id")
op.drop_column("event_outbox", "aggregate_type")
app/core/outbox.py
+33 -5
View File
@@ -34,8 +34,9 @@ logger = logging.getLogger(__name__)
_INSERT_SQL = text(
"""
INSERT INTO event_outbox (tenant_id, event_name, payload)
VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB))
INSERT INTO event_outbox (tenant_id, event_name, payload, aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version)
VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB), :aggregate_type, :aggregate_id, COALESCE(:occurred_at, now()), :correlation_id, COALESCE(:schema_version, 1))
RETURNING id
"""
)
@@ -52,7 +53,8 @@ _CLAIM_SQL = text(
LIMIT :batch_size
FOR UPDATE SKIP LOCKED
)
RETURNING id, tenant_id, event_name, payload, attempts, max_attempts
RETURNING id, tenant_id, event_name, payload, attempts, max_attempts,
aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version
"""
)
@@ -99,6 +101,11 @@ async def enqueue_outbox_event(
tenant_id: uuid.UUID,
event_name: str,
payload: dict[str, Any],
*,
aggregate_type: str | None = None,
aggregate_id: uuid.UUID | None = None,
correlation_id: uuid.UUID | None = None,
schema_version: int = 1,
) -> None:
"""Insert an event into the outbox table within the current transaction.
@@ -110,8 +117,12 @@ async def enqueue_outbox_event(
Args:
db: Active async SQLAlchemy session (part of the business transaction).
tenant_id: Tenant scope for the event.
event_name: Logical event name (e.g. ``"contact.created"``).
event_name: Logical event name (e.g. ``"crm.contact.created.v1"``).
payload: Event payload dict (will be stored as JSONB).
aggregate_type: Type of the aggregate (e.g. 'contact', 'task').
aggregate_id: UUID of the aggregate entity.
correlation_id: Optional correlation UUID for tracing across services.
schema_version: Event schema version (default 1).
"""
await db.execute(
_INSERT_SQL,
@@ -119,6 +130,11 @@ async def enqueue_outbox_event(
"tenant_id": str(tenant_id),
"event_name": event_name,
"payload": _json_payload(payload),
"aggregate_type": aggregate_type,
"aggregate_id": str(aggregate_id) if aggregate_id else None,
"occurred_at": None, # DB defaults to NOW()
"correlation_id": str(correlation_id) if correlation_id else None,
"schema_version": schema_version,
},
)
@@ -161,10 +177,16 @@ async def process_outbox_batch(
for row in rows:
event_id = row[0]
tenant_id = row[1]
event_name = row[2]
payload = row[3]
attempts = row[4]
max_attempts = row[5]
aggregate_type = row[6] if len(row) > 6 else None
aggregate_id = row[7] if len(row) > 7 else None
occurred_at = row[8] if len(row) > 8 else None
correlation_id = row[9] if len(row) > 9 else None
schema_version = row[10] if len(row) > 10 else 1
# payload comes back as a dict from JSONB
if isinstance(payload, str):
@@ -174,10 +196,16 @@ async def process_outbox_batch(
payload_dict = payload
try:
# Enrich payload with event metadata for idempotency
# Enrich payload with standardized event envelope metadata
payload_dict.setdefault("_event_id", str(event_id))
payload_dict.setdefault("_event_name", event_name)
payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat())
payload_dict.setdefault("_tenant_id", str(tenant_id))
payload_dict.setdefault("_aggregate_type", aggregate_type)
payload_dict.setdefault("_aggregate_id", str(aggregate_id) if aggregate_id else None)
payload_dict.setdefault("_occurred_at", occurred_at.isoformat() if occurred_at else None)
payload_dict.setdefault("_correlation_id", str(correlation_id) if correlation_id else None)
payload_dict.setdefault("_schema_version", schema_version)
# Idempotency check: has this event already been processed? (P1.5 fix)
already_processed = await db.execute(