Transactional Outbox.

async def create_order(order_data):
    # Step 1: Save order to database
    order = await db.orders.insert(order_data)
    
    # Step 2: Publish event for downstream services
    await kafka.publish("order.created", order)
    
    return order

# Single database - transactions work!
async with db.transaction():
    await db.execute("INSERT INTO orders ...")
    await db.execute("INSERT INTO order_items ...")
    await db.execute("UPDATE inventory ...")
# All or nothing - atomic!

# Two different systems - no shared transaction!
await db.execute("INSERT INTO orders ...")  # PostgreSQL
await kafka.publish("order.created", ...)   # Kafka

# These are TWO separate operations
# No way to make them atomic

async def create_order(order_data):
    async with db.transaction():
        # Write business data
        order = await db.execute("""
            INSERT INTO orders (user_id, amount, status)
            VALUES ($1, $2, 'created')
            RETURNING *
        """, order_data.user_id, order_data.amount)
        
        # Write event to outbox (same transaction!)
        await db.execute("""
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
            VALUES ('Order', $1, 'OrderCreated', $2)
        """, order.id, json.dumps({
            'order_id': order.id,
            'user_id': order.user_id,
            'amount': str(order.amount),
            'created_at': order.created_at.isoformat()
        }))
        
        return order
    
    # If transaction commits, BOTH writes succeed
    # If transaction fails, NEITHER write happens

async def outbox_publisher():
    while True:
        async with db.transaction():
            # Get unpublished events
            events = await db.fetch("""
                SELECT * FROM outbox 
                WHERE published_at IS NULL
                ORDER BY created_at
                LIMIT 100
                FOR UPDATE SKIP LOCKED
            """)
            
            for event in events:
                # Publish to Kafka
                await kafka.publish(
                    topic=f"{event.aggregate_type.lower()}.events",
                    key=event.aggregate_id,
                    value=event.payload
                )
                
                # Mark as published
                await db.execute("""
                    UPDATE outbox SET published_at = NOW()
                    WHERE id = $1
                """, event.id)
        
        await asyncio.sleep(1)  # Poll interval

-- Outbox table schema
CREATE TABLE outbox (
    id              BIGSERIAL PRIMARY KEY,
    aggregate_type  VARCHAR(255) NOT NULL,    -- e.g., 'Order', 'User', 'Payment'
    aggregate_id    VARCHAR(255) NOT NULL,    -- e.g., order ID, user ID
    event_type      VARCHAR(255) NOT NULL,    -- e.g., 'OrderCreated', 'OrderPaid'
    payload         JSONB NOT NULL,           -- Event data
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMP NULL            -- NULL = not yet published
);

-- Partial index for efficient polling (PostgreSQL)
CREATE INDEX idx_outbox_unpublished ON outbox (created_at) WHERE published_at IS NULL;

-- Optional: Partition by time for easier cleanup
-- CREATE TABLE outbox (...) PARTITION BY RANGE (created_at);

-- Option 1: Delete after publishing (simple, no history)
DELETE FROM outbox WHERE published_at IS NOT NULL;

-- Option 2: Delete old published events (keep recent for debugging)
DELETE FROM outbox 
WHERE published_at IS NOT NULL 
AND published_at < NOW() - INTERVAL '7 days';

-- Option 3: Partition by time, drop old partitions (efficient at scale)
-- Requires table partitioning setup (PostgreSQL: a partition is a table)
DROP TABLE outbox_2024_01;

# Basic Transactional Outbox Implementation
# WARNING: Not production-ready - for learning only

import asyncio
import json
from datetime import datetime
from typing import Any, Dict, Optional
from dataclasses import dataclass

import asyncpg
from aiokafka import AIOKafkaProducer


@dataclass
class OutboxEvent:
    """Represents an event in the outbox."""
    id: int
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: Dict[str, Any]
    created_at: datetime


