Idempotency explained — Part 1: basics, idempotency key and Go implementation

POST /payments
Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000
Content-Type: application/json

{"amount": 100, "currency": "EUR", "customer_id": "cus_8Rn2xM"}

package middleware

import (
    "bytes"
    "net/http"
    "sync"
    "time"
)

type CachedResult struct {
    StatusCode int
    Body       []byte
    CreatedAt  time.Time
}

// IdempotencyStore — in-memory map, demo only.
// For production, see the next section: PostgreSQL store.
type IdempotencyStore struct {
    mu      sync.RWMutex
    results map[string]CachedResult
}

func NewIdempotencyStore() *IdempotencyStore {
    return &IdempotencyStore{results: make(map[string]CachedResult)}
}

func (s *IdempotencyStore) Get(key string) (CachedResult, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    result, ok := s.results[key]
    return result, ok
}

func (s *IdempotencyStore) Set(key string, result CachedResult) {
    s.mu.Lock()
    defer s.mu.Unlock()
    s.results[key] = result
}

// responseRecorder captures the response for storage.
type responseRecorder struct {
    http.ResponseWriter
    code int
    body bytes.Buffer
}

func (r *responseRecorder) WriteHeader(code int) {
    r.code = code
    r.ResponseWriter.WriteHeader(code)
}

func (r *responseRecorder) Write(b []byte) (int, error) {
    r.body.Write(b)
    return r.ResponseWriter.Write(b)
}

func IdempotencyMiddleware(store *IdempotencyStore) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if r.Method != http.MethodPost {
                next.ServeHTTP(w, r)
                return
            }

            key := r.Header.Get("Idempotency-Key")
            if key == "" {
                http.Error(w, `{"error": "Idempotency-Key header required"}`, http.StatusBadRequest)
                return
            }

            // Already cached? Return immediately.
            if cached, ok := store.Get(key); ok {
                w.Header().Set("X-Idempotent-Replayed", "true")
                w.WriteHeader(cached.StatusCode)
                w.Write(cached.Body)
                return
            }

            // First pass: capture and store the response.
            rec := &responseRecorder{ResponseWriter: w, code: http.StatusOK}
            next.ServeHTTP(rec, r)

            store.Set(key, CachedResult{
                StatusCode: rec.code,
                Body:       rec.body.Bytes(),
                CreatedAt:  time.Now(),
            })
        })
    }
}

CREATE TABLE idempotency_keys (
    key         UUID PRIMARY KEY,
    status      VARCHAR(16) NOT NULL DEFAULT 'pending', -- pending | completed
    status_code INT,
    body        TEXT,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Automatic cleanup of old keys (Stripe keeps 24h)
-- Run periodically via a job or a background goroutine
DELETE FROM idempotency_keys
WHERE created_at < NOW() - INTERVAL '24 hours';

package middleware

import (
    "bytes"
    "database/sql"
    "fmt"
    "net/http"
    "time"

    "github.com/jmoiron/sqlx"
)

type PgIdempotencyStore struct {
    db *sqlx.DB
}

func NewPgIdempotencyStore(db *sqlx.DB) *PgIdempotencyStore {
    return &PgIdempotencyStore{db: db}
}

// TryAcquire attempts to set the PENDING lock on the key.
// Returns true if this is the first request, false if another is already in progress or completed.
func (s *PgIdempotencyStore) TryAcquire(ctx context.Context, key string) (bool, error) {
    res, err := s.db.ExecContext(ctx,
        `INSERT INTO idempotency_keys (key) VALUES ($1) ON CONFLICT (key) DO NOTHING`,
        key,
    )
    if err != nil {
        return false, fmt.Errorf("acquiring idempotency key: %w", err)
    }
    n, _ := res.RowsAffected()
    return n == 1, nil // true = first pass
}

// GetCompleted retrieves the result of an already-processed key.
func (s *PgIdempotencyStore) GetCompleted(ctx context.Context, key string) (statusCode int, body []byte, found bool, err error) {
    var row struct {
        Status     string `db:"status"`
        StatusCode sql.NullInt32 `db:"status_code"`
        Body       sql.NullString `db:"body"`
    }
    err = s.db.GetContext(ctx, &row,
        `SELECT status, status_code, body FROM idempotency_keys WHERE key = $1`,
        key,
    )
    if err == sql.ErrNoRows {
        return 0, nil, false, nil
    }
    if err != nil {
        return 0, nil, false, fmt.Errorf("getting idempotency key: %w", err)
    }
    if row.Status != "completed" {
        return 0, nil, false, nil // PENDING — still in progress
    }
    return int(row.StatusCode.Int32), []byte(row.Body.String), true, nil
}

// Complete marks the key as finished and stores the response.
func (s *PgIdempotencyStore) Complete(ctx context.Context, key string, statusCode int, body []byte) error {
    _, err := s.db.ExecContext(ctx,
        `UPDATE idempotency_keys SET status = 'completed', status_code = $2, body = $3 WHERE key = $1`,
        key, statusCode, string(body),
    )
    return err
}

// StartCleanup launches a background goroutine that deletes expired keys.
func (s *PgIdempotencyStore) StartCleanup(ctx context.Context, ttl time.Duration) {
    go func() {
        ticker := time.NewTicker(1 * time.Hour)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                s.db.ExecContext(ctx,
                    `DELETE FROM idempotency_keys WHERE created_at < NOW() - $1::interval`,
                    ttl.String(),
                )
            case <-ctx.Done():
                return
            }
        }
    }()
}

