Transactional Outbox & CDC

# Python — psycopg3 + confluent-kafka
import json, time, uuid
import psycopg
from confluent_kafka import Producer

# ── HALF 1: the atomic write ────────────────────────────────
def place_order(conn, cart):
    # ONE transaction. Both inserts, or neither. No broker involved.
    with conn.transaction():
        order_id = conn.execute(
            "INSERT INTO orders (user_id, total_cents, status) "
            "VALUES (%s, %s, 'PLACED') RETURNING id",
            (cart.user_id, cart.total_cents)).fetchone()[0]

        conn.execute(
            "INSERT INTO outbox "
            "  (event_id, aggregate_type, aggregate_id, event_type, payload) "
            "VALUES (%s, 'order', %s, 'order.created', %s)",
            (str(uuid.uuid4()), str(order_id),
             json.dumps({"order_id": order_id,
                         "user_id": cart.user_id,
                         "total_cents": cart.total_cents})))
    return order_id

# ── HALF 2: the drain loop (run N of these) ─────────────────
BATCH = 500
POLL_S = 0.1   # 500 rows / 100ms = 5,000 events/s per worker

def drain_forever(dsn, producer):
    with psycopg.connect(dsn, autocommit=False) as conn:
        while True:
            n = drain_once(conn, producer)
            # Only sleep when idle: under load we loop with zero added latency.
            if n < BATCH:
                time.sleep(POLL_S)

def drain_once(conn, producer):
    with conn.transaction():
        rows = conn.execute(
            """SELECT id, event_id, aggregate_type, aggregate_id,
                      event_type, payload
                 FROM outbox
                WHERE published_at IS NULL
                ORDER BY id
                LIMIT %s
                FOR UPDATE SKIP LOCKED""",   # ← lets workers scale
            (BATCH,)).fetchall()

        if not rows:
            return 0

        for (_id, event_id, agg_type, agg_id, ev_type, payload) in rows:
            producer.produce(
                topic=f"{agg_type}.events",
                key=agg_id,          # ← per-aggregate ORDERING
                value=json.dumps(payload),
                headers=[("event_id", event_id),   # ← consumer dedup key
                         ("event_type", ev_type)])

        # flush() BEFORE the UPDATE. If flush raises, the transaction rolls
        # back, rows stay unpublished, and we retry. Marking first would
        # mean silently LOSING events — the exact bug we came here to fix.
        if producer.flush(timeout=10) > 0:
            raise RuntimeError("kafka flush incomplete; will retry batch")

        conn.execute("UPDATE outbox SET published_at = now() "
                     "WHERE id = ANY(%s)", ([r[0] for r in rows],))
    return len(rows)

// C++ — libpqxx + librdkafka
#include <pqxx/pqxx>
#include <librdkafka/rdkafkacpp.h>
#include <thread>

constexpr int BATCH = 500;

// ── HALF 1: atomic write ────────────────────────────────────
long place_order(pqxx::connection& c, long user_id, long cents) {
    pqxx::work tx{c};   // RAII: destructor rolls back unless commit() is called

    auto r = tx.exec_params1(
        "INSERT INTO orders (user_id, total_cents, status) "
        "VALUES ($1,$2,'PLACED') RETURNING id", user_id, cents);
    long order_id = r[0].as<long>();

    tx.exec_params(
        "INSERT INTO outbox (event_id, aggregate_type, aggregate_id, "
        "                    event_type, payload) "
        "VALUES (gen_random_uuid(),'order',$1,'order.created',$2::jsonb)",
        std::to_string(order_id),
        R"({"order_id":)" + std::to_string(order_id) + R"(})");

    tx.commit();   // ← the single atomic moment
    return order_id;
}

// ── HALF 2: drain loop ──────────────────────────────────────
int drain_once(pqxx::connection& c, RdKafka::Producer* p) {
    pqxx::work tx{c};

    auto rows = tx.exec_params(
        "SELECT id, event_id::text, aggregate_type, aggregate_id, "
        "       event_type, payload::text "
        "  FROM outbox WHERE published_at IS NULL "
        " ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED", BATCH);

    if (rows.empty()) { tx.commit(); return 0; }

    std::string ids = "{";
    for (auto row : rows) {
        std::string topic = row[2].c_str() + std::string(".events");
        std::string key   = row[3].c_str();
        std::string val   = row[5].c_str();

        RdKafka::Headers* h = RdKafka::Headers::create();
        h->add("event_id", row[1].c_str());

        p->produce(topic, RdKafka::Topic::PARTITION_UA,
                   RdKafka::Producer::RK_MSG_COPY,
                   (void*)val.data(), val.size(),
                   key.data(), key.size(),   // key => partition => ORDER
                   0, h, nullptr);

        ids += (ids.size() > 1 ? "," : "") + row[0].c_str();
    }
    ids += "}";

    // Block until every message is acked. Anything left => throw and
    // let tx destruct (rollback). Rows stay unpublished; we retry.
    if (p->flush(10000) != RdKafka::ERR_NO_ERROR)
        throw std::runtime_error("flush incomplete; batch will retry");

    tx.exec_params("UPDATE outbox SET published_at = now() "
                   "WHERE id = ANY($1::bigint[])", ids);
    tx.commit();
    return (int)rows.size();
}