class SimpleOutbox:
    """
    Basic transactional outbox implementation.
    
    Usage:
        async with db.transaction():
            order = await create_order(...)
            await outbox.add_event(
                aggregate_type="Order",
                aggregate_id=str(order.id),
                event_type="OrderCreated",
                payload={"order_id": order.id, ...}
            )
    """
    
    def __init__(self, db_pool: asyncpg.Pool):
        self.db = db_pool
    
    async def add_event(
        self,
        aggregate_type: str,
        aggregate_id: str,
        event_type: str,
        payload: Dict[str, Any],
        connection: Optional[asyncpg.Connection] = None
    ) -> int:
        """
        Add an event to the outbox.
        
        IMPORTANT: Call this within the same transaction as your business logic!
        """
        conn = connection or await self.db.acquire()
        try:
            result = await conn.fetchval("""
                INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
                VALUES ($1, $2, $3, $4)
                RETURNING id
            """, aggregate_type, aggregate_id, event_type, json.dumps(payload))
            return result
        finally:
            if connection is None:
                await self.db.release(conn)


class SimpleOutboxPublisher:
    """
    Basic polling-based outbox publisher.
    """
    
    def __init__(
        self,
        db_pool: asyncpg.Pool,
        kafka_producer: AIOKafkaProducer,
        poll_interval: float = 1.0,
        batch_size: int = 100
    ):
        self.db = db_pool
        self.kafka = kafka_producer
        self.poll_interval = poll_interval
        self.batch_size = batch_size
        self._running = False
    
    async def start(self):
        """Start the publisher loop."""
        self._running = True
        while self._running:
            try:
                published = await self._publish_batch()
                if published == 0:
                    # No events, wait before next poll
                    await asyncio.sleep(self.poll_interval)
            except Exception as e:
                print(f"Publisher error: {e}")
                await asyncio.sleep(self.poll_interval)
    
    async def stop(self):
        """Stop the publisher."""
        self._running = False
    
    async def _publish_batch(self) -> int:
        """Publish a batch of events. Returns count published."""
        async with self.db.acquire() as conn:
            async with conn.transaction():
                # Lock and fetch unpublished events
                events = await conn.fetch("""
                    SELECT id, aggregate_type, aggregate_id, event_type, payload, created_at
                    FROM outbox
                    WHERE published_at IS NULL
                    ORDER BY created_at
                    LIMIT $1
                    FOR UPDATE SKIP LOCKED
                """, self.batch_size)
                
                if not events:
                    return 0
                
                # Publish each event
                for event in events:
                    topic = f"{event['aggregate_type'].lower()}.events"
                    await self.kafka.send_and_wait(
                        topic,
                        key=event['aggregate_id'].encode(),
                        value=event['payload'].encode()
                    )
                
                # Mark all as published
                event_ids = [e['id'] for e in events]
                await conn.execute("""
                    UPDATE outbox
                    SET published_at = NOW()
                    WHERE id = ANY($1)
                """, event_ids)
                
                return len(events)


# Usage example
async def create_order_example(db: asyncpg.Pool, outbox: SimpleOutbox):
    """Example of creating an order with outbox event."""
    
    async with db.acquire() as conn:
        async with conn.transaction():
            # Create the order
            order = await conn.fetchrow("""
                INSERT INTO orders (user_id, amount, status)
                VALUES ($1, $2, 'created')
                RETURNING id, user_id, amount, status, created_at
            """, "user_123", 99.99)
            
            # Add event to outbox (SAME TRANSACTION!)
            await outbox.add_event(
                aggregate_type="Order",
                aggregate_id=str(order['id']),
                event_type="OrderCreated",
                payload={
                    "order_id": order['id'],
                    "user_id": order['user_id'],
                    "amount": str(order['amount']),
                    "status": order['status'],
                    "created_at": order['created_at'].isoformat()
                },
                connection=conn  # Use same connection!
            )
            
            return dict(order)

# Production-Ready Transactional Outbox Implementation

import asyncio
import json
import signal
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List, Callable
from enum import Enum
import logging
from contextlib import asynccontextmanager
import hashlib

import asyncpg
from aiokafka import AIOKafkaProducer

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class OutboxConfig:
    """Configuration for the transactional outbox."""
    
    # Database settings
    db_dsn: str
    db_pool_min: int = 5
    db_pool_max: int = 20
    
    # Kafka settings
    kafka_bootstrap_servers: str = "localhost:9092"
    kafka_acks: str = "all"
    
    # Publisher settings
    poll_interval_ms: int = 100          # How often to poll (ms)
    batch_size: int = 100                 # Events per batch
    max_publish_attempts: int = 5         # Before giving up on an event
    lock_timeout_seconds: int = 300       # Stale lock detection
    
    # Cleanup settings
    retention_hours: int = 168            # 7 days
    cleanup_batch_size: int = 1000


