How to Implement Idempotency Keys in Go with PostgreSQL

// Command provider is a stand-in for a card processor such as Stripe or Adyen.
// It charges slowly, declines one test card, and honors an Idempotency-Key.
package main

import (
    "context"
    "encoding/json"
    "log"
    "net/http"
    "os"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

type chargeRequest struct {
    Amount int64  `json:"amount"`
    Card   string `json:"card"`
}

func main() {
    url := os.Getenv("PROVIDER_DATABASE_URL")
    if url == "" {
        url = "postgres://postgres:postgres@localhost:5432/provider?pool_max_conns=50" // the local Docker container; one connection per charge in flight
    }
    db, err := pgxpool.New(context.Background(), url)
    if err != nil {
        log.Fatal(err)
    }

    http.HandleFunc("POST /charges", func(w http.ResponseWriter, r *http.Request) {
        var req chargeRequest
        if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
            http.Error(w, "bad request", http.StatusBadRequest)
            return
        }
        ctx := r.Context()
        tx, err := db.Begin(ctx)
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
        defer tx.Rollback(ctx)

        // Like Stripe, answer 409 while another request with the same key is running,
        // and return the original charge once it has finished.
        var key *string
        if k := r.Header.Get("Idempotency-Key"); k != "" {
            key = &k
            var locked bool
            tx.QueryRow(ctx, `SELECT pg_try_advisory_xact_lock(hashtext($1))`, k).Scan(&locked)
            if !locked {
                writeJSON(w, http.StatusConflict, map[string]string{"error": "request_in_progress"})
                return
            }
            var id string
            if tx.QueryRow(ctx, `SELECT id FROM card_charges WHERE idempotency_key = $1`, k).Scan(&id) == nil {
                writeJSON(w, http.StatusCreated, map[string]string{"id": id})
                return
            }
        }

        time.Sleep(time.Second) // the card network is slow
        if req.Card == "tok_declined" {
            writeJSON(w, http.StatusPaymentRequired, map[string]string{"error": "card_declined"})
            return
        }

        // Without a key (NULL), nothing conflicts: every call is a new charge.
        var id string
        err = tx.QueryRow(ctx, `
            INSERT INTO card_charges (idempotency_key, card, amount)
            VALUES ($1, $2, $3)
            RETURNING id`, key, req.Card, req.Amount).Scan(&id)
        if err == nil {
            err = tx.Commit(ctx)
        }
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
        writeJSON(w, http.StatusCreated, map[string]string{"id": id})
    })

    log.Println("provider listening on :8090")
    log.Fatal(http.ListenAndServe(":8090", nil))
}

func writeJSON(w http.ResponseWriter, code int, v any) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(code)
    json.NewEncoder(w).Encode(v)
}

// Package payments is a small payments API that charges a card through a provider
// and records each payment in PostgreSQL.
package payments

import (
    "bytes"
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "net/http"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

// PaymentRequest is what the checkout page sends. Amount is in cents; Card is a
// token from the provider's card form, never a card number.
type PaymentRequest struct {
    Amount int64  `json:"amount"`
    Card   string `json:"card"`
}

// Payment is what the API returns for a successful charge.
type Payment struct {
    ID         int64  `json:"id"`
    Amount     int64  `json:"amount"`
    ProviderID string `json:"provider_charge_id"`
}

// Server holds the API's dependencies.
type Server struct {
    DB       *pgxpool.Pool
    Provider *Provider
}

// ErrDeclined means the provider refused the card. That is a final answer, not a fault.
var ErrDeclined = errors.New("card declined")

// ErrInProgress means the provider is still working on a charge with the same key.
var ErrInProgress = errors.New("charge in progress at the provider")

// Provider is a client for the card processor.
type Provider struct {
    URL  string
    HTTP *http.Client
}

// Charge asks the provider to take amount from card. A non-empty key is sent as
// the provider's Idempotency-Key, so repeating the call cannot charge twice.
func (p *Provider) Charge(ctx context.Context, amount int64, card, key string) (string, error) {
    body, _ := json.Marshal(PaymentRequest{Amount: amount, Card: card})
    req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.URL+"/charges", bytes.NewReader(body))
    if err != nil {
        return "", err
    }
    req.Header.Set("Content-Type", "application/json")
    if key != "" {
        req.Header.Set("Idempotency-Key", key)
    }
    resp, err := p.HTTP.Do(req)
    if err != nil {
        return "", err
    }
    defer resp.Body.Close()

    switch resp.StatusCode {
    case http.StatusCreated:
        var out struct{ ID string }
        err := json.NewDecoder(resp.Body).Decode(&out)
        return out.ID, err
    case http.StatusPaymentRequired:
        return "", ErrDeclined
    case http.StatusConflict:
        return "", ErrInProgress
    default:
        return "", fmt.Errorf("provider returned %s", resp.Status)
    }
}

