Testcontainers Kafka tests for a payment event pipeline: redelivery and ledger state

npm install --save-dev @testcontainers/kafka@12.1.0 @testcontainers/postgresql@12.1.0 \
  @confluentinc/kafka-javascript@1.10.1 pg@8.23.0 @types/pg vitest@5.0.1 typescript

// src/ledger-consumer.ts
import { KafkaJS } from "@confluentinc/kafka-javascript";
import type { Pool } from "pg";

export type PaymentEvent = {
  eventId: string;
  type: "payment.authorised" | "payment.captured";
  paymentId: string; // Kafka message key
  accountId: string;
  amountMinor: number;
  currency: string;
};

export const SCHEMA = `
  CREATE TABLE processed_events (event_id text PRIMARY KEY, kafka_offset text NOT NULL);
  CREATE TABLE ledger_entries (
    id bigserial PRIMARY KEY,
    event_id text NOT NULL REFERENCES processed_events (event_id),
    account text NOT NULL,
    amount_minor bigint NOT NULL,          -- credit positive, debit negative
    currency char(3) NOT NULL
  );`;

// Postings per event type. Each set nets to zero.
function postings(e: PaymentEvent): Array<[string, number]> {
  const available = `customer:${e.accountId}:available`;
  const held = `customer:${e.accountId}:held`;
  return e.type === "payment.authorised"
    ? [[available, -e.amountMinor], [held, e.amountMinor]]
    : [[held, -e.amountMinor], ["merchant:settlement", e.amountMinor]];
}

// Returns "applied" or "duplicate". The inbox row and the postings commit in one transaction.
export async function applyEvent(pool: Pool, e: PaymentEvent, offset: string): Promise<"applied" | "duplicate"> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const inserted = await client.query(
      "INSERT INTO processed_events (event_id, kafka_offset) VALUES ($1, $2) ON CONFLICT (event_id) DO NOTHING RETURNING event_id",
      [e.eventId, offset],
    );
    if (inserted.rowCount === 0) {
      await client.query("ROLLBACK");
      return "duplicate";
    }
    for (const [account, amount] of postings(e)) {
      await client.query(
        "INSERT INTO ledger_entries (event_id, account, amount_minor, currency) VALUES ($1, $2, $3, $4)",
        [e.eventId, account, amount, e.currency],
      );
    }
    await client.query("COMMIT");
    return "applied";
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

export async function startLedgerConsumer(opts: {
  kafka: KafkaJS.Kafka;
  pool: Pool;
  topic: string;
  groupId: string;
  commitOffsets: boolean; // false simulates a crash between the database commit and the offset commit
  onResult: (r: "applied" | "duplicate") => void;
}): Promise<KafkaJS.Consumer> {
  const consumer = opts.kafka.consumer({
    kafkaJS: { groupId: opts.groupId, fromBeginning: true, autoCommit: false },
  });
  await consumer.connect();
  await consumer.subscribe({ topic: opts.topic });
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value!.toString()) as PaymentEvent;
      opts.onResult(await applyEvent(opts.pool, event, message.offset));
      if (opts.commitOffsets) {
        await consumer.commitOffsets([
          { topic, partition, offset: (BigInt(message.offset) + 1n).toString() },
        ]);
      }
    },
  });
  return consumer;
}

// src/payment-pipeline.test.ts
import { KafkaJS } from "@confluentinc/kafka-javascript";
import { KafkaContainer, StartedKafkaContainer } from "@testcontainers/kafka";
import { PostgreSqlContainer, StartedPostgreSqlContainer } from "@testcontainers/postgresql";
import { Pool } from "pg";
import { afterAll, beforeAll, expect, test } from "vitest";
import { PaymentEvent, SCHEMA, startLedgerConsumer } from "./ledger-consumer";

const TOPIC = "payments.events";
let kafkaContainer: StartedKafkaContainer;
let pgContainer: StartedPostgreSqlContainer;
let kafka: KafkaJS.Kafka;
let pool: Pool;

