Use case: a payment system

export const SCHEMA = [
  `CREATE TABLE orders (
     id serial PRIMARY KEY,
     user_id integer NOT NULL,
     total_kobo integer NOT NULL CHECK (total_kobo > 0),
     status text NOT NULL DEFAULT 'unpaid' CHECK (status IN ('unpaid', 'paid')))`,
  `CREATE TABLE payments (
     id serial PRIMARY KEY,
     user_id integer NOT NULL,
     order_id integer NOT NULL REFERENCES orders (id),
     amount_kobo integer NOT NULL CHECK (amount_kobo > 0),
     status text NOT NULL CHECK (status IN ('processing', 'succeeded', 'declined', 'unknown')),
     idempotency_key text NOT NULL,
     request_hash text NOT NULL,
     gateway_charge_id text,
     decline_reason text,
     created_at timestamptz NOT NULL DEFAULT now(),
     UNIQUE (user_id, idempotency_key))`,
  `CREATE UNIQUE INDEX one_live_payment_per_order ON payments (order_id) WHERE status <> 'declined'`,
  `CREATE TABLE outbox (
     id serial PRIMARY KEY,
     type text NOT NULL,
     payload jsonb NOT NULL,
     published_at timestamptz)`,
  `CREATE TABLE notifications (
     payment_id integer NOT NULL REFERENCES payments (id),
     kind text NOT NULL,
     sent_at timestamptz NOT NULL DEFAULT now(),
     PRIMARY KEY (payment_id, kind))`,
];

import { createHttpServer, createNodeHttpAdapter, createResponseContext, createRouter } from "@zudojs/http";

/*
 * "PayGate", a card gateway simulated for tests. The card token decides what happens:
 *   tok_ok        charged
 *   tok_declined  refused with 402 card_declined
 *   tok_flaky     503 on the first attempt, then charged
 *   tok_lost      charged, but the first response never arrives
 *   tok_down      charged, but no response ever arrives
 * Like real gateways, it charges each Idempotency-Key at most once and replays the result.
 */
export interface Charge { readonly id: string; readonly reference: string; readonly amountKobo: number; readonly status: "succeeded" | "declined" }

export async function startFakePayGate(apiKey: string) {
  const charges = new Map<string, Charge>();
  const attempts = new Map<string, number>();
  const hanging = new Set<() => void>();
  const hang = () => new Promise<void>((resolve) => hanging.add(resolve));
  const router = createRouter();

  router.post("/v1/charges", async (ctx) => {
    if (ctx.request.getHeader("authorization") !== `Bearer ${apiKey}`) return createResponseContext({ status: 401 }).json({ error: "bad_key" });
    const key = ctx.request.getHeader("idempotency-key") ?? "";
    const body = JSON.parse(new TextDecoder().decode(ctx.request.body as Uint8Array)) as { amountKobo: number; card: string; reference: string };
    const attempt = (attempts.get(key) ?? 0) + 1;
    attempts.set(key, attempt);
    if (body.card === "tok_flaky" && attempt === 1) return createResponseContext({ status: 503 }).json({ error: "try_again" });

    let charge = charges.get(key);
    if (!charge) {
      const status = body.card === "tok_declined" ? "declined" : "succeeded";
      charge = { id: `ch_${charges.size + 1}`, reference: body.reference, amountKobo: body.amountKobo, status };
      charges.set(key, charge);
    }
    if ((body.card === "tok_lost" && attempt === 1) || body.card === "tok_down") await hang();
    if (charge.status === "declined") return createResponseContext({ status: 402 }).json({ error: "card_declined", charge: charge.id });
    return createResponseContext({ status: 201 }).json(charge);
  });

  router.get("/v1/charges", (ctx) => {
    const found = [...charges.values()].find((charge) => charge.reference === ctx.query.reference);
    return found ? createResponseContext().json(found) : createResponseContext({ status: 404 }).json({ error: "not_found" });
  });

  const server = createHttpServer({
    adapter: createNodeHttpAdapter({ host: "127.0.0.1", port: 0 }),
    handler: async (request) => (await router.dispatch(request)).response,
  });
  await server.start();
  return {
    base: `http://127.0.0.1:${server.address?.port}`,
    charges,
    attempts: (key: string) => attempts.get(key) ?? 0,
    async stop() {
      for (const release of hanging) release();
      await server.stop();
    },
  };
}

import type { LifecycleAdapter } from "@zudojs/adapters";
import { createHealthyHealth, createUnhealthyHealth } from "@zudojs/adapters";
import { HttpClient, HttpClientError, HttpClientNetworkError, HttpClientTimeoutError } from "@zudojs/http";

export interface ChargeRequest {
  readonly reference: string;
  readonly amountKobo: number;
  readonly card: string;
  readonly idempotencyKey: string;
}