// NewProvider returns a client whose timeout bounds every provider call.
func NewProvider(url string) *Provider {
    return &Provider{URL: url, HTTP: &http.Client{Timeout: 5 * time.Second}}
}

// customerID stands in for authentication: a real API takes it from the session or token.
func customerID(r *http.Request) string {
    return r.Header.Get("Customer-ID")
}

func writeJSON(w http.ResponseWriter, code int, v any) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(code)
    json.NewEncoder(w).Encode(v)
}

func writeError(w http.ResponseWriter, code int, msg string) {
    writeJSON(w, code, map[string]string{"error": msg})
}

func decode(r *http.Request) (PaymentRequest, error) {
    var req PaymentRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        return req, err
    }
    if req.Amount <= 0 || req.Card == "" {
        return req, errors.New("amount and card are required")
    }
    return req, nil
}

// Command pay is the checkout client: it sends one payment and retries the way a
// mobile app does when the network is slow.
package main

import (
    "bytes"
    "crypto/rand"
    "flag"
    "fmt"
    "io"
    "net/http"
    "strings"
    "sync"
    "time"
)

func main() {
    amount := flag.Int("amount", 2500, "amount in cents")
    card := flag.String("card", "tok_visa", "card token")
    keyMode := flag.String("key", "once", "once: one key per payment; new: a new key per attempt; none: no key")
    keyValue := flag.String("key-value", "", "use this key instead of a random one")
    firstTimeout := flag.Duration("first-timeout", 600*time.Millisecond, "the first attempt gives up after this long (the phone loses signal)")
    timeout := flag.Duration("timeout", 3*time.Second, "how long later attempts wait for an answer")
    attempts := flag.Int("attempts", 4, "attempts before giving up")
    parallel := flag.Int("parallel", 1, "send this many copies at the same moment")
    flag.Parse()

    key := *keyValue
    if key == "" {
        key = rand.Text() // 26 random characters from crypto/rand
    }
    body := fmt.Sprintf(`{"amount": %d, "card": %q}`, *amount, *card)
    send := func(attempt int) (int, string, error) {
        req, _ := http.NewRequest(http.MethodPost, "http://localhost:8080/payments", strings.NewReader(body))
        req.Header.Set("Content-Type", "application/json")
        req.Header.Set("Customer-ID", "cus_1001")
        switch *keyMode {
        case "once":
            req.Header.Set("Idempotency-Key", key)
        case "new":
            req.Header.Set("Idempotency-Key", rand.Text())
        }
        wait := *timeout
        if attempt == 1 {
            wait = *firstTimeout
        }
        resp, err := (&http.Client{Timeout: wait}).Do(req)
        if err != nil {
            return 0, "", err
        }
        defer resp.Body.Close()
        b, _ := io.ReadAll(resp.Body)
        extra := ""
        if resp.Header.Get("Idempotent-Replayed") == "true" {
            extra = " (replayed)"
        }
        return resp.StatusCode, string(bytes.TrimSpace(b)) + extra, nil
    }

    // One payment, retried with exponential backoff until it gets a final answer.
    pay := func(id int) {
        backoff := 300 * time.Millisecond
        for attempt := 1; attempt <= *attempts; attempt++ {
            code, msg, err := send(attempt)
            prefix := fmt.Sprintf("attempt %d:", attempt)
            if *parallel > 1 {
                prefix = fmt.Sprintf("copy %d %s", id, prefix)
            }
            switch {
            case err != nil:
                fmt.Println(prefix, "no answer:", shorten(err))
            case code == http.StatusConflict || code >= 500:
                fmt.Println(prefix, code, msg)
            default:
                fmt.Println(prefix, code, msg)
                return
            }
            time.Sleep(backoff)
            backoff *= 2
        }
    }

    var wg sync.WaitGroup
    for i := 1; i <= *parallel; i++ {
        wg.Add(1)
        go func() { defer wg.Done(); pay(i) }()
    }
    wg.Wait()
}

