Building Idempotency Keys and Dedupe Stores in TypeScript with Antigravity

// src/lib/idempotency/kv-store.ts
import type { KVNamespace } from "@cloudflare/workers-types";
 
export class KvDedupeStore {
  constructor(private readonly kv: KVNamespace, private readonly ttlSec = 86400) {}
 
  /**
   * Returns true if we acquired the slot (=process this request),
   * false if the key already exists (=duplicate).
   * KV has no compare-and-swap; this is best-effort. Use Postgres or
   * Durable Objects when you need strict exactly-once.
   */
  async tryAcquire(key: string, payload: { startedAt: number }): Promise<boolean> {
    const existing = await this.kv.get(key);
    if (existing !== null) return false;
    await this.kv.put(key, JSON.stringify(payload), { expirationTtl: this.ttlSec });
    return true;
  }
 
  async markCompleted(key: string, result: unknown): Promise<void> {
    const value = JSON.stringify({ status: "completed", result, completedAt: Date.now() });
    await this.kv.put(key, value, { expirationTtl: this.ttlSec });
  }
 
  async getRecord<T>(key: string): Promise<T | null> {
    const v = await this.kv.get(key, "json");
    return (v as T) ?? null;
  }
}

// src/lib/idempotency/durable-object.ts
export class DedupeDO {
  private state: DurableObjectState;
  constructor(state: DurableObjectState) {
    this.state = state;
  }
 
  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const key = url.searchParams.get("k");
    if (!key) return new Response("missing key", { status: 400 });
 
    if (request.method === "POST" && url.pathname === "/acquire") {
      // Storage inside a DO is always strongly consistent.
      const existing = await this.state.storage.get(key);
      if (existing) return Response.json({ acquired: false });
      await this.state.storage.put(key, { status: "in_progress", at: Date.now() });
      return Response.json({ acquired: true });
    }
 
    if (request.method === "POST" && url.pathname === "/complete") {
      const body = (await request.json()) as { result: unknown };
      await this.state.storage.put(key, {
        status: "completed",
        result: body.result,
        at: Date.now(),
      });
      return Response.json({ ok: true });
    }
 
    return new Response("not found", { status: 404 });
  }
}
 
async function withDedupe(env: Env, key: string, fn: () => Promise<unknown>) {
  const id = env.DEDUPE_DO.idFromName(key); // same key -> same instance
  const stub = env.DEDUPE_DO.get(id);
  const acq = await stub.fetch(`https://internal/acquire?k=${encodeURIComponent(key)}`, {
    method: "POST",
  });
  const { acquired } = (await acq.json()) as { acquired: boolean };
  if (!acquired) return { skipped: true };
  const result = await fn();
  await stub.fetch(`https://internal/complete?k=${encodeURIComponent(key)}`, {
    method: "POST",
    body: JSON.stringify({ result }),
  });
  return { skipped: false, result };
}

// src/lib/idempotency/redis-store.ts
import { Redis } from "@upstash/redis";
 
export class RedisDedupeStore {
  constructor(private readonly redis: Redis, private readonly ttlSec = 86400) {}
 
  /**
   * Atomic reservation via SET NX EX.
   *   "acquired"     -> first writer; run the work
   *   "in_progress"  -> another worker is processing; poll or 200-fast
   *   "completed"    -> already done; return the cached response
   */
  async tryAcquire(key: string): Promise<"acquired" | "in_progress" | "completed"> {
    const set = await this.redis.set(
      key,
      JSON.stringify({ status: "in_progress", at: Date.now() }),
      { nx: true, ex: this.ttlSec },
    );
    if (set === "OK") return "acquired";
 
    const existing = await this.redis.get<{ status: string }>(key);
    if (existing?.status === "completed") return "completed";
    return "in_progress";
  }
 
  async markCompleted(key: string, result: unknown): Promise<void> {
    await this.redis.set(
      key,
      JSON.stringify({ status: "completed", result, at: Date.now() }),
      { ex: this.ttlSec },
    );
  }
 