/** What a charge attempt can end in. "unknown" means the gateway may or may not have charged. */
export type ChargeOutcome =
  | { readonly status: "succeeded"; readonly chargeId: string }
  | { readonly status: "declined"; readonly reason: string }
  | { readonly status: "unknown"; readonly reason: string };

/** The port the payment service depends on. Any gateway that fits it can be swapped in. */
export interface PaymentGateway extends LifecycleAdapter {
  charge(request: ChargeRequest): Promise<ChargeOutcome>;
  lookup(reference: string): Promise<ChargeOutcome | null>;
}

interface PayGateCharge { readonly id: string; readonly status: "succeeded" | "declined" }

export function createPayGateAdapter(options: { baseUrl: string; apiKey: string; timeoutMs: number; retries: number }): PaymentGateway {
  const client = new HttpClient({
    baseUrl: options.baseUrl,
    timeout: options.timeoutMs,
    headers: { authorization: `Bearer ${options.apiKey}` },
    // POST is retried only because every charge carries an Idempotency-Key.
    retry: { retries: options.retries, retryDelay: 50, retryMethods: ["POST", "GET"], retryStatusCodes: [502, 503, 504] },
  });

  const toOutcome = (charge: PayGateCharge): ChargeOutcome =>
    charge.status === "succeeded" ? { status: "succeeded", chargeId: charge.id } : { status: "declined", reason: "card_declined" };

  return {
    name: "paygate",
    version: "1.0.0",
    capabilities: { http: true, abortSignal: true },

    async charge({ reference, amountKobo, card, idempotencyKey }) {
      try {
        const { data } = await client.request<PayGateCharge>("/v1/charges", {
          method: "POST",
          headers: { "idempotency-key": idempotencyKey },
          body: { reference, amountKobo, card },
        });
        return toOutcome(data);
      } catch (error) {
        if (error instanceof HttpClientTimeoutError || error instanceof HttpClientNetworkError) {
          return { status: "unknown", reason: error.name };
        }
        if (error instanceof HttpClientError && error.status === 402) return { status: "declined", reason: "card_declined" };
        if (error instanceof HttpClientError && error.status !== undefined && error.status >= 500) {
          return { status: "unknown", reason: `PayGate answered ${error.status}` };
        }
        throw error; // 401, 400: our bug or our configuration, not the customer's problem
      }
    },

    async lookup(reference) {
      try {
        const { data } = await client.request<PayGateCharge>("/v1/charges", { query: { reference } });
        return toOutcome(data);
      } catch (error) {
        if (error instanceof HttpClientError && error.status === 404) return null;
        throw error;
      }
    },

    async health() {
      try {
        await client.request("/v1/charges", { query: { reference: "health-check" }, timeout: 1000, retry: { retries: 0 } });
        return createHealthyHealth();
      } catch (error) {
        return error instanceof HttpClientError && error.status === 404 ? createHealthyHealth() : createUnhealthyHealth("PayGate unreachable");
      }
    },
  };
}

import { sha256 } from "@zudojs/crypto";
import type { DatabaseClient } from "@zudojs/database";
import { isConflictError, withTransaction } from "@zudojs/database";
import { ConflictError, DomainError, NotFoundError, ValidationError } from "@zudojs/errors";
import { schema } from "@zudojs/schema";
import type { ChargeOutcome, PaymentGateway } from "./gateway.js";

export const NewPayment = schema.object({
  orderId: schema.number().int().min(1),
  card: schema.string().regex(/^tok_[a-z0-9_]{2,60}$/),
});

export interface Payment {
  readonly id: number;
  readonly orderId: number;
  readonly amountKobo: number;
  readonly status: "processing" | "succeeded" | "declined" | "unknown";
}
interface Row extends Payment { readonly requestHash: string }

const SELECT = `SELECT id, order_id AS "orderId", amount_kobo AS "amountKobo", status, request_hash AS "requestHash" FROM payments`;
const view = ({ id, orderId, amountKobo, status }: Row): Payment => ({ id, orderId, amountKobo, status });
const KEY = /^[A-Za-z0-9_-]{16,64}$/;

