API Idempotency Keys

idempotency_records
  account_id          TEXT NOT NULL
  key                 TEXT NOT NULL          -- client Idempotency-Key
  request_fingerprint TEXT NOT NULL          -- hash of canonical request
  state               ENUM('in_progress','completed','failed')
  response_status     INT NULL
  response_body       BYTEA / JSONB NULL     -- cached reply
  response_headers    JSONB NULL
  recovery_point      TEXT NULL              -- for multi-step flows
  lease_owner         TEXT NULL              -- worker id
  lease_expires_at    TIMESTAMPTZ NULL
  fencing_token       BIGINT NOT NULL DEFAULT 0
  created_at          TIMESTAMPTZ NOT NULL
  updated_at          TIMESTAMPTZ NOT NULL
  expires_at          TIMESTAMPTZ NOT NULL   -- ~created_at + 24h

PRIMARY KEY / UNIQUE (account_id, key)
INDEX (expires_at)                 -- reaper
INDEX (state, lease_expires_at)    -- zombie recovery

import hashlib, json, uuid
from datetime import datetime, timedelta, timezone
from enum import Enum

import asyncpg
from fastapi import FastAPI, Header, HTTPException, Request, Response

TTL = timedelta(hours=24)
LEASE = timedelta(seconds=30)

class State(str, Enum):
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

def fingerprint(method: str, path: str, body: dict) -> str:
    canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
    raw = f"{method.upper()}|{path}|{canonical}".encode()
    return hashlib.sha256(raw).hexdigest()

async def claim_or_replay(conn, account_id, key, fp, owner):
    now = datetime.now(timezone.utc)
    expires = now + TTL
    lease_exp = now + LEASE
    row = await conn.fetchrow(
        """
        INSERT INTO idempotency_records
          (account_id, key, request_fingerprint, state, lease_owner,
           lease_expires_at, fencing_token, created_at, updated_at, expires_at)
        VALUES ($1,$2,$3,'in_progress',$4,$5,1,$6,$6,$7)
        ON CONFLICT (account_id, key) DO NOTHING
        RETURNING *
        """,
        account_id, key, fp, owner, lease_exp, now, expires,
    )
    if row:
        return None  # we own the claim
    existing = await conn.fetchrow(
        "SELECT * FROM idempotency_records WHERE account_id=$1 AND key=$2 FOR UPDATE",
        account_id, key,
    )
    if existing["request_fingerprint"] != fp:
        raise HTTPException(422, "Idempotency key reused with different request")
    if existing["state"] in (State.COMPLETED, State.FAILED):
        return {
            "status": existing["response_status"],
            "body": existing["response_body"],
            "replayed": True,
        }
    if existing["lease_expires_at"] and existing["lease_expires_at"] > now:
        raise HTTPException(409, "Request in progress; retry later")
    await conn.execute(
        """
        UPDATE idempotency_records
           SET lease_owner=$3, lease_expires_at=$4,
               fencing_token = fencing_token + 1, updated_at=$5,
               recovery_point = COALESCE(recovery_point, 'start')
         WHERE account_id=$1 AND key=$2
        """,
        account_id, key, owner, lease_exp, now,
    )
    return None  # takeover; re-enter execution

async def complete(conn, account_id, key, status, body, ok):
    await conn.execute(
        """
        UPDATE idempotency_records
           SET state=$3, response_status=$4, response_body=$5::jsonb,
               updated_at=NOW(), lease_expires_at=NULL
         WHERE account_id=$1 AND key=$2
        """,
        account_id, key,
        State.COMPLETED if ok else State.FAILED,
        status, json.dumps(body),
    )

app = FastAPI()

@app.post("/v1/charges")
async def create_charge(
    request: Request,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
    account_id: str = Header(..., alias="X-Account-Id"),
):
    body = await request.json()
    fp = fingerprint("POST", "/v1/charges", body)
    owner = str(uuid.uuid4())
    pool = request.app.state.pool
    async with pool.acquire() as conn:
        async with conn.transaction():
            replay = await claim_or_replay(conn, account_id, idempotency_key, fp, owner)
            if replay:
                return Response(
                    content=json.dumps(replay["body"]),
                    status_code=replay["status"],
                    media_type="application/json",
                    headers={"Idempotent-Replayed": "true"},
                )
            charge_id = str(uuid.uuid4())
            amount = body["amount"]
            await conn.execute(
                "INSERT INTO charges(id, account_id, amount) VALUES ($1,$2,$3)",
                charge_id, account_id, amount,
            )
            await conn.execute(
                "UPDATE idempotency_records SET recovery_point='charge_inserted' WHERE account_id=$1 AND key=$2",
                account_id, idempotency_key,
            )
            result = {"id": charge_id, "amount": amount, "status": "succeeded"}
            await complete(conn, account_id, idempotency_key, 200, result, True)
            return result

