Sagas and compensating actions

Saga state lives in a table next to your journal.

CREATE TYPE saga_status AS ENUM (
  'initiated', 'committed', 'failed', 'compensated'
);

CREATE TABLE wire_saga (
  id                uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id           uuid NOT NULL,
  amount            bigint NOT NULL,
  status            saga_status NOT NULL DEFAULT 'initiated',
  initiate_entry_id uuid REFERENCES journal_entry (id),
  compensate_entry_id uuid REFERENCES journal_entry (id),
  last_error        text,
  updated_at        timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX wire_saga_status_idx ON wire_saga (status)
  WHERE status IN ('initiated', 'failed');

The compensation path. New entry, NOT a delete or rollback.

import { Client, DatabaseError } from "pg";

export async function executeWire(client: Client, sagaId: string): Promise<void> {
  const { rows: [saga] } = await client.query<{
    user_id: string; amount: string;
  }>(`SELECT user_id, amount FROM wire_saga WHERE id = $1`, [sagaId]);

  // Step 1 already committed by the API handler. We're picking up here.
  try {
    await sponsorBank.initiateWire({ amount: BigInt(saga.amount), userId: saga.user_id });
    await client.query(
      `UPDATE wire_saga SET status = 'committed', updated_at = now() WHERE id = $1`,
      [sagaId],
    );
  } catch (err) {
    // The sponsor failed. Compensate: post the inverse of step 1 in a
    // fresh transaction, then mark the saga compensated.
    await compensate(client, sagaId, BigInt(saga.amount), saga.user_id, err);
  }
}

async function compensate(
  client: Client,
  sagaId: string,
  amount: bigint,
  userId: string,
  err: unknown,
): Promise<void> {
  try {
    await client.query("BEGIN");

    const { rows } = await client.query<{ id: string }>(
      `INSERT INTO journal_entry (description) VALUES ($1) RETURNING id`,
      [`Compensate failed wire NGN${amount}`],
    );
    const entryId = rows[0].id;

    await client.query(
      `INSERT INTO journal_line (entry_id, account_code, debit, credit) VALUES
         ($1, '2500', $2, 0),
         ($1, '2000', 0, $2)`,
      [entryId, amount],
    );

    await client.query(
      `UPDATE wire_saga
          SET status = 'compensated',
              compensate_entry_id = $1,
              last_error = $2,
              updated_at = now()
        WHERE id = $3`,
      [entryId, err instanceof Error ? err.message : String(err), sagaId],
    );

    await client.query("COMMIT");
  } catch (compensateErr) {
    await client.query("ROLLBACK");
    throw compensateErr;
  }
}