Design a Payment System — Idempotency, Reconciliation, and the Boring Discipline

[Customer] ──▶ [Your API] ──▶ [PSP (Stripe)]
                  │
                  ▼
              [Ledger DB]
                  │
                  ▼
              [Reconciliation job]
                  │
                  ▼
              [Reports / accounting]

async def charge(user_id, amount, idempotency_key):
    # Local check
    if existing := await db.fetchrow(
        "SELECT id, status FROM payments WHERE idempotency_key = $1",
        idempotency_key
    ):
        return existing
    
    # PSP also dedupes via key
    pi = await stripe.PaymentIntent.create(
        amount=amount,
        currency="usd",
        customer=user_id,
        idempotency_key=idempotency_key,
    )
    
    await db.execute(
        "INSERT INTO payments (id, idempotency_key, user_id, amount, status, psp_id) "
        "VALUES (gen_random_uuid(), $1, $2, $3, 'pending', $4)",
        idempotency_key, user_id, amount, pi.id
    )
    return pi

CREATE TABLE journal_entries (
    id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    payment_id  uuid NOT NULL,
    occurred_at timestamptz NOT NULL DEFAULT now(),
    description text NOT NULL
);

CREATE TABLE journal_lines (
    id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    entry_id      uuid NOT NULL REFERENCES journal_entries(id),
    account       text NOT NULL,        -- e.g., 'cash', 'revenue', 'fees'
    direction     char(1) NOT NULL,     -- 'D' or 'C'
    amount_cents  bigint NOT NULL CHECK (amount_cents > 0)
);

-- Constraint: every entry's debits = credits
CREATE FUNCTION check_balanced() RETURNS trigger AS $$
DECLARE d bigint; c bigint;
BEGIN
    SELECT COALESCE(SUM(amount_cents), 0) INTO d FROM journal_lines
        WHERE entry_id = NEW.entry_id AND direction = 'D';
    SELECT COALESCE(SUM(amount_cents), 0) INTO c FROM journal_lines
        WHERE entry_id = NEW.entry_id AND direction = 'C';
    IF d <> c THEN RAISE EXCEPTION 'unbalanced'; END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

DR cash    97
DR fees     3
CR revenue 100

@app.post("/webhooks/stripe")
async def stripe_webhook(req: Request):
    event = stripe.Webhook.construct_event(
        await req.body(), req.headers["stripe-signature"], WEBHOOK_SECRET
    )
    
    # Idempotent — store event id
    if await db.exists("SELECT 1 FROM webhook_events WHERE id = $1", event.id):
        return {"status": "duplicate"}
    
    await db.execute("INSERT INTO webhook_events (id, type) VALUES ($1, $2)",
                     event.id, event.type)
    
    if event.type == "payment_intent.succeeded":
        await mark_payment_succeeded(event.data.object)
    elif event.type == "charge.refunded":
        await record_refund(event.data.object)
    # ... handle others
    
    return {"status": "ok"}

async def reconcile(date):
    # Fetch from Stripe
    psp_charges = await stripe.Charge.list(created={"gte": date, "lt": date+1})
    
    # Fetch from local
    local = await db.fetch(
        "SELECT psp_id, amount FROM payments WHERE created_at::date = $1",
        date
    )
    
    psp_set = {c.id: c.amount for c in psp_charges}
    local_set = {p["psp_id"]: p["amount"] for p in local}
    
    only_psp = set(psp_set) - set(local_set)
    only_local = set(local_set) - set(psp_set)
    mismatched = {k for k in psp_set & local_set if psp_set[k] != local_set[k]}
    
    if only_psp or only_local or mismatched:
        await alert_finance(only_psp, only_local, mismatched)

async def refund(payment_id, amount, idempotency_key):
    payment = await get_payment(payment_id)
    if not payment.refundable: raise InvalidRefund
    
    refund = await stripe.Refund.create(
        payment_intent=payment.psp_id,
        amount=amount,
        idempotency_key=idempotency_key,
    )
    
    # Ledger entry: reverse the original
    await ledger.entry(
        payment_id=payment_id,
        description="refund",
        lines=[
            ("revenue", "D", amount),
            ("cash",    "C", amount),
        ],
    )

@webhook("charge.dispute.created")
async def on_dispute(event):
    await db.execute(
        "INSERT INTO disputes (id, payment_id, amount, reason) VALUES ($1, $2, $3, $4)",
        event.id, payment_id, amount, reason
    )
    # Hold revenue until resolved
    await ledger.entry(...)  # accrue dispute liability
    await notify_team()

CREATE TABLE payment_audit (
    id         bigserial PRIMARY KEY,
    payment_id uuid,
    actor      text,
    action     text,
    before     jsonb,
    after      jsonb,
    occurred_at timestamptz DEFAULT now()
);