func shorten(err error) string {
    if strings.Contains(err.Error(), "Client.Timeout") {
        return "timed out"
    }
    return err.Error()
}

package payments

import (
    "context"
    "errors"
    "net/http"
)

// CreatePaymentNaive charges the card and records the payment. Every call charges again.
func (s *Server) CreatePaymentNaive(w http.ResponseWriter, r *http.Request) {
    req, err := decode(r)
    if err != nil {
        writeError(w, http.StatusBadRequest, err.Error())
        return
    }
    // Finish the payment even if the caller hangs up: a charge must not be half done.
    ctx := context.WithoutCancel(r.Context())

    providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, "")
    if errors.Is(err, ErrDeclined) {
        writeError(w, http.StatusPaymentRequired, "card declined")
        return
    }
    if err != nil {
        writeError(w, http.StatusBadGateway, "payment provider unavailable")
        return
    }

    var p Payment
    err = s.DB.QueryRow(ctx, `
        INSERT INTO charges (customer_id, amount, provider_id)
        VALUES ($1, $2, $3)
        RETURNING id, amount, provider_id`,
        customerID(r), req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID)
    if err != nil {
        writeError(w, http.StatusInternalServerError, "could not record payment")
        return
    }
    writeJSON(w, http.StatusCreated, p)
}

// Command api serves POST /payments in one of three versions, chosen with -mode.
package main

import (
    "context"
    "flag"
    "log"
    "net/http"
    "os"
    "time"

    "example.com/payments/payments"
    "github.com/jackc/pgx/v5/pgxpool"
)

func main() {
    mode := flag.String("mode", "idempotent", "naive, checkfirst or idempotent")
    flag.Parse()

    db, err := pgxpool.New(context.Background(), env("DATABASE_URL", "postgres://postgres:postgres@localhost:5432/shop"))
    if err != nil {
        log.Fatal(err)
    }
    s := &payments.Server{DB: db, Provider: payments.NewProvider("http://localhost:8090")}

    var h http.HandlerFunc
    switch *mode {
    case "naive":
        h = s.CreatePaymentNaive
    default:
        log.Fatalf("unknown -mode %q", *mode)
    }
    http.HandleFunc("POST /payments", h)

    srv := &http.Server{Addr: ":8080", ReadHeaderTimeout: 5 * time.Second}
    log.Printf("api (%s) listening on :8080", *mode)
    log.Fatal(srv.ListenAndServe())
}

// env returns the variable's value, or a default that suits the local Docker container.
func env(name, fallback string) string {
    if v := os.Getenv(name); v != "" {
        return v
    }
    return fallback
}

package payments

import (
    "context"
    "encoding/json"
    "errors"
    "net/http"

    "github.com/jackc/pgx/v5"
)