  async getResult<T>(key: string): Promise<T | null> {
    const v = await this.redis.get<{ status: string; result?: T }>(key);
    return v?.status === "completed" ? (v.result ?? null) : null;
  }
}

-- migrations/idempotency.sql
CREATE TABLE idempotency_records (
  key             TEXT PRIMARY KEY,
  status          TEXT NOT NULL CHECK (status IN ('in_progress','completed','failed')),
  request_hash    TEXT NOT NULL,
  response_body   JSONB,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  expires_at      TIMESTAMPTZ NOT NULL DEFAULT (now() + INTERVAL '24 hours')
);
 
CREATE INDEX idx_idem_expires ON idempotency_records(expires_at);

// src/lib/idempotency/postgres-store.ts
import type { Pool } from "pg";
 
export class PgDedupeStore {
  constructor(private readonly pool: Pool) {}
 
  async tryAcquire(key: string, requestHash: string) {
    // Only the row that wins INSERT is "the reserver". Others are no-ops.
    const inserted = await this.pool.query(
      `INSERT INTO idempotency_records (key, status, request_hash)
       VALUES ($1, 'in_progress', $2)
       ON CONFLICT (key) DO NOTHING
       RETURNING key`,
      [key, requestHash],
    );
 
    if (inserted.rowCount === 1) return { acquired: true } as const;
 
    const { rows } = await this.pool.query(
      `SELECT status, response_body, request_hash FROM idempotency_records WHERE key = $1`,
      [key],
    );
    const r = rows[0];
    if (r.request_hash !== requestHash) {
      // Same key + different body -> client bug; surface it loudly.
      throw new Error("Idempotency-Key reused with different request body");
    }
    return {
      acquired: false,
      status: r.status as "in_progress" | "completed" | "failed",
      response: r.response_body,
    } as const;
  }
 
  async markCompleted(key: string, response: unknown) {
    await this.pool.query(
      `UPDATE idempotency_records
         SET status = 'completed', response_body = $2, updated_at = now()
       WHERE key = $1`,
      [key, response],
    );
  }
}

// idempotency-race.mjs — run with `node idempotency-race.mjs` on Node 22+
import { DatabaseSync } from "node:sqlite";
 
const db = new DatabaseSync(":memory:");
db.exec(`CREATE TABLE idem (key TEXT PRIMARY KEY, status TEXT NOT NULL)`);
 
const sel = db.prepare(`SELECT key FROM idem WHERE key = ?`);
const ins = db.prepare(`INSERT INTO idem (key, status) VALUES (?, 'in_progress')`);
const insIgnore = db.prepare(`INSERT OR IGNORE INTO idem (key, status) VALUES (?, 'in_progress')`);
 
// Stands in for store I/O. Any await hands control to the other in-flight requests.
const tick = () => new Promise((r) => setImmediate(r));
const kv = new Map();
 
let executed = 0;
let failed = 0;
 
// 1. Read-then-write against KV — what a plain Cloudflare KV or Redis GET/SET looks like
async function kvReadThenWrite(key) {
  const hit = kv.get(key);
  await tick();
  if (hit) return "skipped";
  kv.set(key, "in_progress");
  executed++;
  return "executed";
}
 
// 2. Read-then-write against SQL — a primary key exists, but the check is not atomic
async function sqlReadThenWrite(key) {
  const hit = sel.get(key);
  await tick();
  if (hit) return "skipped";
  try {
    ins.run(key);
  } catch {
    failed++;
    return "500";
  }
  executed++;
  return "executed";
}
 
// 3. Atomic reservation — the INSERT ... ON CONFLICT DO NOTHING equivalent
async function atomicReserve(key) {
  const r = insIgnore.run(key);
  await tick();
  if (r.changes === 0) return "skipped";
  executed++;
  return "executed";
}
 
async function run(fn, n) {
  executed = 0;
  failed = 0;
  db.exec("DELETE FROM idem");
  kv.clear();
  const rs = await Promise.all(Array.from({ length: n }, () => fn("evt_dup_001")));
  return `executed=${executed} skipped=${rs.filter((x) => x === "skipped").length} http500=${failed}`;
}
 
