// ❌ The dual-write hazard — appears harmless, silently loses events in production
async function placeOrder(cmd: PlaceOrderCommand): Promise<void> {
  await db('orders').insert({ id: cmd.orderId, userId: cmd.userId, status: 'pending' })
  // ✅ Database committed successfully

  await kafkaProducer.send('order.placed', { orderId: cmd.orderId, userId: cmd.userId })
  // ❌ Process crashes here — database committed, Kafka message never published
  // Downstream consumers never know the order exists
}

-- Outbox table — lives in the same database as the domain tables
CREATE TABLE outbox_events (
  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_type  TEXT NOT NULL,    -- e.g. 'Order'
  aggregate_id    TEXT NOT NULL,    -- e.g. 'order-abc123'
  event_type      TEXT NOT NULL,    -- e.g. 'OrderPlaced'
  payload         JSONB NOT NULL,   -- full event body
  published_at    TIMESTAMPTZ,      -- null = unpublished
  created_at      TIMESTAMPTZ DEFAULT now(),
  idempotency_key TEXT UNIQUE NOT NULL -- prevents duplicate publishing on relay retry
);

CREATE INDEX idx_outbox_unpublished ON outbox_events (created_at)
  WHERE published_at IS NULL;   -- partial index on unpublished rows only

// ✅ Transactional Outbox — domain write and event payload in one atomic transaction
async function placeOrder(cmd: PlaceOrderCommand): Promise<void> {
  await db.transaction(async (trx) => {
    // Write 1: persist domain state
    await trx('orders').insert({
      id: cmd.orderId, user_id: cmd.userId,
      total_cents: cmd.totalCents, status: 'pending'
    })

    // Write 2: persist event payload in the outbox — SAME TRANSACTION
    await trx('outbox_events').insert({
      aggregate_type:  'Order',
      aggregate_id:    cmd.orderId,
      event_type:      'OrderPlaced',
      payload:         JSON.stringify({
        orderId: cmd.orderId, userId: cmd.userId,
        totalCents: cmd.totalCents, placedAt: new Date().toISOString()
      }),
      idempotency_key: `OrderPlaced-${cmd.orderId}`,
    })
    // Either BOTH commit → event will be published to Kafka by relay
    // Or BOTH rollback → no orphaned event, no missing event
  })
}

// ✅ Polling relay — reads unpublished outbox rows and publishes them to Kafka
// Simple to implement; adds polling latency (typically 100ms–1s)
class OutboxPollingRelay {
  async run(): Promise<void> {
    while (true) {
      const unpublished = await db('outbox_events')
        .whereNull('published_at')
        .orderBy('created_at', 'asc')
        .limit(100)
        .forUpdate()  // pessimistic lock — prevents double-publishing from parallel relays
        .skipLocked() // skip rows locked by another relay instance

      for (const row of unpublished) {
        await kafkaProducer.send({
          topic: `${row.aggregate_type.toLowerCase()}.${row.event_type.toLowerCase()}`,
          messages: [{ key: row.aggregate_id, value: row.payload }]
        })
        await db('outbox_events').where({ id: row.id }).update({ published_at: new Date() })
      }

      await sleep(100) // 100ms polling interval
    }
  }
}

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${DB_PASSWORD}",
    "database.dbname": "orders_db",
    "table.include.list": "public.outbox_events",
    "slot.name": "debezium_outbox_slot",
    "plugin.name": "pgoutput",
    "publication.name": "debezium_outbox_publication",

    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "idempotency_key",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "event_type",
    "transforms.outbox.route.topic.replacement": "order.${routedByValue}"
  }
}

-- Consumer side: inbox deduplication table
CREATE TABLE inbox_events (
  idempotency_key TEXT PRIMARY KEY,   -- same key as outbox.idempotency_key
  processed_at    TIMESTAMPTZ DEFAULT now()
);

// ✅ Idempotent consumer — deduplicates using the inbox table
class FulfillmentConsumer {
  async handleOrderPlaced(message: KafkaMessage): Promise<void> {
    const event = JSON.parse(message.value!.toString()) as OrderPlacedEvent
    const idempotencyKey = message.headers?.['idempotency-key']?.toString()

    await db.transaction(async (trx) => {
      // Check if this event was already processed
      const existing = await trx('inbox_events')
        .where({ idempotency_key: idempotencyKey })
        .first()

      if (existing) {
        // Already processed — skip idempotently (no duplicate fulfillment)
        return
      }

      // Process the event
      await trx('fulfillments').insert({
        order_id: event.orderId, status: 'pending', created_at: new Date()
      })

      // Mark as processed in the inbox — in the same transaction
      await trx('inbox_events').insert({ idempotency_key: idempotencyKey })
      // Either both commit (processed + inbox record) or both rollback (retry on next delivery)
    })
  }
}

# Monitor Debezium replication slot lag — critical for storage management
SELECT
  slot_name,
  restart_lsn,
  confirmed_flush_lsn,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_bytes
FROM pg_replication_slots
WHERE active = true;

# A lagging slot causes WAL to accumulate on disk — unbounded growth if CDC is paused
# Alert when lag_bytes > 1GB; set max_slot_wal_keep_size to prevent disk exhaustion

-- postgresql.conf / ALTER SYSTEM
wal_level = logical                      -- required for Debezium
max_replication_slots = 5               -- one per connector
max_wal_senders = 5                     -- one per slot
max_slot_wal_keep_size = '2GB'          -- prevent runaway WAL growth if CDC lags