@dataclass
class OutboxEvent:
    """An event stored in the outbox."""
    id: int                    # outbox row id (serial, used to mark the row)
    event_id: str              # writer's deterministic idempotency key
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: Dict[str, Any]
    created_at: datetime
    publish_attempts: int = 0
    last_error: Optional[str] = None


@dataclass
class PublishResult:
    """Result of publishing an event."""
    event_id: int
    success: bool
    topic: str
    partition: Optional[int] = None
    offset: Optional[int] = None
    error: Optional[str] = None


# =============================================================================
# Outbox Writer (used by application code)
# =============================================================================

class OutboxWriter:
    """
    Writes events to the outbox table.
    
    Usage:
        async with db.transaction() as conn:
            # Your business logic
            order = await create_order(conn, order_data)
            
            # Write to outbox in same transaction
            await outbox.write(
                conn,
                aggregate_type="Order",
                aggregate_id=str(order.id),
                event_type="OrderCreated",
                payload=order.to_event_payload()
            )
    """
    
    def __init__(self, config: OutboxConfig):
        self.config = config
    
    async def write(
        self,
        connection: asyncpg.Connection,
        aggregate_type: str,
        aggregate_id: str,
        event_type: str,
        payload: Dict[str, Any],
        event_id: Optional[str] = None
    ) -> int:
        """
        Write an event to the outbox.
        
        Args:
            connection: Database connection (must be in a transaction!)
            aggregate_type: Type of entity (e.g., "Order", "User")
            aggregate_id: ID of the entity
            event_type: Type of event (e.g., "OrderCreated")
            payload: Event data
            event_id: Optional idempotency key for the event
            
        Returns:
            The outbox event ID
        """
        # Generate event_id if not provided
        if event_id is None:
            event_id = self._generate_event_id(aggregate_type, aggregate_id, event_type, payload)
        
        result = await connection.fetchval("""
            INSERT INTO outbox (
                event_id, aggregate_type, aggregate_id, event_type, payload
            )
            VALUES ($1, $2, $3, $4, $5)
            ON CONFLICT (event_id) DO NOTHING
            RETURNING id
        """, event_id, aggregate_type, aggregate_id, event_type, json.dumps(payload))
        
        if result is None:
            # Event already exists (idempotency)
            result = await connection.fetchval(
                "SELECT id FROM outbox WHERE event_id = $1", event_id
            )
            logger.debug(f"Event {event_id} already exists, returning existing ID {result}")
        
        return result
    
    async def write_batch(
        self,
        connection: asyncpg.Connection,
        events: List[Dict[str, Any]]
    ) -> List[int]:
        """
        Write multiple events to the outbox.
        
        Args:
            connection: Database connection
            events: List of event dicts with aggregate_type, aggregate_id, event_type, payload
            
        Returns:
            List of outbox event IDs
        """
        ids = []
        for event in events:
            event_id = await self.write(
                connection,
                aggregate_type=event['aggregate_type'],
                aggregate_id=event['aggregate_id'],
                event_type=event['event_type'],
                payload=event['payload'],
                event_id=event.get('event_id')
            )
            ids.append(event_id)
        return ids
    
    def _generate_event_id(
        self,
        aggregate_type: str,
        aggregate_id: str,
        event_type: str,
        payload: Dict[str, Any]
    ) -> str:
        """Generate a deterministic event ID for idempotency."""
        content = f"{aggregate_type}:{aggregate_id}:{event_type}:{json.dumps(payload, sort_keys=True)}"
        return hashlib.sha256(content.encode()).hexdigest()[:32]


# =============================================================================
# Outbox Publisher (background process)
# =============================================================================