// responseRecorder captures the response for storage.
type responseRecorder struct {
    http.ResponseWriter
    code int
    body bytes.Buffer
}

func (r *responseRecorder) WriteHeader(code int) {
    r.code = code
    r.ResponseWriter.WriteHeader(code)
}

func (r *responseRecorder) Write(b []byte) (int, error) {
    r.body.Write(b)
    return r.ResponseWriter.Write(b)
}

// IdempotencyMiddleware guarantees idempotency of POST requests via PostgreSQL.
func IdempotencyMiddleware(store *PgIdempotencyStore) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if r.Method != http.MethodPost {
                next.ServeHTTP(w, r)
                return
            }

            key := r.Header.Get("Idempotency-Key")
            if key == "" {
                http.Error(w, `{"error": "Idempotency-Key header required"}`, http.StatusBadRequest)
                return
            }

            // Already processed? Return the cached result.
            if statusCode, body, found, err := store.GetCompleted(r.Context(), key); found && err == nil {
                w.Header().Set("X-Idempotent-Replayed", "true")
                w.WriteHeader(statusCode)
                w.Write(body)
                return
            }

            // Attempt to acquire the PENDING lock.
            acquired, err := store.TryAcquire(r.Context(), key)
            if err != nil {
                http.Error(w, `{"error": "internal error"}`, http.StatusInternalServerError)
                return
            }
            if !acquired {
                // Another request is already processing this key.
                // Simple option: 409 — client should retry in a few seconds.
                http.Error(w, `{"error": "request in progress, retry shortly"}`, http.StatusConflict)
                return
            }

            // First pass: process and store the result.
            rec := &responseRecorder{ResponseWriter: w, code: http.StatusOK}
            next.ServeHTTP(rec, r)

            store.Complete(r.Context(), key, rec.code, rec.body.Bytes())
        })
    }
}

store := middleware.NewPgIdempotencyStore(db)
store.StartCleanup(ctx, 24*time.Hour) // deletes keys > 24h in background

handler := middleware.IdempotencyMiddleware(store)(myHandler)

CREATE TABLE payments (
    id               UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    idempotency_key  UUID UNIQUE NOT NULL,
    customer_id      VARCHAR(64) NOT NULL,
    amount           DECIMAL(10,2) NOT NULL,
    currency         CHAR(3) NOT NULL,
    status           VARCHAR(20) NOT NULL DEFAULT 'pending',
    created_at       TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

INSERT INTO payments (idempotency_key, customer_id, amount, currency)
VALUES ($1, $2, $3, $4)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id, status, created_at;

func CreatePayment(ctx context.Context, db *sqlx.DB, req PaymentRequest) (*Payment, error) {
    var payment Payment

    err := db.GetContext(ctx, &payment, `
        INSERT INTO payments (idempotency_key, customer_id, amount, currency)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id, idempotency_key, customer_id, amount, currency, status, created_at
    `, req.IdempotencyKey, req.CustomerID, req.Amount, req.Currency)

    if err == sql.ErrNoRows {
        // Key already known — return the existing payment.
        err = db.GetContext(ctx, &payment,
            "SELECT * FROM payments WHERE idempotency_key = $1",
            req.IdempotencyKey,
        )
        if err != nil {
            return nil, fmt.Errorf("fetching existing payment: %w", err)
        }
        return &payment, nil
    }
    if err != nil {
        return nil, fmt.Errorf("inserting payment: %w", err)
    }

    return &payment, nil
}