export function createPayments(db: DatabaseClient, gateway: PaymentGateway, relay: () => Promise<void>) {
  async function find(where: string, values: unknown[]): Promise<Row | undefined> {
    return (await db.queryRawUnsafe<Row[]>(`${SELECT} WHERE ${where}`, values))[0];
  }

  /** Records a final outcome once. The status check makes a second, late outcome a no-op. */
  async function settle(id: number, outcome: ChargeOutcome): Promise<void> {
    await withTransaction(db, async (tx) => {
      const [row] = await tx.$queryRawUnsafe<{ order_id: number; amount_kobo: number; user_id: number }[]>(
        `UPDATE payments SET status = $2, gateway_charge_id = $3, decline_reason = $4
         WHERE id = $1 AND status IN ('processing', 'unknown') RETURNING order_id, amount_kobo, user_id`,
        id, outcome.status,
        outcome.status === "succeeded" ? outcome.chargeId : null,
        outcome.status === "succeeded" ? null : outcome.reason,
      );
      if (!row || outcome.status === "unknown") return;
      if (outcome.status === "succeeded") await tx.$executeRawUnsafe("UPDATE orders SET status = 'paid' WHERE id = $1", row.order_id);
      await tx.$executeRawUnsafe(
        "INSERT INTO outbox (type, payload) VALUES ($1, $2)",
        `payment.${outcome.status}`, { paymentId: id, orderId: row.order_id, userId: row.user_id, amountKobo: row.amount_kobo },
      );
    });
    await relay();
  }

  /**
   * Asks the gateway what happened to a payment we are unsure about. Never charges.
   * `giveUp`: the payment is old enough that "the gateway has no such charge" means it never will.
   */
  async function check(payment: Row, giveUp: boolean): Promise<void> {
    const outcome = await gateway.lookup(`pay_${payment.id}`);
    if (outcome) await settle(payment.id, outcome);
    else if (giveUp) await settle(payment.id, { status: "declined", reason: "never_reached_gateway" });
  }

  return {
    async create(userId: number, key: string | undefined, input: unknown): Promise<{ payment: Payment; replayed: boolean }> {
      if (key === undefined || !KEY.test(key)) {
        throw new ValidationError("Send an Idempotency-Key header of 16 to 64 letters, digits, - or _");
      }
      const { orderId, card } = NewPayment.parse(input);
      const requestHash = await sha256(JSON.stringify({ orderId, card }));

      const earlier = await find("user_id = $1 AND idempotency_key = $2", [userId, key]);
      if (earlier) {
        if (earlier.requestHash !== requestHash) {
          throw new DomainError("This Idempotency-Key was used for a different payment", { code: "IDEMPOTENCY_KEY_REUSED" });
        }
        if (earlier.status === "processing" || earlier.status === "unknown") await check(earlier, false);
        return { payment: view((await find("id = $1", [earlier.id]))!), replayed: true };
      }

      let created: Row;
      try {
        created = await withTransaction(db, async (tx) => {
          const [order] = await tx.$queryRawUnsafe<{ total_kobo: number; status: string }[]>(
            "SELECT total_kobo, status FROM orders WHERE id = $1 AND user_id = $2 FOR UPDATE", orderId, userId,
          );
          if (!order) throw new NotFoundError(`Order ${orderId} not found`);
          if (order.status === "paid") throw new ConflictError(`Order ${orderId} is already paid`);
          const [row] = await tx.$queryRawUnsafe<Row[]>(
            `INSERT INTO payments (user_id, order_id, amount_kobo, status, idempotency_key, request_hash)
             VALUES ($1, $2, $3, 'processing', $4, $5)
             RETURNING id, order_id AS "orderId", amount_kobo AS "amountKobo", status, request_hash AS "requestHash"`,
            userId, orderId, order.total_kobo, key, requestHash,
          );
          return row!;
        });
      } catch (error) {
        if (!isConflictError(error)) throw error;
        const same = await find("user_id = $1 AND idempotency_key = $2", [userId, key]);
        throw new ConflictError(same ? "This payment is already being processed; retry shortly" : `Order ${orderId} already has a payment`);
      }

      const outcome = await gateway.charge({
        reference: `pay_${created.id}`, amountKobo: created.amountKobo, card, idempotencyKey: `oja-pay-${created.id}`,
      });
      await settle(created.id, outcome);
      return { payment: view((await find("id = $1", [created.id]))!), replayed: false };
    },

    async get(userId: number, id: number): Promise<Payment> {
      const row = await find("id = $1 AND user_id = $2", [id, userId]);
      if (!row) throw new NotFoundError(`Payment ${id} not found`);
      return view(row);
    },

    /** The reconciliation job: settles every payment we are unsure about by asking the gateway. */
    async reconcile(olderThanSeconds: number): Promise<number> {
      const open = await db.queryRawUnsafe<Row[]>(
        `${SELECT} WHERE status IN ('processing', 'unknown') AND created_at < now() - make_interval(secs => $1::integer) ORDER BY id`,
        [olderThanSeconds],
      );
      for (const payment of open) await check(payment, true);
      return open.length;
    },
  };
}

export type Payments = ReturnType<typeof createPayments>;

import type { DatabaseClient } from "@zudojs/database";
import type { EventBus } from "@zudojs/events";