class OutboxPublisher:
    """
    Publishes events from the outbox to Kafka.
    
    Features:
    - Batch publishing for throughput
    - At-least-once publishing; consumers dedupe on event_id
    - Automatic retry with backoff
    - Stale lock detection
    - Graceful shutdown
    - Metrics and monitoring
    """
    
    def __init__(
        self,
        config: OutboxConfig,
        topic_resolver: Optional[Callable[[OutboxEvent], str]] = None
    ):
        self.config = config
        self.topic_resolver = topic_resolver or self._default_topic_resolver
        
        self.db_pool: Optional[asyncpg.Pool] = None
        self.kafka: Optional[AIOKafkaProducer] = None
        
        self._running = False
        self._shutdown_event = asyncio.Event()
        
        # Metrics
        self.events_published = 0
        self.events_failed = 0
        self.batches_processed = 0
        self.last_publish_time: Optional[datetime] = None
    
    async def start(self) -> None:
        """Initialize connections and start publishing."""
        logger.info("Starting outbox publisher...")
        
        # Initialize database pool
        self.db_pool = await asyncpg.create_pool(
            self.config.db_dsn,
            min_size=self.config.db_pool_min,
            max_size=self.config.db_pool_max
        )
        
        # Initialize Kafka producer
        self.kafka = AIOKafkaProducer(
            bootstrap_servers=self.config.kafka_bootstrap_servers,
            acks=self.config.kafka_acks,
            enable_idempotence=True  # Kafka-level idempotence
        )
        await self.kafka.start()
        
        self._running = True
        logger.info("Outbox publisher started")
    
    async def stop(self) -> None:
        """Gracefully stop the publisher."""
        logger.info("Stopping outbox publisher...")
        self._running = False
        self._shutdown_event.set()
        
        # Wait for current batch to complete
        await asyncio.sleep(0.5)
        
        if self.kafka:
            await self.kafka.stop()
        
        if self.db_pool:
            await self.db_pool.close()
        
        logger.info(
            f"Outbox publisher stopped. "
            f"Published: {self.events_published}, Failed: {self.events_failed}"
        )
    
    async def run(self) -> None:
        """Main publishing loop."""
        while self._running:
            try:
                published = await self._process_batch()
                
                if published == 0:
                    # No events to process, wait before next poll
                    try:
                        await asyncio.wait_for(
                            self._shutdown_event.wait(),
                            timeout=self.config.poll_interval_ms / 1000
                        )
                    except asyncio.TimeoutError:
                        pass  # Normal timeout, continue polling
                else:
                    self.batches_processed += 1
                    
            except asyncio.CancelledError:
                logger.info("Publisher loop cancelled")
                break
            except Exception as e:
                logger.error(f"Publisher error: {e}", exc_info=True)
                await asyncio.sleep(1)  # Back off on error
    
    async def _process_batch(self) -> int:
        """Process a batch of events. Returns count published."""
        async with self.db_pool.acquire() as conn:
            # Use advisory lock for distributed coordination
            lock_acquired = await conn.fetchval(
                "SELECT pg_try_advisory_lock(hashtext('outbox_publisher'))"
            )
            
            if not lock_acquired:
                # Another publisher has the lock
                return 0
            
            try:
                return await self._process_batch_with_lock(conn)
            finally:
                await conn.execute(
                    "SELECT pg_advisory_unlock(hashtext('outbox_publisher'))"
                )
    
    async def _process_batch_with_lock(self, conn: asyncpg.Connection) -> int:
        """Process batch while holding the lock."""
        async with conn.transaction():
            # Fetch unpublished events
            events = await conn.fetch("""
                SELECT 
                    id, event_id, aggregate_type, aggregate_id, 
                    event_type, payload, created_at, publish_attempts
                FROM outbox
                WHERE published_at IS NULL
                AND publish_attempts < $1
                ORDER BY created_at
                LIMIT $2
                FOR UPDATE SKIP LOCKED
            """, self.config.max_publish_attempts, self.config.batch_size)
            
            if not events:
                return 0
            
            # Process each event
            published_ids = []
            failed_updates = []
            
            for row in events:
                event = OutboxEvent(
                    id=row['id'],
                    event_id=row['event_id'],
                    aggregate_type=row['aggregate_type'],
                    aggregate_id=row['aggregate_id'],
                    event_type=row['event_type'],
                    payload=json.loads(row['payload']),
                    created_at=row['created_at'],
                    publish_attempts=row['publish_attempts']
                )
                
                result = await self._publish_event(event)
                
                if result.success:
                    published_ids.append(event.id)
                    self.events_published += 1
                else:
                    failed_updates.append((event.id, result.error))
                    self.events_failed += 1
            
            # Mark successful events as published
            if published_ids:
                await conn.execute("""
                    UPDATE outbox
                    SET published_at = NOW()
                    WHERE id = ANY($1)
                """, published_ids)
            
            # Update failed events (increment attempts, record error)
            for event_id, error in failed_updates:
                await conn.execute("""
                    UPDATE outbox
                    SET 
                        publish_attempts = publish_attempts + 1,
                        last_error = $2,
                        last_attempt_at = NOW()
                    WHERE id = $1
                """, event_id, error[:1000])  # Truncate error
            
            self.last_publish_time = datetime.now(timezone.utc)
            return len(published_ids)
    
    async def _publish_event(self, event: OutboxEvent) -> PublishResult:
        """Publish a single event to Kafka."""
        topic = self.topic_resolver(event)
        
        try:
            # Serialize payload
            value = json.dumps({
                'event_id': event.event_id,  # dedupe key, not the serial row id
                'event_type': event.event_type,
                'aggregate_type': event.aggregate_type,
                'aggregate_id': event.aggregate_id,
                'payload': event.payload,
                'created_at': event.created_at.isoformat()
            }).encode('utf-8')
            
            key = event.aggregate_id.encode('utf-8')
            
            # Send to Kafka
            result = await self.kafka.send_and_wait(topic, value=value, key=key)
            
            logger.debug(
                f"Published event {event.id} to {topic}:{result.partition}@{result.offset}"
            )
            
            return PublishResult(
                event_id=event.id,
                success=True,
                topic=topic,
                partition=result.partition,
                offset=result.offset
            )
            
        except Exception as e:
            logger.warning(f"Failed to publish event {event.id}: {e}")
            return PublishResult(
                event_id=event.id,
                success=False,
                topic=topic,
                error=str(e)
            )
    
    def _default_topic_resolver(self, event: OutboxEvent) -> str:
        """Default topic naming: lowercase aggregate type + .events"""
        return f"{event.aggregate_type.lower()}.events"


