CQRS Outbox Polling Failures — Connection Pool Exhaustion

-- ============================================================
-- WRITE SIDE (Command Model) — PostgreSQL
-- Normalized, constraint-heavy, optimized for correctness.
-- This schema is NEVER queried by the UI directly.
-- ============================================================

-- Core customer record — source of truth for identity
CREATE TABLE customers (
    customer_id     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email           TEXT NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    -- Enforce unique emails at the DB level, not just the app layer
    CONSTRAINT uq_customers_email UNIQUE (email)
);

-- Orders table — normalized, references customers by FK
CREATE TABLE orders (
    order_id        UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    customer_id     UUID NOT NULL REFERENCES customers(customer_id),
    status          TEXT NOT NULL DEFAULT 'pending',
    total_cents     INTEGER NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    -- Prevent nonsense states at the DB level
    CONSTRAINT chk_orders_status CHECK (status IN ('pending','confirmed','shipped','cancelled')),
    CONSTRAINT chk_orders_total  CHECK (total_cents >= 0)
);

-- Order line items — fully normalized, no denormalized product names here
CREATE TABLE order_items (
    item_id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_id        UUID NOT NULL REFERENCES orders(order_id) ON DELETE CASCADE,
    product_id      UUID NOT NULL,
    quantity        INTEGER NOT NULL,
    unit_price_cents INTEGER NOT NULL,
    CONSTRAINT chk_order_items_qty   CHECK (quantity > 0),
    CONSTRAINT chk_order_items_price CHECK (unit_price_cents > 0)
);

-- ============================================================
-- DOMAIN EVENTS TABLE — the outbox that feeds the read side
-- Every successful command inserts a row here atomically
-- in the same transaction as the state change.
-- ============================================================
CREATE TABLE domain_events (
    event_id        UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type  TEXT NOT NULL,          -- e.g. 'Order'
    aggregate_id    UUID NOT NULL,          -- e.g. the order_id
    event_type      TEXT NOT NULL,          -- e.g. 'OrderConfirmed'
    payload         JSONB NOT NULL,         -- full event data
    occurred_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    published       BOOLEAN NOT NULL DEFAULT FALSE  -- outbox pattern flag
);

-- Index for the outbox poller — only scans unpublished events
CREATE INDEX idx_domain_events_unpublished
    ON domain_events (occurred_at)
    WHERE published = FALSE;

-- ============================================================
-- Example: confirming an order (run as a single transaction)
-- Both the state change AND the event are written atomically.
-- If either fails, both roll back — no lost events.
-- ============================================================
BEGIN;

    -- 1. Update the write-side state
    UPDATE orders
    SET    status     = 'confirmed',
           updated_at = now()
    WHERE  order_id   = 'a1b2c3d4-0000-0000-0000-000000000001'
    AND    status     = 'pending';   -- optimistic guard

    -- 2. Write the domain event in the SAME transaction (Outbox Pattern)
    INSERT INTO domain_events (aggregate_type, aggregate_id, event_type, payload)
    VALUES (
        'Order',
        'a1b2c3d4-0000-0000-0000-000000000001',
        'OrderConfirmed',
        '{"order_id": "a1b2c3d4-0000-0000-0000-000000000001", "confirmed_at": "2024-01-15T10:30:00Z"}'
    );

COMMIT;
-- Both rows land or neither does. The read side cannot miss this event.

-- ============================================================
-- READ SIDE (Query Models) — PostgreSQL read replica
-- Denormalized. No foreign keys enforced. No joins needed at query time.
-- These tables are populated by the event consumer, NOT by the app.
-- ============================================================

-- Projection 1: Order Summary List
-- Powers the "My Orders" page — returns everything needed in ONE query
CREATE TABLE order_summary_projection (
    order_id            UUID PRIMARY KEY,
    customer_id         UUID NOT NULL,
    customer_email      TEXT NOT NULL,    -- denormalized from Customers
    customer_name       TEXT NOT NULL,    -- denormalized from Customers
    status              TEXT NOT NULL,
    item_count          INTEGER NOT NULL, -- pre-aggregated
    total_cents         INTEGER NOT NULL,
    first_item_name     TEXT,            -- denormalized first item for display
    created_at          TIMESTAMPTZ NOT NULL,
    last_updated_at     TIMESTAMPTZ NOT NULL
);

-- Covering index: all columns needed for the list query are in the index
-- This means Postgres can answer the query without touching the heap at all
CREATE INDEX idx_order_summary_customer_created
    ON order_summary_projection (customer_id, created_at DESC)
    INCLUDE (status, item_count, total_cents, first_item_name);