// CreatePaymentCheckFirst looks the key up, and only charges if it has never seen it.
// The lookup and the save are separate steps, so two requests can both miss.
func (s *Server) CreatePaymentCheckFirst(w http.ResponseWriter, r *http.Request) {
    key := r.Header.Get("Idempotency-Key")
    if key == "" {
        writeError(w, http.StatusBadRequest, "Idempotency-Key header is required")
        return
    }
    req, err := decode(r)
    if err != nil {
        writeError(w, http.StatusBadRequest, err.Error())
        return
    }
    ctx := context.WithoutCancel(r.Context())

    // 1. Have we answered this key before? If so, send the same answer again.
    var code int
    var body []byte
    err = s.DB.QueryRow(ctx, `
        SELECT response_code, response_body FROM idempotency_keys
        WHERE customer_id = $1 AND key = $2`, customerID(r), key).Scan(&code, &body)
    if err == nil {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(code)
        w.Write(body)
        return
    }
    if !errors.Is(err, pgx.ErrNoRows) {
        writeError(w, http.StatusInternalServerError, "could not read idempotency key")
        return
    }

    // 2. New key: charge and record the payment.
    providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, "")
    if err != nil {
        writeError(w, http.StatusBadGateway, "payment provider unavailable")
        return
    }
    var p Payment
    err = s.DB.QueryRow(ctx, `
        INSERT INTO charges (customer_id, amount, provider_id)
        VALUES ($1, $2, $3)
        RETURNING id, amount, provider_id`,
        customerID(r), req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID)
    if err != nil {
        writeError(w, http.StatusInternalServerError, "could not record payment")
        return
    }

    // 3. Remember the answer for next time.
    body, _ = json.Marshal(p)
    _, err = s.DB.Exec(ctx, `
        INSERT INTO idempotency_keys (customer_id, key, request_hash, status, response_code, response_body)
        VALUES ($1, $2, '', 'completed', 201, $3)
        ON CONFLICT DO NOTHING`, customerID(r), key, body)
    if err != nil {
        writeError(w, http.StatusInternalServerError, "could not save idempotency key")
        return
    }
    writeJSON(w, http.StatusCreated, p)
}

    case "checkfirst":
        h = s.CreatePaymentCheckFirst

// Command api serves POST /payments in one of three versions, chosen with -mode.
package main

import (
    "context"
    "flag"
    "log"
    "net/http"
    "os"
    "time"

    "example.com/payments/payments"
    "github.com/jackc/pgx/v5/pgxpool"
)

func main() {
    mode := flag.String("mode", "idempotent", "naive, checkfirst or idempotent")
    lease := flag.Duration("lease", 0, "take over a started key after this long (0 = never)")
    passKey := flag.Bool("pass-key", false, "send the idempotency key to the provider")
    crash := flag.Bool("crash-after-charge", false, "exit after the provider charges")
    flag.Parse()

    db, err := pgxpool.New(context.Background(), env("DATABASE_URL", "postgres://postgres:postgres@localhost:5432/shop"))
    if err != nil {
        log.Fatal(err)
    }
    s := &payments.Server{DB: db, Provider: payments.NewProvider("http://localhost:8090")}

    var h http.HandlerFunc
    switch *mode {
    case "naive":
        h = s.CreatePaymentNaive
    case "checkfirst":
        h = s.CreatePaymentCheckFirst
    case "idempotent":
        h = s.CreatePayment(payments.Options{Lease: *lease, PassKeyToCharge: *passKey, CrashAfterCharge: *crash})
    default:
        log.Fatalf("unknown -mode %q", *mode)
    }
    http.HandleFunc("POST /payments", h)

    srv := &http.Server{Addr: ":8080", ReadHeaderTimeout: 5 * time.Second}
    log.Printf("api (%s) listening on :8080", *mode)
    log.Fatal(srv.ListenAndServe())
}

// env returns the variable's value, or a default that suits the local Docker container.
func env(name, fallback string) string {
    if v := os.Getenv(name); v != "" {
        return v
    }
    return fallback
}