/**
 * Publishes events that were committed to the outbox table. Called right after a commit, and by a
 * periodic sweep that catches anything a crash left behind. Delivery is at least once.
 */
export function createOutboxRelay(db: DatabaseClient, bus: EventBus) {
  let running: Promise<void> | undefined;

  async function drain(): Promise<void> {
    const rows = await db.queryRawUnsafe<{ id: number; type: string; payload: Record<string, unknown> }[]>(
      "SELECT id, type, payload FROM outbox WHERE published_at IS NULL ORDER BY id LIMIT 100",
    );
    for (const row of rows) {
      const result = await bus.publishEvent({ type: row.type, payload: row.payload });
      if (result.failed > 0) return; // keep it, and everything after it, for the next sweep
      await db.executeRawUnsafe("UPDATE outbox SET published_at = now() WHERE id = $1", [row.id]);
    }
  }

  /** One drain at a time in this process; callers that arrive meanwhile wait for it. */
  return async function relay(): Promise<void> {
    while (running) await running;
    running = drain().finally(() => (running = undefined));
    await running;
  };
}

import type { DatabaseClient } from "@zudojs/database";
import { JobDuplicateError } from "@zudojs/errors";
import type { Event, EventBus } from "@zudojs/events";
import type { Queue } from "@zudojs/queue";

export interface PaymentEvent { readonly paymentId: number; readonly orderId: number; readonly userId: number; readonly amountKobo: number }
export interface NotifyJob extends PaymentEvent { readonly kind: "receipt" | "declined" }

/** The SMS provider's port. `key` lets the provider drop a message it has already sent. */
export interface SmsSender {
  send(to: string, text: string, key: string): Promise<void>;
}

const naira = (kobo: number) => `₦${(kobo / 100).toLocaleString("en-NG", { minimumFractionDigits: 2 })}`;

/** payment.* events become notification jobs, at most one per payment and kind. */
export function queueNotifications(bus: EventBus, queue: Queue<NotifyJob>): void {
  const enqueue = (kind: NotifyJob["kind"]) => async (event: Event<PaymentEvent>) => {
    try {
      await queue.add("notify", { ...event.payload, kind }, { deduplicationKey: `${kind}:${event.payload.paymentId}` });
    } catch (error) {
      if (!(error instanceof JobDuplicateError)) throw error; // already queued: nothing to do
    }
  };
  bus.on<Event<PaymentEvent>>("payment.succeeded", enqueue("receipt"), { id: "queue-receipt" });
  bus.on<Event<PaymentEvent>>("payment.declined", enqueue("declined"), { id: "queue-declined-notice" });
}

/** The worker. It may run twice for one job, so it checks, sends with a key, then records. */
export function processNotifications(queue: Queue<NotifyJob>, db: DatabaseClient, sms: SmsSender): void {
  queue.process("notify", async (job) => {
    const { paymentId, orderId, userId, amountKobo, kind } = job.data;
    const done = await db.queryRawUnsafe<unknown[]>("SELECT 1 FROM notifications WHERE payment_id = $1 AND kind = $2", [paymentId, kind]);
    if (done.length > 0) return;
    const text = kind === "receipt"
      ? `Oja Market: we received ${naira(amountKobo)} for order ${orderId}. Thank you!`
      : `Oja Market: your card was declined for order ${orderId}. No money was taken.`;
    await sms.send(`customer-${userId}`, text, `${kind}-${paymentId}`);
    await db.executeRawUnsafe("INSERT INTO notifications (payment_id, kind) VALUES ($1, $2) ON CONFLICT DO NOTHING", [paymentId, kind]);
  });
}

import type { TokenConfig } from "@zudojs/auth";
import { parseBearerToken, verifyAccessToken } from "@zudojs/auth";
import { AuthenticationError } from "@zudojs/errors";
import { badRequest, createHttpServer, createNodeHttpAdapter, createRateLimitMiddleware, createResponseContext, createRouter } from "@zudojs/http";
import type { HttpMiddleware, HttpRouterContext } from "@zudojs/http";
import type { Logger } from "@zudojs/logger";
import { schema } from "@zudojs/schema";
import { createErrorHandler } from "./errors.js";
import type { Payment, Payments } from "./payments.js";

const Id = schema.object({ id: schema.coerce.number().int().min(1).max(2_147_483_647) });
const STATUS: Record<Payment["status"], number> = { succeeded: 201, declined: 402, processing: 202, unknown: 202 };