for (const n of [2, 10, 50]) {
  console.log(`N=${n} KV     : ${await run(kvReadThenWrite, n)}`);
  console.log(`N=${n} SQL    : ${await run(sqlReadThenWrite, n)}`);
  console.log(`N=${n} atomic : ${await run(atomicReserve, n)}`);
}

// src/lib/idempotency/redis-store.ts (lease-aware)
const LEASE_SEC = 120;        // reservation — roughly 3× the p99 of your business logic
const COMPLETED_SEC = 604800; // completion — longer than the sender's retry window (7 days)
 
async tryAcquire(key: string): Promise<"acquired" | "in_progress" | "completed"> {
  const set = await this.redis.set(
    key,
    JSON.stringify({ status: "in_progress", at: Date.now() }),
    { nx: true, ex: LEASE_SEC },   // a short lease, not 24 hours
  );
  if (set === "OK") return "acquired";
 
  const existing = await this.redis.get<{ status: string }>(key);
  if (existing?.status === "completed") return "completed";
  // A stranded in_progress key disappears after LEASE_SEC, so retries can take over
  return "in_progress";
}
 
async markCompleted(key: string, result: unknown): Promise<void> {
  await this.redis.set(
    key,
    JSON.stringify({ status: "completed", result, at: Date.now() }),
    { ex: COMPLETED_SEC },         // the long TTL is written only on success
  );
}
 
async markFailed(key: string, reason: string): Promise<void> {
  // On failure, drop the key so the next retry runs normally. Keep the reason in logs.
  await this.redis.del(key);
  console.warn("[idempotency] released after failure", { key, reason });
}

ALTER TABLE idempotency_records
  ADD COLUMN lease_expires_at TIMESTAMPTZ NOT NULL DEFAULT (now() + INTERVAL '2 minutes');

async tryAcquire(key: string, requestHash: string) {
  const inserted = await this.pool.query(
    `INSERT INTO idempotency_records (key, status, request_hash, lease_expires_at)
     VALUES ($1, 'in_progress', $2, now() + INTERVAL '2 minutes')
     ON CONFLICT (key) DO UPDATE
       SET status = 'in_progress',
           lease_expires_at = now() + INTERVAL '2 minutes',
           updated_at = now()
     WHERE idempotency_records.status = 'in_progress'
       AND idempotency_records.lease_expires_at < now()
     RETURNING key`,
    [key, requestHash],
  );
 
  // rowCount is 1 for a fresh reservation and for reclaiming an expired lease
  if (inserted.rowCount === 1) return { acquired: true } as const;
 
  const { rows } = await this.pool.query(
    `SELECT status, response_body, request_hash FROM idempotency_records WHERE key = $1`,
    [key],
  );
  const r = rows[0];
  // Reached when a cleanup job deleted the row between the INSERT and the SELECT. Retry once.
  if (!r) return this.tryAcquire(key, requestHash);
  if (r.request_hash !== requestHash) {
    throw new Error("Idempotency-Key reused with different request body");
  }
  return {
    acquired: false,
    status: r.status as "in_progress" | "completed" | "failed",
    response: r.response_body,
  } as const;
}

Create src/lib/idempotency/postgres-store.test.ts.
Test the PgDedupeStore class methods tryAcquire and markCompleted using
pg-mem (in-memory Postgres). Cover:
 
1. First call returns acquired: true
2. Second call with same key + same request_hash returns acquired: false, status: "in_progress"
3. Third call after completion returns status: "completed", response_body matches
4. Same key + different request_hash throws "Idempotency-Key reused..."
5. 10 concurrent calls with the same key result in exactly one acquired: true
 
Each test should have // arrange, // act, // assert comments.

$ npm test -- src/lib/idempotency/postgres-store.test.ts
 
 FAIL  src/lib/idempotency/postgres-store.test.ts
  PgDedupeStore
    × tryAcquire returns acquired on first call
    × tryAcquire returns in_progress on duplicate
    ...
  Tests: 5 failed, 5 total