// claim tries to make this request the owner of the key. It returns true, and the
// claim's locked_at, if this request should do the work. Otherwise it returns the
// key's current record.
func (s *Server) claim(ctx context.Context, customer, key string, hash []byte, opt Options) (bool, keyRecord, error) {
    var rec keyRecord
    var err error
    if opt.Lease == 0 {
        err = s.DB.QueryRow(ctx, `
            INSERT INTO idempotency_keys (customer_id, key, request_hash)
            VALUES ($1, $2, $3)
            ON CONFLICT (customer_id, key) DO NOTHING
            RETURNING locked_at`, customer, key, hash).Scan(&rec.lockedAt)
    } else {
        // Also take over a key whose owner started long ago and never finished.
        err = s.DB.QueryRow(ctx, `
            INSERT INTO idempotency_keys (customer_id, key, request_hash)
            VALUES ($1, $2, $3)
            ON CONFLICT (customer_id, key) DO UPDATE SET locked_at = now()
            WHERE idempotency_keys.status = 'started'
              AND idempotency_keys.request_hash = EXCLUDED.request_hash
              AND idempotency_keys.locked_at < now() - make_interval(secs => $4)
            RETURNING locked_at`, customer, key, hash, opt.Lease.Seconds()).Scan(&rec.lockedAt)
    }
    if err == nil {
        return true, rec, nil
    }
    if !errors.Is(err, pgx.ErrNoRows) {
        return false, keyRecord{}, err
    }

    // Someone else holds the key. Read what they have so far.
    err = s.DB.QueryRow(ctx, `
        SELECT status, request_hash FROM idempotency_keys
        WHERE customer_id = $1 AND key = $2`,
        customer, key).Scan(&rec.status, &rec.hash)
    if errors.Is(err, pgx.ErrNoRows) {
        // Released a moment ago: report it as in progress so the client retries.
        rec.status, rec.hash = "started", hash
        return false, rec, nil
    }
    return false, rec, err
}

// CreatePayment charges at most once per Idempotency-Key and replays the first answer.
func (s *Server) CreatePayment(opt Options) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        customer := customerID(r)
        key := idempotencyKey(r)
        if key == "" || len(key) > 255 {
            writeError(w, http.StatusBadRequest, "Idempotency-Key header is required (at most 255 characters)")
            return
        }
        req, err := decode(r)
        if err != nil {
            writeError(w, http.StatusBadRequest, err.Error())
            return
        }
        hash := fingerprint(r, req)
        ctx := context.WithoutCancel(r.Context())

        claimed, rec, err := s.claim(ctx, customer, key, hash, opt)
        switch {
        case err != nil:
            writeError(w, http.StatusInternalServerError, "could not claim idempotency key")
            return
        case claimed:
            // This request owns the key; carry on below.
        case !bytes.Equal(rec.hash, hash):
            writeError(w, http.StatusUnprocessableEntity, "Idempotency-Key was already used with a different request")
            return
        case rec.status == "completed":
            s.replay(ctx, w, customer, key)
            return
        default:
            w.Header().Set("Retry-After", "1")
            writeError(w, http.StatusConflict, "a request with this Idempotency-Key is still in progress")
            return
        }

        var p Payment
        record := func(tx pgx.Tx) ([]byte, error) {
            err := tx.QueryRow(ctx, `
                INSERT INTO charges (customer_id, amount, provider_id)
                VALUES ($1, $2, $3)
                RETURNING id, amount, provider_id`,
                customer, req.Amount, providerID).Scan(&p.ID, &p.Amount, &p.ProviderID)
            if err != nil {
                return nil, err
            }
            return json.Marshal(p)
        }
        err = s.finish(ctx, customer, key, http.StatusCreated, nil, record)
        if errors.Is(err, errLostKey) {
            s.replay(ctx, w, customer, key) // a takeover finished first; answer as it did
            return
        }
        if err != nil {
            writeError(w, http.StatusInternalServerError, "could not record payment")
            return
        }
        writeJSON(w, http.StatusCreated, p)