# =============================================================================
# Outbox Cleaner (periodic cleanup)
# =============================================================================

class OutboxCleaner:
    """Cleans up old published events from the outbox."""
    
    def __init__(self, db_pool: asyncpg.Pool, config: OutboxConfig):
        self.db = db_pool
        self.config = config
        self._running = False
    
    async def run_once(self) -> int:
        """Run a single cleanup pass. Returns number of deleted rows."""
        cutoff = datetime.now(timezone.utc) - timedelta(hours=self.config.retention_hours)
        
        async with self.db.acquire() as conn:
            # PostgreSQL DELETE has no LIMIT; batch via a subquery
            result = await conn.execute("""
                DELETE FROM outbox
                WHERE id IN (
                    SELECT id FROM outbox
                    WHERE published_at IS NOT NULL
                    AND published_at < $1
                    LIMIT $2
                )
            """, cutoff, self.config.cleanup_batch_size)
            
            deleted = int(result.split()[-1])
            if deleted > 0:
                logger.info(f"Cleaned up {deleted} old outbox events")
            return deleted
    
    async def run(self, interval_seconds: int = 3600):
        """Run cleanup periodically."""
        self._running = True
        while self._running:
            try:
                await self.run_once()
            except Exception as e:
                logger.error(f"Cleanup error: {e}")
            
            await asyncio.sleep(interval_seconds)
    
    def stop(self):
        self._running = False


# =============================================================================
# Database Schema
# =============================================================================