void drain_forever(pqxx::connection& c, RdKafka::Producer* p) {
    for (;;) {
        if (drain_once(c, p) < BATCH)
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

// Java — Spring @Transactional + KafkaTemplate
import org.springframework.transaction.annotation.Transactional;
import org.springframework.scheduling.annotation.Scheduled;

@Service
public class OrderService {

    private final OrderRepo orders;
    private final OutboxRepo outbox;

    // ── HALF 1: atomic write ────────────────────────────────
    // NOTE: @Transactional works here ONLY because both writes go to the
    // SAME datasource. Add kafkaTemplate.send() to this method and the
    // annotation silently stops protecting you.
    @Transactional
    public long placeOrder(Cart cart) {
        Order o = orders.save(new Order(cart.userId(), cart.totalCents(), "PLACED"));

        outbox.save(new OutboxEvent(
            UUID.randomUUID(),          // event_id -> consumer dedup key
            "order",                    // aggregate_type -> topic
            String.valueOf(o.getId()),  // aggregate_id -> partition key
            "order.created",
            """
            {"order_id":%d,"user_id":%d,"total_cents":%d}
            """.formatted(o.getId(), cart.userId(), cart.totalCents())));

        return o.getId();   // commit happens as this method returns
    }
}

@Component
public class OutboxPublisher {

    private static final int BATCH = 500;
    private final JdbcTemplate jdbc;
    private final KafkaTemplate<String, String> kafka;

    // ── HALF 2: drain loop ──────────────────────────────────
    @Scheduled(fixedDelay = 100)
    @Transactional
    public void drain() {
        List<Map<String, Object>> rows = jdbc.queryForList("""
            SELECT id, event_id, aggregate_type, aggregate_id,
                   event_type, payload::text AS payload
              FROM outbox
             WHERE published_at IS NULL
             ORDER BY id
             LIMIT ?
             FOR UPDATE SKIP LOCKED
            """, BATCH);

        if (rows.isEmpty()) return;

        List<CompletableFuture<?>> sends = new ArrayList<>();
        List<Long> ids = new ArrayList<>();

        for (var r : rows) {
            var rec = new ProducerRecord<>(
                r.get("aggregate_type") + ".events",
                (String) r.get("aggregate_id"),        // ORDERING key
                (String) r.get("payload"));
            rec.headers().add("event_id",
                r.get("event_id").toString().getBytes(UTF_8));

            sends.add(kafka.send(rec).toCompletableFuture());
            ids.add(((Number) r.get("id")).longValue());
        }

        // Await ALL acks before marking. An exception here rolls back the
        // UPDATE, so the rows are simply retried on the next tick.
        CompletableFuture
            .allOf(sends.toArray(CompletableFuture[]::new))
            .join();

        jdbc.update("UPDATE outbox SET published_at = now() "
                  + "WHERE id = ANY(?)", ids.toArray(Long[]::new));
    }
}

// Go — pgx + segmentio/kafka-go
package outbox

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/google/uuid"
    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/segmentio/kafka-go"
)

const batchSize = 500

// ── HALF 1: atomic write ────────────────────────────────────
func PlaceOrder(ctx context.Context, pool *pgxpool.Pool,
    userID, cents int64) (int64, error) {

    tx, err := pool.Begin(ctx)
    if err != nil {
        return 0, err
    }
    defer tx.Rollback(ctx) // no-op after a successful Commit

    var orderID int64
    if err := tx.QueryRow(ctx,
        `INSERT INTO orders (user_id, total_cents, status)
         VALUES ($1,$2,'PLACED') RETURNING id`,
        userID, cents).Scan(&orderID); err != nil {
        return 0, err
    }

    payload, _ := json.Marshal(map[string]any{
        "order_id": orderID, "user_id": userID, "total_cents": cents,
    })

    if _, err := tx.Exec(ctx,
        `INSERT INTO outbox
           (event_id, aggregate_type, aggregate_id, event_type, payload)
         VALUES ($1,'order',$2,'order.created',$3)`,
        uuid.New(), fmt.Sprint(orderID), payload); err != nil {
        return 0, err
    }

    return orderID, tx.Commit(ctx) // ← the atomic moment
}

// ── HALF 2: drain loop ──────────────────────────────────────
func DrainOnce(ctx context.Context, pool *pgxpool.Pool,
    w *kafka.Writer) (int, error) {

    tx, err := pool.Begin(ctx)
    if err != nil {
        return 0, err
    }
    defer tx.Rollback(ctx)

    rows, err := tx.Query(ctx, `
        SELECT id, event_id, aggregate_type, aggregate_id,
               event_type, payload
          FROM outbox
         WHERE published_at IS NULL
         ORDER BY id
         LIMIT $1
         FOR UPDATE SKIP LOCKED`, batchSize) // ← safe multi-worker
    if err != nil {
        return 0, err
    }

    var (
        msgs []kafka.Message
        ids  []int64
    )
    for rows.Next() {
        var (
            id                        int64
            eventID                   uuid.UUID
            aggType, aggID, eventType string
            payload                   []byte
        )
        if err := rows.Scan(&id, &eventID, &aggType, &aggID,
            &eventType, &payload); err != nil {
            rows.Close()
            return 0, err
        }
        msgs = append(msgs, kafka.Message{
            Topic: aggType + ".events",
            Key:   []byte(aggID), // key => partition => per-order ORDER
            Value: payload,
            Headers: []kafka.Header{
                {Key: "event_id", Value: []byte(eventID.String())},
                {Key: "event_type", Value: []byte(eventType)},
            },
        })
        ids = append(ids, id)
    }
    rows.Close()

    if len(ids) == 0 {
        return 0, tx.Commit(ctx)
    }

    // Publish FIRST. On error we return without marking, the deferred
    // Rollback fires, and the batch is retried. Never mark-then-publish.
    if err := w.WriteMessages(ctx, msgs...); err != nil {
        return 0, err
    }

    if _, err := tx.Exec(ctx,
        `UPDATE outbox SET published_at = now() WHERE id = ANY($1)`,
        ids); err != nil {
        return 0, err
    }

    return len(ids), tx.Commit(ctx)
}

func DrainForever(ctx context.Context, pool *pgxpool.Pool, w *kafka.Writer) {
    for {
        n, err := DrainOnce(ctx, pool, w)
        switch {
        case err != nil:
            time.Sleep(1 * time.Second) // back off on failure
        case n < batchSize:
            time.Sleep(100 * time.Millisecond) // idle
        }
        // A full batch => loop immediately, no added latency.
    }
}

// TypeScript — node-postgres + kafkajs
import { Pool } from 'pg';
import { Kafka, Producer } from 'kafkajs';
import { randomUUID } from 'node:crypto';

const pool = new Pool();
const BATCH = 500;

// ── HALF 1: atomic write ────────────────────────────────────
export async function placeOrder(cart: Cart): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const { rows } = await client.query<{ id: number }>(
      `INSERT INTO orders (user_id, total_cents, status)
       VALUES ($1,$2,'PLACED') RETURNING id`,
      [cart.userId, cart.totalCents]);
    const orderId = rows[0].id;

    // Same client, same transaction. THIS is what makes it atomic.
    await client.query(
      `INSERT INTO outbox
         (event_id, aggregate_type, aggregate_id, event_type, payload)
       VALUES ($1,'order',$2,'order.created',$3)`,
      [randomUUID(), String(orderId),
       JSON.stringify({ order_id: orderId, user_id: cart.userId,
                        total_cents: cart.totalCents })]);

    await client.query('COMMIT');
    return orderId;
  } catch (e) {
    await client.query('ROLLBACK');  // order AND event both vanish
    throw e;
  } finally {
    client.release();
  }
}

