// ❌ Naive dual-write — four distinct failure windows
async function processPayment(payment: Payment): Promise<void> {
  // Window 1: DB write succeeds, process crashes before produce
  await db.query('INSERT INTO payments (id, status) VALUES ($1, $2)', [payment.id, 'charged'])

  // Window 2: Produce times out / network error after DB write
  await producer.send({
    topic:    'payments.charged',
    messages: [{ key: payment.id, value: JSON.stringify(payment) }]
  })
  // If send() throws: DB row exists, event missing → unfulfilled order

  // Window 3: Produce succeeds, process crashes before returning
  // (rare but possible with at-most-once acks=0 or acks=1)

  // Window 4: Produce succeeds, DB write was actually rolled back by concurrent transaction
  // (possible if DB write and produce are not in the same transaction)
}

-- Outbox table: lives in the same PostgreSQL schema as business tables
CREATE TABLE outbox (
  id              UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  aggregate_type  VARCHAR(100) NOT NULL,   -- 'payment', 'order', etc.
  aggregate_id    VARCHAR(255) NOT NULL,   -- the business entity ID
  event_type      VARCHAR(100) NOT NULL,   -- 'payment.charged'
  payload         JSONB        NOT NULL,
  topic           VARCHAR(255) NOT NULL,   -- target Kafka topic
  partition_key   VARCHAR(255),            -- Kafka message key
  created_at      TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
  published_at    TIMESTAMPTZ,             -- NULL = unpublished
  retry_count     INT          NOT NULL DEFAULT 0
);

CREATE INDEX idx_outbox_unpublished ON outbox (created_at)
  WHERE published_at IS NULL;

// ✅ Outbox write: DB + event in a single transaction — atomically committed or rolled back
async function processPaymentWithOutbox(payment: Payment): Promise<void> {
  await db.transaction(async (trx) => {
    // 1. Business write
    await trx.query(
      'INSERT INTO payments (id, status, amount_cents) VALUES ($1, $2, $3)',
      [payment.id, 'charged', payment.amountCents]
    )

    // 2. Outbox write — same transaction, same atomicity boundary
    await trx.query(
      `INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, topic, partition_key)
       VALUES ($1, $2, $3, $4, $5, $6)`,
      [
        'payment',
        payment.id,
        'payment.charged',
        JSON.stringify({ paymentId: payment.id, amountCents: payment.amountCents }),
        'payments.charged',
        payment.id,  // partition key: same payment → same partition → ordered
      ]
    )
    // If either write fails: entire transaction rolls back
    // If process crashes after commit: outbox row exists, relay will publish it
  })
}

// ✅ Outbox relay: polls for unpublished rows with SKIP LOCKED (no blocking)
async function runOutboxRelay(): Promise<void> {
  while (true) {
    await db.transaction(async (trx) => {
      // Lock up to 100 unpublished outbox rows — skip any currently locked by other relay instances
      const { rows } = await trx.query<OutboxRow>(`
        SELECT id, topic, partition_key, payload, aggregate_id, event_type
        FROM outbox
        WHERE published_at IS NULL
        ORDER BY created_at
        LIMIT 100
        FOR UPDATE SKIP LOCKED
      `)

      if (rows.length === 0) return

      // Publish all rows to Kafka
      await producer.sendBatch({
        topicMessages: rows.map(row => ({
          topic:    row.topic,
          messages: [{
            key:     row.partition_key ?? row.aggregate_id,
            value:   JSON.stringify(row.payload),
            headers: {
              'event-type':   row.event_type,
              'outbox-id':    row.id,
              'x-correlation-id': row.aggregate_id,
            }
          }]
        }))
      })

      // Mark as published only after successful Kafka write
      const ids = rows.map(r => r.id)
      await trx.query(
        'UPDATE outbox SET published_at = NOW() WHERE id = ANY($1)',
        [ids]
      )
      // If Kafka publish fails: transaction rolls back, published_at stays NULL → retry next cycle
    })

    await new Promise(resolve => setTimeout(resolve, 100))  // 100ms polling interval
  }
}

// Debezium PostgreSQL connector configuration
{
  "name": "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":           "payments",
    "plugin.name":               "pgoutput",
    "slot.name":                 "debezium_outbox_slot",
    "publication.name":          "debezium_publication",
    "table.include.list":        "public.outbox",

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

    "key.converter":   "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

-- PostgreSQL: configure logical replication for Debezium
-- Run once as superuser:
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 5;
ALTER SYSTEM SET max_wal_senders = 5;

CREATE ROLE debezium REPLICATION LOGIN PASSWORD 'xxx';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
CREATE PUBLICATION debezium_publication FOR TABLE outbox;

// ✅ Consumer Inbox table — deduplication gate for Debezium-delivered events
// (identical to Series 1, Part 5 idempotency gate — the transport changed, not the pattern)

async function processPaymentEvent(event: PaymentEvent): Promise<void> {
  // outbox-id header = the UUID from the outbox table row
  // Stable across Debezium retries — same outbox row → same outbox-id
  const idempotencyKey = event.outboxId  // from Kafka message header

  await db.query(`
    INSERT INTO payment_inbox (idempotency_key, payment_id, amount_cents, processed_at)
    VALUES ($1, $2, $3, NOW())
    ON CONFLICT (idempotency_key) DO NOTHING
  `, [idempotencyKey, event.paymentId, event.amountCents])
  // Debezium redelivery → ON CONFLICT DO NOTHING → no duplicate side effect
}