Design a Payment Flow

// The relay is the only thing that publishes. It is allowed to publish the same
// event twice; it is not allowed to lose one or to publish out of order within
// an aggregate.
import { pool } from './db.js';
import { broker } from './broker.js';

export async function drainOutbox(batchSize = 100) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    // SKIP LOCKED lets several relay instances run at once without contending.
    // Ordering by id preserves per-aggregate order as long as one aggregate's
    // events are not split across concurrent batches - hence the FILTER below
    // in production: partition by aggregate_id, or run a single relay.
    const { rows } = await client.query(
      `SELECT id, type, payload
         FROM outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT $1
          FOR UPDATE SKIP LOCKED`,
      [batchSize]
    );
    if (rows.length === 0) { await client.query('ROLLBACK'); return 0; }

    for (const row of rows) {
      // Carry the outbox id as the broker-level message id so consumers can
      // deduplicate without parsing the payload.
      await broker.publish(row.type, row.payload, { messageId: String(row.id) });
    }

    // If the process dies between publish and this UPDATE, the batch is sent
    // again on the next run. At-least-once by construction.
    await client.query(
      `UPDATE outbox SET published_at = now() WHERE id = ANY($1::bigint[])`,
      [rows.map((r) => r.id)]
    );
    await client.query('COMMIT');
    return rows.length;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

import express from 'express';
import Stripe from 'stripe';
import { pool } from './db.js';

const stripe = new Stripe(process.env.STRIPE_SECRET_KEY);
const app = express();

// Ranking the states lets us ignore a late event without a timestamp comparison.
const RANK = {
  requires_payment_method: 0,
  requires_confirmation: 1,
  requires_action: 2,
  processing: 3,
  succeeded: 4,
  canceled: 4,
};

app.post(
  '/webhooks/stripe',
  // The signature covers the exact bytes Stripe sent, so this one route must
  // NOT use a JSON body parser. Mounting express.json() globally breaks it.
  express.raw({ type: 'application/json' }),
  async (req, res) => {
    let event;
    try {
      event = stripe.webhooks.constructEvent(
        req.body,
        req.headers['stripe-signature'],
        process.env.STRIPE_WEBHOOK_SECRET
      );
    } catch {
      // Also throws on a replayed event outside the default 300-second
      // tolerance, which is what makes this more than a shared secret check.
      return res.status(400).send('invalid signature');
    }

    // Delivery is at-least-once. The unique index, not the handler, enforces once.
    const claim = await pool.query(
      `INSERT INTO processed_events (event_id, type) VALUES ($1, $2)
       ON CONFLICT (event_id) DO NOTHING
       RETURNING event_id`,
      [event.id, event.type]
    );
    if (claim.rowCount === 0) return res.sendStatus(200); // already handled

    if (event.type.startsWith('payment_intent.')) {
      const intent = event.data.object;
      // Events arrive out of order. Never move a payment backwards: a late
      // 'processing' must not overwrite a recorded 'succeeded'.
      await pool.query(
        `UPDATE payments
            SET status = $2, updated_at = now()
          WHERE provider_intent_id = $1
            AND $3 > COALESCE((SELECT rank FROM payment_status_rank WHERE status = payments.status), -1)`,
        [intent.id, intent.status, RANK[intent.status] ?? 0]
      );
    }

    // 200 quickly. Slow handlers get retried, which multiplies the load.
    res.sendStatus(200);
  }
);

// Reconciliation assumes both sides are wrong until proven otherwise. Three
// buckets, and only the first is boring.
import Stripe from 'stripe';
import { pool } from './db.js';

const stripe = new Stripe(process.env.STRIPE_SECRET_KEY);

export async function reconcile(fromUnix, toUnix) {
  const theirs = new Map(); // intent id -> amount in minor units

  // stripe-node auto-paginates with for await, so a busy day is one loop.
  for await (const charge of stripe.charges.list({
    created: { gte: fromUnix, lt: toUnix },
    limit: 100,
  })) {
    if (charge.status === 'succeeded') {
      theirs.set(charge.payment_intent, charge.amount);
    }
  }

  const { rows } = await pool.query(
    `SELECT provider_intent_id, status, amount_cents
       FROM payments
      WHERE created_at >= to_timestamp($1) AND created_at < to_timestamp($2)`,
    [fromUnix, toUnix]
  );
  const ours = new Map(rows.map((r) => [r.provider_intent_id, r]));

  const report = { matched: [], theirsOnly: [], oursOnly: [], amountMismatch: [] };

  for (const [intentId, amount] of theirs) {
    const mine = ours.get(intentId);
    // Charged by them, unknown to us: exactly the timeout case. Either record
    // the payment and fulfil, or refund. Never ignore.
    if (!mine) { report.theirsOnly.push({ intentId, amount }); continue; }
    if (mine.amount_cents !== amount) { report.amountMismatch.push({ intentId }); continue; }
    report.matched.push(intentId);
  }

  for (const [intentId, mine] of ours) {
    // We believe it succeeded and they have no record: a bug, or the wrong
    // account, or a test key in production. Page someone.
    if (mine.status === 'succeeded' && !theirs.has(intentId)) {
      report.oursOnly.push({ intentId });
    }
  }

  return report;
}