// errLostKey means another request finished this key first.
var errLostKey = errors.New("idempotency key already finished")

// finish saves the answer for a key, and the payment row when there is one, in one
// transaction. Only a request that still sees the key as started may finish it.
func (s *Server) finish(ctx context.Context, customer, key string, code int, body []byte, record func(pgx.Tx) ([]byte, error)) error {
    tx, err := s.DB.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    var status string
    err = tx.QueryRow(ctx, `
        SELECT status FROM idempotency_keys
        WHERE customer_id = $1 AND key = $2
        FOR UPDATE`, customer, key).Scan(&status)
    if err != nil {
        return err
    }
    if status != "started" {
        return errLostKey
    }
    if record != nil {
        if body, err = record(tx); err != nil {
            return err
        }
    }
    _, err = tx.Exec(ctx, `
        UPDATE idempotency_keys
        SET status = 'completed', response_code = $3, response_body = $4
        WHERE customer_id = $1 AND key = $2`, customer, key, code, body)
    if err != nil {
        return err
    }
    return tx.Commit(ctx)
}

// replay sends the stored answer for a finished key.
func (s *Server) replay(ctx context.Context, w http.ResponseWriter, customer, key string) {
    var code int
    var body []byte
    err := s.DB.QueryRow(ctx, `
        SELECT response_code, response_body FROM idempotency_keys
        WHERE customer_id = $1 AND key = $2`, customer, key).Scan(&code, &body)
    if err != nil {
        writeError(w, http.StatusInternalServerError, "could not read stored result")
        return
    }
    w.Header().Set("Content-Type", "application/json")
    w.Header().Set("Idempotent-Replayed", "true")
    w.WriteHeader(code)
    w.Write(body)
}

// fingerprint identifies the request a key was first used with. Hashing the decoded
// struct, not the raw bytes, ignores whitespace and field order.
func fingerprint(r *http.Request, req PaymentRequest) []byte {
    canonical, _ := json.Marshal(req)
    sum := sha256.Sum256(append([]byte(r.Method+" "+r.URL.Path+"\n"), canonical...))
    return sum[:]
}

// idempotencyKey reads the header. The IETF draft sends it as a quoted string,
// most providers send it bare; accept both.
func idempotencyKey(r *http.Request) string {
    k := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
    if len(k) >= 2 && k[0] == '"' && k[len(k)-1] == '"' {
        k = k[1 : len(k)-1]
    }
    return k
}

        providerID, err := s.Provider.Charge(ctx, req.Amount, req.Card, chargeKey)

        if errors.Is(err, ErrDeclined) {
            // A decline is a final answer: store it so a retry gets the same one.
            body, _ := json.Marshal(map[string]string{"error": "card declined"})
            err := s.finish(ctx, customer, key, http.StatusPaymentRequired, body, nil)
            if errors.Is(err, errLostKey) {
                s.replay(ctx, w, customer, key)
                return
            }
            if err != nil {
                writeError(w, http.StatusInternalServerError, "could not save result")
                return
            }
            w.Header().Set("Content-Type", "application/json")
            w.WriteHeader(http.StatusPaymentRequired)
            w.Write(body)
            return
        }
        if errors.Is(err, ErrInProgress) {
            // The provider is still charging under this key (an earlier claim we took
            // over is mid-call). Not a fault: keep the key and let the client retry.
            w.Header().Set("Retry-After", "1")
            writeError(w, http.StatusConflict, "a request with this Idempotency-Key is still in progress")
            return
        }
        if err != nil {
            // A fault is not an answer. Release the key so a retry can run the charge
            // again, but only if it is still our claim and not a later takeover's.
            s.DB.Exec(ctx, `
                DELETE FROM idempotency_keys
                WHERE customer_id = $1 AND key = $2
                  AND status = 'started' AND locked_at = $3`, customer, key, rec.lockedAt)
            writeError(w, http.StatusBadGateway, "payment provider unavailable")
            return
        }