-- Projection 2: Customer Lifetime Value
-- Powers analytics — pre-aggregated so no GROUP BY at query time
CREATE TABLE customer_ltv_projection (
    customer_id         UUID PRIMARY KEY,
    customer_email      TEXT NOT NULL,
    total_orders        INTEGER NOT NULL DEFAULT 0,
    total_spent_cents   BIGINT  NOT NULL DEFAULT 0,
    last_order_at       TIMESTAMPTZ,
    first_order_at      TIMESTAMPTZ
);

-- ============================================================
-- Event consumer: applies domain events to keep projections current
-- This runs in a background worker that polls the domain_events outbox
-- (or listens to Kafka/RabbitMQ if you use a broker)
-- ============================================================

-- Handler for 'OrderConfirmed' event
-- Called by your event consumer with the raw event payload
CREATE OR REPLACE FUNCTION apply_order_confirmed_event(
    p_order_id      UUID,
    p_customer_id   UUID,
    p_confirmed_at  TIMESTAMPTZ
) RETURNS VOID AS $$
BEGIN
    -- Update the order summary projection status
    UPDATE order_summary_projection
    SET    status          = 'confirmed',
           last_updated_at = p_confirmed_at
    WHERE  order_id        = p_order_id;

    -- No need to touch order_items or customers — they're already denormalized
END;
$$ LANGUAGE plpgsql;

-- Handler for 'OrderPlaced' event — builds the initial projection row
CREATE OR REPLACE FUNCTION apply_order_placed_event(
    p_order_id          UUID,
    p_customer_id       UUID,
    p_customer_email    TEXT,
    p_customer_name     TEXT,
    p_item_count        INTEGER,
    p_total_cents       INTEGER,
    p_first_item_name   TEXT,
    p_placed_at         TIMESTAMPTZ
) RETURNS VOID AS $$
BEGIN
    -- INSERT ... ON CONFLICT makes this handler idempotent:
    -- replaying the same event twice won't create duplicate rows
    INSERT INTO order_summary_projection (
        order_id, customer_id, customer_email, customer_name,
        status, item_count, total_cents, first_item_name,
        created_at, last_updated_at
    ) VALUES (
        p_order_id, p_customer_id, p_customer_email, p_customer_name,
        'pending', p_item_count, p_total_cents, p_first_item_name,
        p_placed_at, p_placed_at
    )
    ON CONFLICT (order_id) DO NOTHING;  -- idempotent: safe to replay

    -- Also update or insert the customer LTV projection
    INSERT INTO customer_ltv_projection (
        customer_id, customer_email,
        total_orders, total_spent_cents,
        last_order_at, first_order_at
    ) VALUES (
        p_customer_id, p_customer_email,
        1, p_total_cents,
        p_placed_at, p_placed_at
    )
    ON CONFLICT (customer_id) DO UPDATE SET
        total_orders      = customer_ltv_projection.total_orders + 1,
        total_spent_cents = customer_ltv_projection.total_spent_cents + EXCLUDED.total_spent_cents,
        last_order_at     = GREATEST(customer_ltv_projection.last_order_at, EXCLUDED.last_order_at);
END;
$$ LANGUAGE plpgsql;

-- ============================================================
-- The actual query — no joins, no aggregations, blazing fast
-- ============================================================
SELECT
    order_id,
    status,
    item_count,
    total_cents,
    first_item_name,
    created_at
FROM  order_summary_projection
WHERE customer_id = 'cust-uuid-here'
ORDER BY created_at DESC
LIMIT 20;

# ============================================================
# Outbox Poller + Event Consumer — Python with psycopg2
# Runs as a separate background service alongside your main app.
# Polls the domain_events outbox, publishes to a channel,
# and applies events to read-side projections.
# ============================================================

import psycopg2
import psycopg2.extras
import json
import time
import logging
from datetime import datetime, timezone
from typing import Dict, Callable, Any

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)

# ── Database connection settings ──────────────────────────────
WRITE_DB_DSN = "postgresql://app_user:secret@write-db:5432/orders_write"
READ_DB_DSN  = "postgresql://app_user:secret@read-db:5432/orders_read"

# ── How many events to process in one polling batch ──────────
BATCH_SIZE = 100

# ── Pause between polls when no events are pending (seconds) ─
IDLE_SLEEP_SECONDS = 0.5


def fetch_unpublished_events(write_conn, batch_size: int) -> list:
    """Pull a batch of unpublished events from the outbox, ordered by time.
    We use SELECT ... FOR UPDATE SKIP LOCKED so multiple poller instances
    never process the same event — safe for horizontal scaling."""
    with write_conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
        cursor.execute("""
            SELECT event_id, aggregate_type, aggregate_id,
                   event_type, payload, occurred_at
            FROM   domain_events
            WHERE  published = FALSE
            ORDER  BY occurred_at ASC
            LIMIT  %s
            FOR UPDATE SKIP LOCKED
        """, (batch_size,))
        return cursor.fetchall()