Create src/lib/idempotency/postgres-store.ts.
Write the minimal implementation that passes
@/src/lib/idempotency/postgres-store.test.ts. No transactions —
use INSERT ... ON CONFLICT DO NOTHING. Add a one-line comment for
each line explaining "why".

// tests/integration/concurrent-acquire.test.ts
import { describe, expect, it } from "vitest";
import { Pool } from "pg";
import { PgDedupeStore } from "../../src/lib/idempotency/postgres-store";
 
describe("PgDedupeStore (integration)", () => {
  it("100 parallel callers acquire exactly once", async () => {
    const pool = new Pool({ connectionString: process.env.DATABASE_URL });
    const store = new PgDedupeStore(pool);
    const key = `test:${Date.now()}`;
    const hash = "abc";
 
    const results = await Promise.all(
      Array.from({ length: 100 }, () => store.tryAcquire(key, hash)),
    );
 
    const acquired = results.filter((r) => r.acquired).length;
    expect(acquired).toBe(1);
    await pool.end();
  });
});

// activities/charge-payment.ts
import { PgDedupeStore } from "../src/lib/idempotency/postgres-store";
 
export async function chargePayment(input: {
  idempotencyKey: string;       // workflow ID + activity name + retry count
  userId: string;
  amount: number;
}): Promise<{ chargeId: string }> {
  const store = new PgDedupeStore(getPool());
  const hash = sha256(`${input.userId}|${input.amount}`);
  const acq = await store.tryAcquire(input.idempotencyKey, hash);
  if (!acq.acquired && acq.status === "completed") {
    return acq.response as { chargeId: string };
  }
  const charge = await stripe.charges.create(
    { amount: input.amount, currency: "jpy", customer: input.userId },
    { idempotencyKey: input.idempotencyKey }, // Stripe gets the same key
  );
  await store.markCompleted(input.idempotencyKey, { chargeId: charge.id });
  return { chargeId: charge.id };
}

import { inngest } from "../inngest";
 
export const sendWelcomeEmail = inngest.createFunction(
  { id: "send-welcome-email", concurrency: { limit: 50 } },
  { event: "user/created" },
  async ({ event, step }) => {
    const dedupKey = `welcome-email:${event.data.userId}`;
 
    await step.run("dedupe-and-send", async () => {
      const store = getStore();
      const status = await store.tryAcquire(dedupKey);
      if (status === "completed") return { skipped: true };
 
      await sendgrid.send({
        to: event.data.email,
        from: "no-reply@example.com",
        templateId: "welcome",
      });
      await store.markCompleted(dedupKey, { sentAt: Date.now() });
      return { skipped: false };
    });
  },
);

import { metrics } from "@opentelemetry/api";
 
const counter = metrics.getMeter("dedupe").createCounter("dedupe_outcome_total");
 
export async function trackedTryAcquire(
  store: { tryAcquire: (k: string) => Promise<"acquired" | "in_progress" | "completed"> },
  key: string,
  source: "ext" | "cli" | "biz",
) {
  const outcome = await store.tryAcquire(key);
  counter.add(1, { outcome, source });
  return outcome;
}

const latency = metrics.getMeter("dedupe").createHistogram("dedupe_latency_ms");
 
export async function timed<T>(label: string, fn: () => Promise<T>): Promise<T> {
  const start = performance.now();
  try {
    return await fn();
  } finally {
    latency.record(performance.now() - start, { label });
  }
}
 
// usage
await timed("dedupe.acquire", () => store.tryAcquire(key));

import { trace } from "@opentelemetry/api";
 
const tracer = trace.getTracer("payments");
 
export async function chargeWithDedupe(key: string, amount: number) {
  return tracer.startActiveSpan("charge.dedupe", async (span) => {
    span.setAttribute("idempotency_key", key);
    span.setAttribute("amount", amount);
    try {
      // ... dedupe + charge
    } finally {
      span.end();
    }
  });
}

const acq = await store.tryAcquire(key, hash);
if (!acq.acquired) return acq.response;
try {
  const result = await doBusinessLogic();
  await store.markCompleted(key, result);
  return result;
} catch (err) {
  await store.markFailed(key, String(err)); // <- do not skip this
  throw err;
}