export function createPaymentsApi(payments: Payments, tokens: TokenConfig, logger: Logger) {
  /** Access tokens come from the authentication platform; here we only verify them. */
  const requireUser: HttpMiddleware = async (ctx, next) => {
    const token = parseBearerToken(ctx.request.getHeader("authorization"));
    const result = token === null ? undefined : verifyAccessToken(token, tokens);
    if (!result?.valid || !result.payload) throw new AuthenticationError("Sign in first");
    ctx.state.set("userId", Number(result.payload.sub));
    return next();
  };
  const userId = (ctx: HttpRouterContext) => ctx.state.get("userId") as number;
  const router = createRouter();

  router.post("/v1/payments", async (ctx) => {
    let body: unknown;
    try {
      body = JSON.parse(new TextDecoder().decode(ctx.request.body as Uint8Array));
    } catch {
      throw badRequest("The body is not valid JSON");
    }
    const { payment, replayed } = await payments.create(userId(ctx), ctx.request.getHeader("idempotency-key"), body);
    const response = createResponseContext({ status: STATUS[payment.status] }).setHeader("location", `/v1/payments/${payment.id}`);
    if (replayed) response.setHeader("idempotent-replayed", "true");
    return response.json(payment);
  }, { middleware: [requireUser, createRateLimitMiddleware({ max: 30, windowMs: 60_000 })] });

  router.get("/v1/payments/:id", async (ctx) => payments.get(userId(ctx), Id.parse(ctx.params).id), { middleware: [requireUser] });

  return createHttpServer({
    adapter: createNodeHttpAdapter({ host: "127.0.0.1", port: 0 }),
    handler: async (request) => (await router.dispatch(request)).response,
    errorHandler: createErrorHandler(logger),
  });
}

import { randomBytes } from "node:crypto";
import { AdapterRegistry } from "@zudojs/adapters";
import { createTokenPair, toUserId } from "@zudojs/auth";
import { createEventBus } from "@zudojs/events";
import { createLogger } from "@zudojs/logger";
import { createExponentialBackoff, createInMemoryQueue, createQueueName } from "@zudojs/queue";
import { createPayGateAdapter, type PaymentGateway } from "./gateway.js";
import { createPaymentsApi } from "./http.js";
import { processNotifications, queueNotifications, type NotifyJob, type SmsSender } from "./notifications.js";
import { createOutboxRelay } from "./outbox.js";
import { startFakePayGate } from "./paygate-fake.js";
import { createPayments } from "./payments.js";
import { SCHEMA } from "./schema.js";
import { openDatabase } from "./sql.js";

/* DEMO ONLY: random secrets for each run. Your app reads them from the environment. */
const secret = () => randomBytes(32).toString("base64url");

export interface Reply { readonly status: number; readonly headers: Headers; readonly body: any }

/** Starts Oja's payment service with a fake PayGate and a fake SMS provider. */
export async function startOjaPay(options: { log?: (line: string) => void; sms?: SmsSender; gatewayTimeoutMs?: number } = {}) {
  const log = options.log ?? console.log;
  const apiKey = secret();
  const paygate = await startFakePayGate(apiKey);
  const registry = new AdapterRegistry();
  registry.register(createPayGateAdapter({ baseUrl: paygate.base, apiKey, timeoutMs: options.gatewayTimeoutMs ?? 1500, retries: 2 }));
  await registry.initializeAll();
  const gateway = registry.require<PaymentGateway>("paygate");

  const db = await openDatabase(SCHEMA);
  const bus = createEventBus();
  const queue = createInMemoryQueue<NotifyJob>(createQueueName("notifications"), {
    defaultJobOptions: { attempts: 5, backoff: createExponentialBackoff(20, { jitter: "none" }) },
  });
  const texts: string[] = [];
  const sms: SmsSender = options.sms ?? { send: async (to, text) => void texts.push(`${to}: ${text}`) };
  queueNotifications(bus, queue);
  processNotifications(queue, db, sms);

  const relay = createOutboxRelay(db, bus);
  const payments = createPayments(db, gateway, relay);
  const tokens = { accessSecret: secret(), refreshSecret: secret() };
  const logger = createLogger({
    name: "oja-pay",
    transports: [(entry) => log(`[${entry.levelName}] ${entry.message}: ${(entry.metadata as { error?: { message: string } }).error?.message}`)],
  });
  const server = createPaymentsApi(payments, tokens, logger);
  await server.start();
  const base = `http://127.0.0.1:${server.address?.port}`;

  /** An order for a user, as the orders service would have created it. */
  async function order(userId: number, totalKobo: number): Promise<number> {
    const [row] = await db.queryRawUnsafe<{ id: number }[]>("INSERT INTO orders (user_id, total_kobo) VALUES ($1, $2) RETURNING id", [userId, totalKobo]);
    return row!.id;
  }
  const tokenFor = (userId: number) => createTokenPair(toUserId(String(userId)), tokens).accessToken;

  async function pay(userId: number, key: string | undefined, body: unknown): Promise<Reply> {
    const response = await fetch(`${base}/v1/payments`, {
      method: "POST",
      headers: { "content-type": "application/json", authorization: `Bearer ${tokenFor(userId)}`, ...(key ? { "idempotency-key": key } : {}) },
      body: JSON.stringify(body),
    });
    return { status: response.status, headers: response.headers, body: await response.json() };
  }

  /** Waits until the queue has nothing left to run. */
  async function settled(): Promise<void> {
    for (let i = 0; i < 200; i++) {
      const stats = await queue.getStats();
      if (stats.waiting + stats.active + stats.delayed + stats.retrying === 0) return;
      await new Promise((resolve) => setTimeout(resolve, 20));
    }
  }

  async function stop(): Promise<void> {
    await server.stop();
    await queue.close();
    await registry.disposeAll();
    await paygate.stop();
    await db.disconnect();
  }

  return { db, base, paygate, payments, registry, queue, texts, relay, order, pay, tokenFor, settled, stop };
}

