diff --git a/alembic/versions/0075_outbox_envelope.py b/alembic/versions/0075_outbox_envelope.py new file mode 100644 index 0000000..229bf23 --- /dev/null +++ b/alembic/versions/0075_outbox_envelope.py @@ -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") diff --git a/app/core/outbox.py b/app/core/outbox.py index a8a7296..6966004 100644 --- a/app/core/outbox.py +++ b/app/core/outbox.py @@ -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(