def mark_events_published(write_conn, event_ids: list) -> None:
    """Mark the batch as published only AFTER the read side has committed.
    If the read-side apply fails, these stay unpublished and will be retried."""
    with write_conn.cursor() as cursor:
        cursor.execute("""
            UPDATE domain_events
            SET    published = TRUE
            WHERE  event_id  = ANY(%s)
        """, (event_ids,))
    write_conn.commit()  # commit the mark-published update


def apply_order_placed(read_conn, payload: Dict[str, Any]) -> None:
    """Update the order_summary_projection and customer_ltv_projection.
    Uses ON CONFLICT to be idempotent — safe to replay."""
    with read_conn.cursor() as cursor:
        cursor.execute("""
            SELECT apply_order_placed_event(
                %s::uuid, %s::uuid, %s, %s, %s, %s, %s, %s::timestamptz
            )
        """, (
            payload['order_id'],
            payload['customer_id'],
            payload['customer_email'],
            payload['customer_name'],
            payload['item_count'],
            payload['total_cents'],
            payload.get('first_item_name'),
            payload['placed_at']
        ))
    read_conn.commit()
    logger.info(f"Applied OrderPlaced for order {payload['order_id']}")


def apply_order_confirmed(read_conn, payload: Dict[str, Any]) -> None:
    """Update status in the order summary projection."""
    with read_conn.cursor() as cursor:
        cursor.execute("""
            SELECT apply_order_confirmed_event(%s::uuid, %s::uuid, %s::timestamptz)
        """, (
            payload['order_id'],
            payload['customer_id'],
            payload['confirmed_at']
        ))
    read_conn.commit()
    logger.info(f"Applied OrderConfirmed for order {payload['order_id']}")


# ── Registry mapping event_type strings to handler functions ─
# Adding a new event type is one line here + one handler function.
EVENT_HANDLERS: Dict[str, Callable] = {
    'OrderPlaced':     apply_order_placed,
    'OrderConfirmed':  apply_order_confirmed,
    # 'OrderShipped':  apply_order_shipped,   # add as needed
    # 'OrderCancelled': apply_order_cancelled,
}


def run_outbox_poller():
    """Main loop: poll → apply → mark published. Runs forever."""
    write_conn = psycopg2.connect(WRITE_DB_DSN)
    read_conn  = psycopg2.connect(READ_DB_DSN)
    write_conn.autocommit = False  # we control transactions manually
    read_conn.autocommit  = False

    logger.info("Outbox poller started — watching for domain events...")

    while True:
        try:
            # Open a transaction on the write side to lock the batch
            events = fetch_unpublished_events(write_conn, BATCH_SIZE)

            if not events:
                write_conn.rollback()  # release the FOR UPDATE lock
                time.sleep(IDLE_SLEEP_SECONDS)
                continue

            logger.info(f"Processing batch of {len(events)} event(s)")
            processed_ids = []

            for event in events:
                event_type = event['event_type']
                payload    = event['payload']  # already a dict from JSONB

                handler = EVENT_HANDLERS.get(event_type)

                if handler is None:
                    # Unknown event type — log and skip (don't crash the poller)
                    logger.warning(f"No handler for event type '{event_type}' — skipping")
                    processed_ids.append(event['event_id'])
                    continue

                try:
                    # Apply the event to the read-side projection
                    handler(read_conn, payload)
                    processed_ids.append(event['event_id'])

                except Exception as apply_error:
                    # Read-side apply failed — rollback read side, stop batch.
                    # These events remain unpublished and will retry next poll.
                    read_conn.rollback()
                    write_conn.rollback()
                    logger.error(
                        f"Failed to apply event {event['event_id']} "
                        f"({event_type}): {apply_error}"
                    )
                    break  # stop this batch, retry after sleep

            if processed_ids:
                # Only mark as published AFTER read side committed successfully
                mark_events_published(write_conn, processed_ids)
                logger.info(f"Marked {len(processed_ids)} event(s) as published")

        except Exception as poll_error:
            logger.error(f"Poller error: {poll_error}")
            try:
                write_conn.rollback()
                read_conn.rollback()
            except Exception:
                pass
            time.sleep(2)  # back off before retrying


if __name__ == '__main__':
    run_outbox_poller()

# ============================================================
# Minimal Event Store Client — io.thecodeforge.eventstore
# Append-only stream for Event Sourcing.
# Uses PostgreSQL as the store with snapshot support.
# ============================================================

import psycopg2
import psycopg2.extras
import json
import uuid
from datetime import datetime, timezone
from typing import Optional, Dict, List, Any


