Antigravity × TypeScript で作る冪等性キーと重複排除ストア本番設計ガイド

// 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) {}
 
  /**
   * キーを「予約」できれば true（=新規処理）、既存なら false（=重複）。
   * KV は Compare-And-Swap がないので、put + get で競合を緩和する。
   * 厳密な exactly-once が必要なら Postgres / Durable Objects を使う。
   */
  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") {
      // ストレージは DO 内で常に強整合
      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); // 同じキー → 同じインスタンス
  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) {}
 
  /**
   * SET NX EX による Atomic な予約。
   * 戻り値:
   *   "acquired"     — 新規予約成功（→ 本処理を実行する）
   *   "in_progress"  — 別ワーカーが処理中（→ ポーリングか即 200 で返す）
   *   "completed"    — 既に完了（→ 保存済み結果を返す）
   */
  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) {
    // INSERT が成功した行は「自分が予約した」のみ。他は ON CONFLICT で何もしない。
    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) {
      // 同じキーで違うリクエスト = クライアントの設計ミス
      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],
    );
  }
}

src/lib/idempotency/postgres-store.test.ts を作成してください。
PgDedupeStore クラスの tryAcquire / markCompleted を pg-mem (in-memory Postgres) で
テストします。以下のケースを網羅:
 
1. 初回呼び出しは acquired: true
2. 同一キー + 同一 request_hash の二回目は acquired: false, status: "in_progress"
3. 完了後の三回目は status: "completed", response_body が一致
4. 同一キー + 違う request_hash は throws "Idempotency-Key reused..."
5. 並列で同じキーを 10 回叩いたとき、acquired: true は厳密に1回のみ
 
各テストには // arrange, // act, // assert コメントを付けてください。

$ 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

src/lib/idempotency/postgres-store.ts を作成してください。
@/src/lib/idempotency/postgres-store.test.ts のテストを全てパスする
最小限の実装を書いてください。トランザクションは使わず、
INSERT ... ON CONFLICT DO NOTHING で実装。
コメントで「なぜそうするか」を1文ずつ添える。

// 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リクエストで acquired は厳密に1回", 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;       // ワークフローID + アクティビティ名 + リトライ回数 で合成
  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 }; // 完了済み結果を即返す
  }
  // 実際の Stripe 呼び出し（Stripe 側にも Idempotency-Key を渡す）
  const charge = await stripe.charges.create(
    { amount: input.amount, currency: "jpy", customer: input.userId },
    { idempotencyKey: input.idempotencyKey },
  );
  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(); // RedisDedupeStore など
      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 });
  }
}
 
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)); // ← これを忘れない
  throw err;
}