import { randomUUID } from "node:crypto";
import { startOjaPay, type Reply } from "./oja-pay.js";

const oja = await startOjaPay();
const show = (label: string, r: Reply) =>
  console.log(`${label.padEnd(22)} ${r.status}`, r.body.error ? r.body.error.code : `${r.body.status} ₦${r.body.amountKobo / 100}`,
    r.headers.get("idempotent-replayed") ? "(replayed)" : "");

const ada = 1;
const rice = await oja.order(ada, 1_250_000);
const key = randomUUID();
show("pay for the rice", await oja.pay(ada, key, { orderId: rice, card: "tok_ok" }));
show("same request again", await oja.pay(ada, key, { orderId: rice, card: "tok_ok" }));
show("same key, other card", await oja.pay(ada, key, { orderId: rice, card: "tok_other" }));
show("new key, paid order", await oja.pay(ada, randomUUID(), { orderId: rice, card: "tok_ok" }));
show("no key", await oja.pay(ada, undefined, { orderId: rice, card: "tok_ok" }));
show("someone else's order", await oja.pay(2, randomUUID(), { orderId: rice, card: "tok_ok" }));

const oil = await oja.order(ada, 280_000);
show("declined card", await oja.pay(ada, randomUUID(), { orderId: oil, card: "tok_declined" }));
show("flaky gateway", await oja.pay(ada, randomUUID(), { orderId: oil, card: "tok_flaky" }));

await oja.settled();
console.log("charges at PayGate:", [...oja.paygate.charges.values()].map((c) => `${c.reference} ${c.status}`));
console.log(oja.texts);
await oja.stop();

pay for the rice       201 succeeded ₦12500
same request again     201 succeeded ₦12500 (replayed)
same key, other card   422 IDEMPOTENCY_KEY_REUSED
new key, paid order    409 ERR_CONFLICT
no key                 400 ERR_VALIDATION_FAILED
someone else's order   404 ERR_RESOURCE_NOT_FOUND
declined card          402 declined ₦2800
flaky gateway          201 succeeded ₦2800
charges at PayGate: [ 'pay_1 succeeded', 'pay_2 declined', 'pay_3 succeeded' ]
[
  'customer-1: Oja Market: we received ₦12,500.00 for order 1. Thank you!',
  'customer-1: Oja Market: your card was declined for order 2. No money was taken.',
  'customer-1: Oja Market: we received ₦2,800.00 for order 2. Thank you!'
]

import { randomUUID } from "node:crypto";
import { startOjaPay } from "./oja-pay.js";

const oja = await startOjaPay();
const ada = 1;
const statusOf = async (id: number) =>
  (await oja.db.queryRawUnsafe<{ status: string }[]>("SELECT status FROM payments WHERE id = $1", [id]))[0]?.status;

// 1. The gateway charges, but its first answer is lost. Our retry carries the same Idempotency-Key.
const lost = await oja.pay(ada, randomUUID(), { orderId: await oja.order(ada, 50_000), card: "tok_lost" });
console.log("lost answer:", lost.status, lost.body.status, "| PayGate attempts:", oja.paygate.attempts("oja-pay-1"), "charges:", oja.paygate.charges.size);

// 2. The gateway charges, and no answer ever arrives.
const down = await oja.pay(ada, randomUUID(), { orderId: await oja.order(ada, 80_000), card: "tok_down" });
console.log("no answer:  ", down.status, down.body.status);
console.log("reconciled:", await oja.payments.reconcile(0), "payment(s) ->", await statusOf(down.body.id));

