A Payment Webhook

NODE_ENV=development
# The secret your payment provider signs its webhook requests with (modules/payments): `verifyWebhook('payments')` refuses anything else.
PAYMENTS_WEBHOOK_SECRET=replace-with-the-secret-the-provider-shows

NODE_ENV=development
# The secret your payment provider signs its webhook requests with (modules/payments): `verifyWebhook('payments')` refuses anything else.
PAYMENTS_WEBHOOK_SECRET=replace-with-the-secret-the-provider-shows

import { defineConfig, env, type ConfigOf } from '@marmeon/core';

/** The secret a payment provider signs its requests with — shown in the provider's dashboard, kept in the environment. */
export const PaymentsConfig = defineConfig('payments', {
  env: env({
    PAYMENTS_WEBHOOK_SECRET: env.string().required("Missing — copy the signing secret of the endpoint from the provider's dashboard"),
  }),
  resolve: (e) => ({ webhookSecret: e.PAYMENTS_WEBHOOK_SECRET }),
});
export type PaymentsConfig = ConfigOf<typeof PaymentsConfig>;

import { clientKey, Limit, RateLimiter } from '@marmeon/cache';
import type { Application, ServiceProvider } from '@marmeon/core';
import { Webhooks } from '@marmeon/http';
import { PaymentsConfig } from './config.ts';

export class PaymentsServiceProvider implements ServiceProvider {
  boot(app: Application): void {
    // The sender's name is what the route says in `verifyWebhook('payments')`. The secret comes from the environment, never from the route file.
    app.container.make(Webhooks).define('payments', { scheme: 'stripe', secrets: app.container.make(PaymentsConfig).webhookSecret });
    // Forged requests count too, so a client that guesses signatures is slowed down; a real sender sends far fewer. Size the limit to the provider's peak: it sends from few addresses.
    app.container.make(RateLimiter).for('webhooks.payments', (ctx) => Limit.perMinute(120).by(clientKey(ctx.ip)));
  }
}

import { defineModule } from '@marmeon/core';
import { PaymentsConfig } from './config.ts';
import { IssueReceiptJob } from './jobs/IssueReceiptJob.ts';
import { PaymentsServiceProvider } from './PaymentsServiceProvider.ts';
import routes from './routes.ts';

export default defineModule({
  name: 'payments',
  config: [PaymentsConfig],
  jobs: [IssueReceiptJob],
  migrations: new URL('./migrations/', import.meta.url),
  providers: [PaymentsServiceProvider],
  routes,
});

import { throttle } from '@marmeon/cache';
import { defineRoutes, everyone, verifyWebhook } from '@marmeon/http';
import { ReceivePaymentEventController } from './controllers/ReceivePaymentEventController.ts';

export default defineRoutes((Route) => {
  // Open to everyone, signed by the sender: no session and no token, so no `csrfExcept` either. The throttle comes first: a forged
  // request counts too.
  Route.authorize(everyone).middleware(throttle('webhooks.payments')).middleware(verifyWebhook('payments')).post('/webhooks/payments', ReceivePaymentEventController);
});

import { defineMigration } from '@marmeon/database';

export default defineMigration({
  up: (schema) =>
    schema.create('invoices', (table) => {
      table.id();
      table.string('reference', 80).unique();
      table.integer('amount_cents');
      table.timestamp('paid_at').nullable();
      table.timestamp('receipt_issued_at').nullable();
      table.timestamps();
    }),
  down: (schema) => schema.drop('invoices'),
});

import { defineMigration } from '@marmeon/database';

export default defineMigration({
  up: (schema) =>
    schema.create('webhook_events', (table) => {
      table.id();
      // The sender's id of the event. The unique index lets the first delivery in and refuses every repeat, however many arrive at once.
      table.string('event_id', 255).unique();
      table.string('type', 100);
      table.timestamps();
    }),
  down: (schema) => schema.drop('webhook_events'),
});

import { Repository, type Row } from '@marmeon/database';

