Idempotency and Exactly-Once

// The key must be generated BEFORE the first attempt and reused by every
// retry. Generating it inside the retry loop defeats the whole mechanism.
import { randomUUID } from "node:crypto";
// randomUUID has been in node:crypto since Node 14.17; since Node 19 the same
// function is also reachable as the global crypto.randomUUID(). Same function,
// two spellings - the global form is the shorthand you will see in newer code.

export async function charge(body, { attempts = 4 } = {}) {
  const idempotencyKey = randomUUID(); // once, for this logical operation

  for (let attempt = 1; ; attempt++) {
    try {
      const res = await fetch("https://api.example.com/charges", {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          "Idempotency-Key": idempotencyKey,
        },
        body: JSON.stringify(body), // byte-identical on every retry
        signal: AbortSignal.timeout(10_000),
      });

      if (res.status === 409) throw new RetryableError("in progress");
      if (res.status >= 500) throw new RetryableError(`upstream ${res.status}`);
      return await res.json();
    } catch (err) {
      if (attempt >= attempts) throw err;
      // Jittered backoff: a 409 means the first attempt is still running.
      await new Promise((r) => setTimeout(r, Math.random() * 200 * 2 ** attempt));
    }
  }
}

// SQS and Kafka both deliver at-least-once. Dedup where the effect lands, in
// the same transaction, so a crash cannot split the two apart.
//
// CREATE TABLE processed_messages (
//   message_id  text PRIMARY KEY,
//   consumer    text        NOT NULL,
//   processed_at timestamptz NOT NULL DEFAULT now()
// );
export async function handleMessage(pool, message) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");

    const seen = await client.query(
      `INSERT INTO processed_messages (message_id, consumer)
       VALUES ($1, 'invoice-emailer') ON CONFLICT (message_id) DO NOTHING
       RETURNING message_id`,
      [message.MessageId],
    );
    if (seen.rowCount === 0) {
      await client.query("ROLLBACK");
      return { skipped: "duplicate" }; // then delete the message from the queue
    }

    await applyEffect(client, JSON.parse(message.Body)); // database work only
    await client.query("COMMIT");
    return { applied: true };
  } catch (err) {
    await client.query("ROLLBACK").catch(() => {});
    throw err; // leave the message on the queue; the retry is now safe
  } finally {
    client.release();
  }
}

// Note what cannot go inside the transaction: sending the actual email. Queue
// that as an outbox row and let a second worker send it with its own key.

// Sometimes you do not need a key at all: make the write a compare-and-set so
// the second attempt affects zero rows.

// 1. SQL compare-and-set on a state machine. The WHERE clause is the guard.
const { rowCount } = await pool.query(
  `UPDATE payouts SET status = 'sent', sent_at = now()
   WHERE id = $1 AND status = 'pending'`,
  [payoutId],
);
if (rowCount === 0) {
  // Already sent by an earlier attempt. Not an error - the desired state holds.
  return { alreadySent: true };
}
await bank.send(payoutId); // only the attempt that won the transition does this

// 2. The same idea in DynamoDB: the condition fails on the replay.
import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
try {
  await ddb.send(new PutItemCommand({
    TableName: "payouts",
    Item: { id: { S: payoutId }, status: { S: "sent" } },
    ConditionExpression: "attribute_not_exists(id)", // first writer only
  }));
} catch (err) {
  if (err.name !== "ConditionalCheckFailedException") throw err;
  return { alreadySent: true };
}