OUTBOX_SCHEMA = """
-- Outbox table for transactional outbox pattern
CREATE TABLE IF NOT EXISTS outbox (
    id              BIGSERIAL PRIMARY KEY,
    event_id        VARCHAR(64) NOT NULL UNIQUE,  -- Idempotency key
    aggregate_type  VARCHAR(255) NOT NULL,
    aggregate_id    VARCHAR(255) NOT NULL,
    event_type      VARCHAR(255) NOT NULL,
    payload         JSONB NOT NULL,
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMP NULL,
    publish_attempts INTEGER NOT NULL DEFAULT 0,
    last_attempt_at TIMESTAMP NULL,
    last_error      TEXT NULL
);

-- Index for efficient polling of unpublished events
CREATE INDEX IF NOT EXISTS idx_outbox_unpublished 
ON outbox (created_at) 
WHERE published_at IS NULL;

-- Index for cleanup of old published events
CREATE INDEX IF NOT EXISTS idx_outbox_published 
ON outbox (published_at) 
WHERE published_at IS NOT NULL;

-- Index for event_id lookups (idempotency checks)
CREATE INDEX IF NOT EXISTS idx_outbox_event_id 
ON outbox (event_id);
"""


# =============================================================================
# Example Usage
# =============================================================================

async def example_usage():
    """Complete example of using the transactional outbox."""
    
    config = OutboxConfig(
        db_dsn="postgresql://user:pass@localhost/mydb",
        kafka_bootstrap_servers="localhost:9092"
    )
    
    # Initialize components
    db_pool = await asyncpg.create_pool(config.db_dsn)
    outbox_writer = OutboxWriter(config)
    outbox_publisher = OutboxPublisher(config)
    
    # Start publisher in background
    await outbox_publisher.start()
    publisher_task = asyncio.create_task(outbox_publisher.run())
    
    try:
        # Application code: Create an order with outbox event
        async with db_pool.acquire() as conn:
            async with conn.transaction():
                # Business logic
                order = await conn.fetchrow("""
                    INSERT INTO orders (user_id, amount, status)
                    VALUES ($1, $2, 'created')
                    RETURNING *
                """, "user_123", 99.99)
                
                # Write to outbox (same transaction!)
                await outbox_writer.write(
                    conn,
                    aggregate_type="Order",
                    aggregate_id=str(order['id']),
                    event_type="OrderCreated",
                    payload={
                        "order_id": order['id'],
                        "user_id": order['user_id'],
                        "amount": float(order['amount']),
                        "status": order['status']
                    }
                )
                
                print(f"Created order {order['id']}")
        
        # Wait for event to be published
        await asyncio.sleep(2)
        
    finally:
        # Cleanup
        await outbox_publisher.stop()
        publisher_task.cancel()
        await db_pool.close()


if __name__ == "__main__":
    asyncio.run(example_usage())

# docker-compose.yml for Debezium CDC
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.1
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.0.1
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092  # containers reach the broker as kafka:9092; add a second listener for host access
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  postgres:
    image: postgres:14
    ports:
      - "5432:5432"
    environment:
      POSTGRES_PASSWORD: postgres
    command: 
      - "postgres"
      - "-c"
      - "wal_level=logical"  # Required for CDC

  debezium:
    image: debezium/connect:2.0
    depends_on:
      - kafka
      - postgres
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: debezium_configs
      OFFSET_STORAGE_TOPIC: debezium_offsets
      STATUS_STORAGE_TOPIC: debezium_status

// Debezium connector configuration
// POST to http://localhost:8083/connectors
{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "mydb",
    "plugin.name": "pgoutput",
    "topic.prefix": "myapp",
    
    "table.include.list": "public.outbox",
    
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "event_id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.type": "event_type",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "${routedByValue}.events"
  }
}

-- Create partitioned outbox table
CREATE TABLE outbox (
    id              BIGSERIAL,
    event_id        VARCHAR(64) NOT NULL,
    aggregate_type  VARCHAR(255) NOT NULL,
    aggregate_id    VARCHAR(255) NOT NULL,
    event_type      VARCHAR(255) NOT NULL,
    payload         JSONB NOT NULL,
    created_at      TIMESTAMP NOT NULL DEFAULT NOW(),
    published_at    TIMESTAMP NULL,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);

-- Create partitions for each day
CREATE TABLE outbox_2024_01_15 PARTITION OF outbox
    FOR VALUES FROM ('2024-01-15') TO ('2024-01-16');
    
CREATE TABLE outbox_2024_01_16 PARTITION OF outbox
    FOR VALUES FROM ('2024-01-16') TO ('2024-01-17');

-- Drop old partitions (instant, no row-by-row delete)
DROP TABLE outbox_2024_01_08;  -- Drop week-old partition