// ── HALF 2: drain loop ──────────────────────────────────────
export async function drainOnce(producer: Producer): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const { rows } = await client.query(
      `SELECT id, event_id, aggregate_type, aggregate_id,
              event_type, payload
         FROM outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT $1
        FOR UPDATE SKIP LOCKED`, [BATCH]);   // ← multi-worker safe

    if (rows.length === 0) {
      await client.query('COMMIT');
      return 0;
    }

    // Group by topic — kafkajs sends one batch per topic.
    const byTopic = new Map<string, any[]>();
    for (const r of rows) {
      const topic = `${r.aggregate_type}.events`;
      if (!byTopic.has(topic)) byTopic.set(topic, []);
      byTopic.get(topic)!.push({
        key: r.aggregate_id,          // ← ORDERING per aggregate
        value: JSON.stringify(r.payload),
        headers: { event_id: r.event_id, event_type: r.event_type },
      });
    }

    // acks:-1 (all in-sync replicas) — set on the producer. If this
    // throws, we ROLLBACK and the rows are retried. Publish, THEN mark.
    await producer.sendBatch({
      acks: -1,
      topicMessages: [...byTopic].map(([topic, messages]) => ({ topic, messages })),
    });

    await client.query(
      `UPDATE outbox SET published_at = now() WHERE id = ANY($1)`,
      [rows.map(r => r.id)]);

    await client.query('COMMIT');
    return rows.length;
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}