BullMQ job idempotency via dedupe id

import IORedis from "ioredis";

export interface StoredResult {
  chargeId: string;
}

type BeginOutcome =
  | { status: "cached"; result: StoredResult }
  | { status: "locked"; token: string }
  | { status: "busy" };

export class IdempotencyStore {
  constructor(private redis: IORedis, private ttlMs = 60_000) {}

  private lockKey(key: string) { return `lock:${key}`; }
  private resultKey(key: string) { return `result:${key}`; }

  async begin(key: string, token: string): Promise<BeginOutcome> {
    const existing = await this.redis.get(this.resultKey(key));
    if (existing) {
      return { status: "cached", result: JSON.parse(existing) as StoredResult };
    }
    const acquired = await this.redis.set(this.lockKey(key), token, "PX", this.ttlMs, "NX");
    return acquired ? { status: "locked", token } : { status: "busy" };
  }

  async complete(key: string, token: string, result: StoredResult): Promise<void> {
    await this.redis.set(this.resultKey(key), JSON.stringify(result));
    await this.release(key, token);
  }

  private static RELEASE = `
    if redis.call("get", KEYS[1]) == ARGV[1] then
      return redis.call("del", KEYS[1])
    else
      return 0
    end`;

  async release(key: string, token: string): Promise<void> {
    await this.redis.eval(IdempotencyStore.RELEASE, 1, this.lockKey(key), token);
  }
}

import { Worker, DelayedError, Job } from "bullmq";
import IORedis from "ioredis";
import { randomUUID } from "crypto";
import { connection, ChargePayload } from "./queue";
import { IdempotencyStore } from "./idempotency";
import { chargeCustomer } from "./payments";

const redis = new IORedis(connection);
const store = new IdempotencyStore(redis);

async function process(job: Job<ChargePayload>) {
  const key = `pi:${job.data.paymentIntentId}`;
  const token = randomUUID();

  const begin = await store.begin(key, token);
  if (begin.status === "cached") {
    return begin.result; // already charged on a prior attempt
  }
  if (begin.status === "busy") {
    await job.moveToDelayed(Date.now() + 5000, job.token);
    throw new DelayedError();
  }

  try {
    const charge = await chargeCustomer(job.data.paymentIntentId, job.data.amountCents, job.data.currency);
    const result = { chargeId: charge.id };
    await store.complete(key, begin.token, result);
    return result;
  } catch (err) {
    await store.release(key, begin.token);
    throw err;
  }
}

export const chargeWorker = new Worker<ChargePayload>("charges", process, {
  connection,
  concurrency: 8,
});

chargeWorker.on("failed", (job, err) => {
  console.error(`charge job ${job?.id} failed: ${err.message}`);
});