export class InvoiceRepository extends Repository<'invoices'> {
  static table = 'invoices' as const;

  /** Marks the invoice with this reference as paid; the invoice, or `undefined` when there is none or it was paid already. */
  async markPaid(reference: string, paidAt: string): Promise<Row<'invoices'> | undefined> {
    // One statement: the condition `paid_at is null` sits in the UPDATE, so two events for one invoice cannot both pass a check
    // before either writes. Reading first and updating after would let both of them queue a receipt.
    return this.db.updateTable('invoices').set({ paid_at: paidAt, updated_at: paidAt }).where('reference', '=', reference).where('paid_at', 'is', null).returningAll().executeTakeFirst();
  }
}

import { Repository } from '@marmeon/database';

/** The events that were handled, by the id the sender gave them. The unique index on `event_id` is what makes a repeat visible. */
export class WebhookEventRepository extends Repository<'webhook_events'> {
  static table = 'webhook_events' as const;
}

import { Clock } from '@marmeon/core';
import { Connections } from '@marmeon/database';
import { Controller, defineRequest, type ContextOf } from '@marmeon/http';
import { Queue } from '@marmeon/queue';
import { rules as r } from '@marmeon/validation';
import { InvoiceRepository } from '../InvoiceRepository.ts';
import { IssueReceiptJob } from '../jobs/IssueReceiptJob.ts';
import { WebhookEventRepository } from '../WebhookEventRepository.ts';

/** The signed body, checked before the controller reads it. Fields the schema does not name are left out. */
export const PaymentEventRequest = defineRequest({
  schema: r.object({
    id: r.string().min(1).max(255),
    type: r.string().max(100),
    data: r.object({ invoice: r.string().max(80).optional() }).optional(),
  }),
});

/**
 * A payment provider's event. By the time this runs `verifyWebhook('payments')` has checked the signature and the time, and the
 * schema has checked the shape of the body. The provider repeats an event until it gets a `2xx`, so the effect runs once per event id.
 */
export class ReceivePaymentEventController extends Controller {
  static request = PaymentEventRequest;

  readonly #connections: Connections;
  readonly #events: WebhookEventRepository;
  readonly #invoices: InvoiceRepository;
  readonly #queue: Queue;
  readonly #clock: Clock;

  constructor(connections: Connections, events: WebhookEventRepository, invoices: InvoiceRepository, queue: Queue, clock: Clock) {
    super();
    this.#connections = connections;
    this.#events = events;
    this.#invoices = invoices;
    this.#queue = queue;
    this.#clock = clock;
  }

  async handle(ctx: ContextOf<typeof PaymentEventRequest>) {
    const event = ctx.body;
    const reference = event.data?.invoice;
    // An event this app does not act on is acknowledged, or the sender would repeat it for days.
    if (event.type !== 'invoice.paid' || reference === undefined) return this.json({ received: true });

    // One transaction: the note of the event, the payment and the queued job all count, or none does. When anything fails, the sender's
    // next delivery finds no note and does the work again. The queued job is a row in this transaction, so a worker never sees it early.
    await this.#connections.transaction(async () => {
      const first = await this.#events.tryInsert({ event_id: event.id, type: event.type });
      if (!first.ok) return; // seen before: the unique index on `event_id` refused it
      const invoice = await this.#invoices.markPaid(reference, new Date(this.#clock.now()).toISOString());
      if (invoice) await this.#queue.dispatch(IssueReceiptJob, { invoiceId: invoice.id });
    });
    return this.json({ received: true });
  }
}

import { Clock } from '@marmeon/core';
import type { JobContext, JobPayload } from '@marmeon/queue';
import { rules as r } from '@marmeon/validation';
import { InvoiceRepository } from '../InvoiceRepository.ts';