class EventStore:
    def __init__(self, dsn: str):
        self.dsn = dsn

    def append_event(self, stream_id: str, event_type: str,
                     data: Dict[str, Any], expected_version: Optional[int] = None) -> int:
        """Append an event to the stream. If expected_version is provided,
        the operation will fail if the stream version doesn't match (optimistic concurrency)."""
        conn = psycopg2.connect(self.dsn)
        try:
            with conn.cursor() as cur:
                # Retrieve current version number for the stream
                cur.execute("""
                    SELECT version FROM io_thecodeforge.event_streams
                    WHERE stream_id = %s
                    FOR UPDATE
                """, (stream_id,))
                row = cur.fetchone()
                current_version = row[0] if row else 0

                if expected_version is not None and current_version != expected_version:
                    raise ValueError(
                        f"Concurrency error: expected version {expected_version}, "
                        f"current version {current_version}"
                    )

                new_version = current_version + 1

                # Upsert the stream row with new version
                cur.execute("""
                    INSERT INTO io_thecodeforge.event_streams (stream_id, version, last_updated)
                    VALUES (%s, %s, %s)
                    ON CONFLICT (stream_id) DO UPDATE SET
                        version = EXCLUDED.version,
                        last_updated = EXCLUDED.last_updated
                """, (stream_id, new_version, datetime.now(timezone.utc)))

                # Insert the event
                cur.execute("""
                    INSERT INTO io_thecodeforge.events (event_id, stream_id, version, event_type, data, occurred_at)
                    VALUES (%s, %s, %s, %s, %s, %s)
                """, (
                    str(uuid.uuid4()), stream_id, new_version,
                    event_type, json.dumps(data), datetime.now(timezone.utc)
                ))
                conn.commit()
                return new_version
        finally:
            conn.close()

    def read_events(self, stream_id: str, from_version: int = 1) -> List[Dict[str, Any]]:
        """Return all events for a stream starting from a given version."""
        conn = psycopg2.connect(self.dsn)
        try:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                cur.execute("""
                    SELECT event_id, version, event_type, data, occurred_at
                    FROM io_thecodeforge.events
                    WHERE stream_id = %s
                    AND version >= %s
                    ORDER BY version ASC
                """, (stream_id, from_version))
                return cur.fetchall()
        finally:
            conn.close()

    def create_snapshot(self, stream_id: str, snapshot_data: Dict[str, Any], version: int) -> None:
        """Store a snapshot for efficient rebuilding of current state."""
        conn = psycopg2.connect(self.dsn)
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO io_thecodeforge.event_snapshots (stream_id, version, snapshot_data, created_at)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (stream_id, version) DO NOTHING
                """, (stream_id, version, json.dumps(snapshot_data), datetime.now(timezone.utc)))
                conn.commit()
        finally:
            conn.close()

# ============================================================
# Multi-store event consumer — io.thecodeforge.eventconsumer
# Applies the same event to different projections in different stores.
# Each projection is idempotent and can fail independently.
# ============================================================

import json
import logging
from typing import Dict, Any, Callable

logger = logging.getLogger(__name__)

class ProjectionHandler:
    """Base class for a projection that updates a specific store."""
    def handle(self, event_type: str, payload: Dict[str, Any]) -> None:
        raise NotImplementedError

class PostgresOrderSummaryProjection(ProjectionHandler):
    def handle(self, event_type: str, payload: Dict[str, Any]) -> None:
        # ... PostgreSQL upsert logic
        pass

class ElasticsearchProductSearchProjection(ProjectionHandler):
    def handle(self, event_type: str, payload: Dict[str, Any]) -> None:
        # ... Elasticsearch index update logic
        pass

class RedisAnalyticsProjection(ProjectionHandler):
    def handle(self, event_type: str, payload: Dict[str, Any]) -> None:
        # ... Redis sorted set increments
        pass


def process_event(event: Dict[str, Any], handlers: Dict[str, ProjectionHandler]):
    """Apply event to all registered projections.
    Each projection runs independently; failures are logged but don't block others."""
    event_type = event['event_type']
    payload = event['payload']
    for name, handler in handlers.items():
        try:
            handler.handle(event_type, payload)
            logger.info(f"Projection '{name}' updated for event {event['event_id']}")
        except Exception as e:
            logger.error(f"Projection '{name}' failed for event {event['event_id']}: {e}")
            # Optionally send to dead-letter queue

# Example usage
handlers = {
    'order_summary': PostgresOrderSummaryProjection(),
    'product_search': ElasticsearchProductSearchProjection(),
    'analytics': RedisAnalyticsProjection(),
}

for event in event_batch:
    process_event(event, handlers)