// 3. The charge works, then our own database fails while recording it.
await oja.db.executeRawUnsafe("ALTER TABLE outbox RENAME TO outbox_offline");
const key = randomUUID();
const orderId = await oja.order(ada, 120_000);
const crashed = await oja.pay(ada, key, { orderId, card: "tok_ok" });
console.log("db failure: ", crashed.status, crashed.body.error.code, "| payment 3 is", await statusOf(3));
await oja.db.executeRawUnsafe("ALTER TABLE outbox_offline RENAME TO outbox");
const retried = await oja.pay(ada, key, { orderId, card: "tok_ok" });
console.log("app retries:", retried.status, retried.body.status, retried.headers.get("idempotent-replayed"));

await oja.settled();
console.log("charges:", [...oja.paygate.charges.values()].map((c) => c.reference).join(", "));
console.log(oja.texts.length, "texts:", oja.texts.map((text) => text.match(/₦[\d,.]+/)?.[0]));
await oja.stop();

lost answer: 201 succeeded | PayGate attempts: 2 charges: 1
no answer:   202 unknown
reconciled: 1 payment(s) -> succeeded
[error] Request failed: relation "outbox" does not exist
db failure:  500 INTERNAL | payment 3 is processing
app retries: 201 succeeded true
charges: pay_1, pay_2, pay_3
3 texts: [ '₦500.00', '₦800.00', '₦1,200.00' ]

import { defineConfig } from "vitest/config";

export default defineConfig({ test: { include: ["tests/**/*.test.ts"], testTimeout: 30_000, hookTimeout: 50_000 } });

import { startVitest } from "vitest/node";

/** Runs test files with Vitest and prints one line per test. */
export async function runTests(...files: string[]): Promise<void> {
  const vitest = await startVitest("test", files, { watch: false, reporters: [] });
  for (const file of vitest.state.getTestModules()) {
    for (const error of file.errors()) console.log(`× ${file.relativeModuleId}: ${error.message}`);
    for (const test of file.children.allTests()) {
      const { state, errors = [] } = test.result();
      console.log(`${state === "passed" ? "✓" : "×"} ${test.fullName}`);
      for (const error of errors) console.log(`    ${error.message}`);
    }
  }
  await vitest.close();
}

import { randomUUID } from "node:crypto";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { startOjaPay } from "../oja-pay.js";

let oja: Awaited<ReturnType<typeof startOjaPay>>;
beforeAll(async () => {
  oja = await startOjaPay({ log: () => {} });
});
afterAll(() => oja.stop());

const chargesFor = (orderId: number) =>
  oja.db.queryRawUnsafe<{ n: number }[]>("SELECT count(*)::int AS n FROM payments WHERE order_id = $1 AND status = 'succeeded'", [orderId]);

describe("double submission", () => {
  it("replays a repeated request instead of charging again", async () => {
    const orderId = await oja.order(1, 50_000);
    const key = randomUUID();
    const first = await oja.pay(1, key, { orderId, card: "tok_ok" });
    const second = await oja.pay(1, key, { orderId, card: "tok_ok" });
    expect([first.status, second.status, second.headers.get("idempotent-replayed")]).toEqual([201, 201, "true"]);
    expect(second.body).toEqual(first.body);
    expect(oja.paygate.attempts(`oja-pay-${first.body.id}`)).toBe(1);
  });

  it("lets exactly one of two simultaneous requests charge, whatever their keys", async () => {
    const sameKey = await oja.order(1, 60_000);
    const key = randomUUID();
    const a = await Promise.all([1, 2].map(() => oja.pay(1, key, { orderId: sameKey, card: "tok_ok" })));
    const otherKeys = await oja.order(1, 70_000);
    const b = await Promise.all([1, 2].map(() => oja.pay(1, randomUUID(), { orderId: otherKeys, card: "tok_ok" })));
    for (const pair of [a, b]) {
      // The loser gets 409 (still in progress, or order already paid) or, if it came late enough, the replay.
      expect(pair.filter((r) => r.status === 201 && !r.headers.get("idempotent-replayed"))).toHaveLength(1);
      expect(pair.every((r) => r.status === 201 || r.status === 409)).toBe(true);
    }
    expect([await chargesFor(sameKey), await chargesFor(otherKeys)]).toEqual([[{ n: 1 }], [{ n: 1 }]]);
  });

  it("refuses a key reused for another payment, and a request without a key", async () => {
    const orderId = await oja.order(1, 10_000);
    const key = randomUUID();
    await oja.pay(1, key, { orderId, card: "tok_declined" });
    expect((await oja.pay(1, key, { orderId, card: "tok_ok" })).status).toBe(422);
    expect((await oja.pay(1, undefined, { orderId, card: "tok_ok" })).status).toBe(400);
  });

  it("charges the order's total, never an amount from the client, and only the owner's order", async () => {
    const orderId = await oja.order(1, 99_900);
    const paid = await oja.pay(1, randomUUID(), { orderId, card: "tok_ok", amountKobo: 1 });
    expect(paid.body.amountKobo).toBe(99_900);
    expect([...oja.paygate.charges.values()].at(-1)?.amountKobo).toBe(99_900);
    expect((await oja.pay(2, randomUUID(), { orderId: await oja.order(1, 5_000), card: "tok_ok" })).status).toBe(404);
  });
});