beforeAll(async () => {
  [kafkaContainer, pgContainer] = await Promise.all([
    new KafkaContainer("confluentinc/cp-kafka:8.3.2").start(), // tag >= 8.0.0 starts in KRaft mode
    new PostgreSqlContainer("postgres:18.6-alpine").start(),
  ]);
  kafka = new KafkaJS.Kafka({
    kafkaJS: {
      brokers: [`${kafkaContainer.getHost()}:${kafkaContainer.getMappedPort(9093)}`],
      logLevel: KafkaJS.logLevel.ERROR,
    },
  });
  const admin = kafka.admin();
  await admin.connect();
  await admin.createTopics({ topics: [{ topic: TOPIC, numPartitions: 6 }] });
  await admin.disconnect();
  pool = new Pool({ connectionString: pgContainer.getConnectionUri() });
  await pool.query(SCHEMA);
}, 240_000);

afterAll(async () => {
  await pool?.end();
  await kafkaContainer?.stop();
  await pgContainer?.stop();
});

async function waitFor(check: () => boolean | Promise<boolean>, timeoutMs = 30_000) {
  const deadline = Date.now() + timeoutMs;
  while (!(await check())) {
    if (Date.now() > deadline) throw new Error("condition not met in time");
    await new Promise((r) => setTimeout(r, 200));
  }
}

async function balances(accountId: string) {
  const { rows } = await pool.query(
    `SELECT account, SUM(amount_minor)::bigint AS balance
       FROM ledger_entries
      WHERE account LIKE $1 OR account = 'merchant:settlement'
      GROUP BY account ORDER BY account`,
    [`customer:${accountId}:%`],
  );
  return Object.fromEntries(rows.map((r) => [r.account, Number(r.balance)]));
}

test("redelivery after a crash before the offset commit posts once", async () => {
  const authorised: PaymentEvent = {
    eventId: "evt_0001", type: "payment.authorised", paymentId: "pay_0001",
    accountId: "acc_0001", amountMinor: 2599, currency: "EUR",
  };
  const captured: PaymentEvent = { ...authorised, eventId: "evt_0002", type: "payment.captured" };

  const producer = kafka.producer({ kafkaJS: { idempotent: true } });
  await producer.connect();
  const meta = await producer.send({
    topic: TOPIC,
    messages: [authorised, authorised, captured].map((e) => ({ key: e.paymentId, value: JSON.stringify(e) })),
  });
  await producer.disconnect();
  // The second authorised message is an upstream resend: same eventId, new offset.
  // Ordering per key holds only if all three messages share a partition.
  expect(new Set(meta.map((m) => m.partition)).size).toBe(1);

  // Run 1: processes all three messages, commits the database, never commits offsets, then "crashes".
  const run1: string[] = [];
  const crashed = await startLedgerConsumer({
    kafka, pool, topic: TOPIC, groupId: "ledger", commitOffsets: false, onResult: (r) => run1.push(r),
  });
  await waitFor(() => run1.length === 3);
  await crashed.disconnect();

  // Run 2: same group, no committed offset, so Kafka redelivers all three messages.
  const run2: string[] = [];
  const restarted = await startLedgerConsumer({
    kafka, pool, topic: TOPIC, groupId: "ledger", commitOffsets: true, onResult: (r) => run2.push(r),
  });
  await waitFor(() => run2.length === 3);
  await restarted.disconnect();

  expect(run1).toEqual(["applied", "duplicate", "applied"]);
  expect(run2).toEqual(["duplicate", "duplicate", "duplicate"]);

  // Ledger state: one hold opened and released, one capture, nothing doubled.
  expect(await balances("acc_0001")).toEqual({
    "customer:acc_0001:available": -2599,
    "customer:acc_0001:held": 0,
    "merchant:settlement": 2599,
  });
  const { rows } = await pool.query(
    "SELECT event_id, SUM(amount_minor)::bigint AS net, COUNT(*)::int AS entries FROM ledger_entries GROUP BY event_id ORDER BY event_id",
  );
  expect(rows).toEqual([
    { event_id: "evt_0001", net: "0", entries: 2 },
    { event_id: "evt_0002", net: "0", entries: 2 },
  ]);
}, 120_000);