/** Issues the receipt of a paid invoice. The payload holds the invoice's id; the job reads the invoice as it is when it runs. */
export class IssueReceiptJob {
  /** Stable — the queue stores this name, never the class name. */
  static readonly job = 'payments.issue-receipt';
  static readonly schema = r.object({ invoiceId: r.integer() });
  static readonly tries = 3;
  static readonly backoff = [10, 60];

  readonly #invoices: InvoiceRepository;
  readonly #clock: Clock;

  constructor(invoices: InvoiceRepository, clock: Clock) {
    this.#invoices = invoices;
    this.#clock = clock;
  }

  async handle(payload: JobPayload<typeof IssueReceiptJob>, job: JobContext): Promise<void> {
    const invoice = await this.#invoices.find(payload.invoiceId);
    if (!invoice) return job.discard('the invoice is gone');
    // A job can run twice; the stamp is what keeps the second run from issuing another receipt.
    if (invoice.receipt_issued_at !== null) return;
    // The receipt itself (a PDF, a mail) goes here.
    await this.#invoices.update(invoice.id, { receipt_issued_at: new Date(this.#clock.now()).toISOString() });
  }
}

import { randomBytes } from 'node:crypto';
import { Queue } from '@marmeon/queue';
import { createTestApp, signedWebhook, type TestApp } from '@marmeon/testing';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import application from '../../bootstrap/app.ts';
import { InvoiceRepository } from './InvoiceRepository.ts';
import { IssueReceiptJob } from './jobs/IssueReceiptJob.ts';
import { WebhookEventRepository } from './WebhookEventRepository.ts';

// What the payment endpoint accepts, what it does once for each event, and what it leaves alone.

// A secret built at run time, so that no literal in the source looks like a real one to a scanner.
const SECRET = ['whsec', randomBytes(24).toString('hex')].join('_');

let app: TestApp;
// `csrf: true` checks requests as the app does: the sender sends no token, and the route needs none.
beforeEach(async () => {
  app = await createTestApp(application, { database: 'refresh', csrf: true, env: { PAYMENTS_WEBHOOK_SECRET: SECRET } });
}, 30_000);

const paid = (id: string, invoice = 'inv_1001') => JSON.stringify({ id, type: 'invoice.paid', data: { invoice } });
/** Sends `body` as the provider would, signed with `SECRET` unless the test passes other headers. */
const send = (body: string, headers: Record<string, string> = signedWebhook(body, { scheme: 'stripe', secret: SECRET })) =>
  app.expectingJson().post('/webhooks/payments', { body, headers: { 'content-type': 'application/json', ...headers } });
const invoice = async () => (await app.make(InvoiceRepository).findBy('reference', 'inv_1001'))!;