import { runTests } from "./vitest-run.js";

await runTests("tests/double.test.ts");

✓ double submission > replays a repeated request instead of charging again
✓ double submission > lets exactly one of two simultaneous requests charge, whatever their keys
✓ double submission > refuses a key reused for another payment, and a request without a key
✓ double submission > charges the order's total, never an amount from the client, and only the owner's order

import { randomUUID } from "node:crypto";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { startOjaPay } from "../oja-pay.js";

let smsCalls = 0;
const delivered = new Map<string, string>();
let oja: Awaited<ReturnType<typeof startOjaPay>>;
beforeAll(async () => {
  oja = await startOjaPay({
    log: () => {},
    // An SMS provider that fails twice, then works, and drops a message whose key it has seen.
    sms: {
      send: async (_to, text, key) => {
        if (++smsCalls <= 2) throw new Error("SMS provider unavailable");
        delivered.set(key, text);
      },
    },
  });
});
afterAll(() => oja.stop());

const statusOf = async (id: number) =>
  (await oja.db.queryRawUnsafe<{ status: string }[]>("SELECT status FROM payments WHERE id = $1", [id]))[0]?.status;

describe("failures", () => {
  it("retries a lost answer with the same key and charges once", async () => {
    const paid = await oja.pay(1, randomUUID(), { orderId: await oja.order(1, 10_000), card: "tok_lost" });
    expect(paid.body.status).toBe("succeeded");
    expect(oja.paygate.attempts(`oja-pay-${paid.body.id}`)).toBe(2);
    expect([...oja.paygate.charges.values()].filter((c) => c.reference === `pay_${paid.body.id}`)).toHaveLength(1);
  });

  it("never calls a timed-out payment failed, and settles it by asking the gateway", async () => {
    const pending = await oja.pay(1, randomUUID(), { orderId: await oja.order(1, 20_000), card: "tok_down" });
    expect([pending.status, pending.body.status]).toEqual([202, "unknown"]);
    await oja.payments.reconcile(0);
    expect(await statusOf(pending.body.id)).toBe("succeeded");
  });

  it("recovers from a database failure after the charge without charging twice", async () => {
    await oja.db.executeRawUnsafe("ALTER TABLE outbox RENAME TO outbox_offline");
    const key = randomUUID();
    const orderId = await oja.order(1, 30_000);
    expect((await oja.pay(1, key, { orderId, card: "tok_ok" })).status).toBe(500);
    await oja.db.executeRawUnsafe("ALTER TABLE outbox_offline RENAME TO outbox");
    const retried = await oja.pay(1, key, { orderId, card: "tok_ok" });
    expect(retried.body.status).toBe("succeeded");
    expect(oja.paygate.attempts(`oja-pay-${retried.body.id}`)).toBe(1);
  });

  it("delivers every notification once, even though the SMS provider failed", async () => {
    await oja.settled();
    const succeeded = await oja.db.queryRawUnsafe<{ id: number }[]>("SELECT id FROM payments WHERE status = 'succeeded' ORDER BY id");
    expect([...delivered.keys()].sort()).toEqual(succeeded.map((p) => `receipt-${p.id}`).sort());
    expect(await oja.db.queryRawUnsafe("SELECT count(*)::int AS n FROM outbox WHERE published_at IS NULL")).toEqual([{ n: 0 }]);
  });

  it("declines a payment the gateway never received, once it is old enough", async () => {
    const [row] = await oja.db.queryRawUnsafe<{ id: number }[]>(
      `INSERT INTO payments (user_id, order_id, amount_kobo, status, idempotency_key, request_hash)
       VALUES (1, $1, 1000, 'processing', 'crashed-before-the-charge', 'x') RETURNING id`,
      [await oja.order(1, 1000)],
    );
    await oja.payments.reconcile(0);
    expect(await statusOf(row!.id)).toBe("declined");
  });
});

import { runTests } from "./vitest-run.js";

await runTests("tests/failures.test.ts");

✓ failures > retries a lost answer with the same key and charges once
✓ failures > never calls a timed-out payment failed, and settles it by asking the gateway
✓ failures > recovers from a database failure after the charge without charging twice
✓ failures > delivers every notification once, even though the SMS provider failed
✓ failures > declines a payment the gateway never received, once it is old enough