import crypto from "crypto";
import express from "express";
import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const TTL_MS = 24 * 60 * 60 * 1000;
const LEASE_MS = 30_000;

type Cached = { status: number; body: unknown; replayed: true };

function fingerprint(method: string, path: string, body: unknown): string {
  const canonical = JSON.stringify(body, Object.keys(body as object).sort());
  return crypto
    .createHash("sha256")
    .update(`${method.toUpperCase()}|${path}|${canonical}`)
    .digest("hex");
}

async function claimOrReplay(
  accountId: string,
  key: string,
  fp: string,
  owner: string
): Promise<Cached | null> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const now = new Date();
    const expires = new Date(now.getTime() + TTL_MS);
    const leaseExp = new Date(now.getTime() + LEASE_MS);
    const inserted = await client.query(
      `INSERT INTO idempotency_records
         (account_id, key, request_fingerprint, state, lease_owner,
          lease_expires_at, fencing_token, created_at, updated_at, expires_at)
       VALUES ($1,$2,$3,'in_progress',$4,$5,1,$6,$6,$7)
       ON CONFLICT (account_id, key) DO NOTHING
       RETURNING id`,
      [accountId, key, fp, owner, leaseExp, now, expires]
    );
    if (inserted.rowCount === 1) {
      await client.query("COMMIT");
      return null;
    }
    const { rows } = await client.query(
      `SELECT * FROM idempotency_records
        WHERE account_id=$1 AND key=$2 FOR UPDATE`,
      [accountId, key]
    );
    const row = rows[0];
    if (row.request_fingerprint !== fp) {
      await client.query("ROLLBACK");
      const err: { status?: number } & Error = new Error("fingerprint mismatch");
      err.status = 422;
      throw err;
    }
    if (row.state === "completed" || row.state === "failed") {
      await client.query("COMMIT");
      return { status: row.response_status, body: row.response_body, replayed: true };
    }
    if (row.lease_expires_at && new Date(row.lease_expires_at) > now) {
      await client.query("ROLLBACK");
      const err: { status?: number } & Error = new Error("in progress");
      err.status = 409;
      throw err;
    }
    await client.query(
      `UPDATE idempotency_records
          SET lease_owner=$3, lease_expires_at=$4,
              fencing_token = fencing_token + 1, updated_at=$5
        WHERE account_id=$1 AND key=$2`,
      [accountId, key, owner, leaseExp, now]
    );
    await client.query("COMMIT");
    return null;
  } catch (e) {
    await client.query("ROLLBACK");
    throw e;
  } finally {
    client.release();
  }
}

const app = express();
app.use(express.json());
app.post("/v1/charges", async (req, res) => {
  const key = req.header("Idempotency-Key");
  const accountId = req.header("X-Account-Id");
  if (!key || !accountId) return res.status(400).json({ error: "missing headers" });
  const fp = fingerprint("POST", "/v1/charges", req.body);
  const owner = crypto.randomUUID();
  try {
    const replay = await claimOrReplay(accountId, key, fp, owner);
    if (replay) {
      res.setHeader("Idempotent-Replayed", "true");
      return res.status(replay.status).json(replay.body);
    }
    const chargeId = crypto.randomUUID();
    await pool.query(
      `INSERT INTO charges(id, account_id, amount) VALUES ($1,$2,$3)`,
      [chargeId, accountId, req.body.amount]
    );
    const body = { id: chargeId, amount: req.body.amount, status: "succeeded" };
    await pool.query(
      `UPDATE idempotency_records
          SET state='completed', response_status=200,
              response_body=$3::jsonb, lease_expires_at=NULL, updated_at=NOW(),
              recovery_point='charge_inserted'
        WHERE account_id=$1 AND key=$2`,
      [accountId, key, JSON.stringify(body)]
    );
    return res.status(200).json(body);
  } catch (e: any) {
    const status = e.status ?? 500;
    if (status === 409) res.setHeader("Retry-After", "1");
    return res.status(status).json({ error: e.message });
  }
});
app.listen(3000);