describe('the payment webhook', { timeout: 30_000 }, () => {
  beforeEach(async () => {
    await app.make(InvoiceRepository).insert({ reference: 'inv_1001', amount_cents: 4900 });
  });

  it('marks the invoice as paid and queues its receipt', async () => {
    await send(paid('evt_1')).assertOk().assertExactJson({ received: true });

    expect((await invoice()).paid_at).not.toBeNull();
    app.queue.assertDispatchedTimes(IssueReceiptJob, 1);
    expect(app.queue.dispatched(IssueReceiptJob)[0]!.payload).toEqual({ invoiceId: (await invoice()).id });
    expect(await app.queue.work()).toMatchObject({ processed: 1, failed: 0 });
    expect((await invoice()).receipt_issued_at).not.toBeNull();
  });

  it('does the work once when the provider sends an event again, also with a new signature', async () => {
    const body = paid('evt_1');
    await send(body).assertOk();
    const first = (await invoice()).paid_at;

    app.travel({ seconds: 60 });
    await send(body, signedWebhook(body, { scheme: 'stripe', secret: SECRET, timestamp: new Date(app.clock.now()) })).assertOk().assertExactJson({ received: true });

    app.queue.assertDispatchedTimes(IssueReceiptJob, 1);
    expect((await invoice()).paid_at).toBe(first);
    await app.assertDatabaseCount('webhook_events', 1);
  });

  it('queues no second receipt when another event reports the same invoice', async () => {
    await send(paid('evt_1')).assertOk();
    await send(paid('evt_2')).assertOk();
    app.queue.assertDispatchedTimes(IssueReceiptJob, 1);
    await app.assertDatabaseCount('webhook_events', 2);
  });

  it('acknowledges an event it does not act on, and an invoice nobody knows', async () => {
    const other = JSON.stringify({ id: 'evt_2', type: 'customer.created', data: {} });
    await send(other).assertOk().assertExactJson({ received: true });
    await send(paid('evt_3', 'inv_unknown')).assertOk();
    app.queue.assertNothingDispatched();
    expect((await invoice()).paid_at).toBeNull();
  });

  it('refuses a body that does not have the shape of an event', async () => {
    const body = JSON.stringify({ type: 'invoice.paid' });
    await send(body).assertStatus(422);
    await app.assertDatabaseCount('webhook_events', 0);
  });

  it('keeps nothing of an event whose first delivery failed, so that the next one does the work', async () => {
    const queue = app.make(Queue);
    vi.spyOn(queue, 'dispatch').mockRejectedValueOnce(new Error('the queue is down'));

    await send(paid('evt_1')).assertStatus(500);
    expect((await invoice()).paid_at).toBeNull();
    await app.assertDatabaseCount('webhook_events', 0);

    await send(paid('evt_1')).assertOk();
    expect((await invoice()).paid_at).not.toBeNull();
    app.queue.assertDispatchedTimes(IssueReceiptJob, 1);
  });

  it('issues one receipt when the job runs twice', async () => {
    await send(paid('evt_1')).assertOk();
    expect(await app.queue.work()).toMatchObject({ processed: 1 });
    const issued = (await invoice()).receipt_issued_at;

    app.travel({ seconds: 60 });
    const { invoiceId } = app.queue.dispatched(IssueReceiptJob)[0]!.payload;
    await app.make(Queue).dispatch(IssueReceiptJob, { invoiceId });
    expect(await app.queue.work()).toMatchObject({ failed: 0 });
    expect((await invoice()).receipt_issued_at).toBe(issued);
  });

  it.runIf(process.env['DB_DRIVER'] === 'pgsql')('on Postgres, two events about one invoice that arrive together queue one receipt', async () => {
    // The test runs in one transaction on one connection, so the two requests are served one after the other; the condition in the
    // UPDATE is what lets only one of them through when the requests really overlap.
    await Promise.all([send(paid('evt_1')).assertOk(), send(paid('evt_2')).assertOk()]);
    app.queue.assertDispatchedTimes(IssueReceiptJob, 1);
  });

  it('refuses a request with no signature, a wrong one, a changed body or an old time, and changes nothing', async () => {
    const refused = { error: 'webhook_signature_invalid' };
    const body = paid('evt_1');
    await send(body, {}).assertUnauthorized().assertExactJson(refused);
    await send(body, signedWebhook(body, { scheme: 'stripe', secret: `${SECRET}-other` })).assertUnauthorized().assertExactJson(refused);
    await send(paid('evt_1', 'inv_2'), signedWebhook(body, { scheme: 'stripe', secret: SECRET })).assertUnauthorized().assertExactJson(refused);
    await send(body, signedWebhook(body, { scheme: 'stripe', secret: SECRET, timestamp: new Date(app.clock.now() - 3_600_000) })).assertUnauthorized().assertExactJson(refused);

    expect((await invoice()).paid_at).toBeNull();
    app.queue.assertNothingDispatched();
    await app.make(WebhookEventRepository).query().select('id').execute().then((rows) => expect(rows).toEqual([]));
  });

  it('answers 429 once a client has sent 120 requests in a minute, signed or not', async () => {
    for (let count = 0; count < 120; count++) await send('{}', {}).assertUnauthorized();
    await send(paid('evt_1')).assertStatus(429);
    app.travel({ seconds: 61 });
    await send(paid('evt_1')).assertOk();
  });
});