Atharva Pandey/Lesson 27: Idempotency in Concurrent Systems — Assume everything runs twice

// WRONG — payment handler that charges twice on retry
func handleCreatePayment(w http.ResponseWriter, r *http.Request) {
    var req PaymentRequest
    json.NewDecoder(r.Body).Decode(&req)

    // If the client retries this request (network timeout, etc.),
    // we'll charge the customer again.
    charge, err := stripe.Charges.New(&stripe.ChargeParams{
        Amount:   stripe.Int64(req.Amount),
        Currency: stripe.String("usd"),
        Source:   stripe.String(req.CardToken),
    })
    if err != nil {
        http.Error(w, "payment failed", http.StatusInternalServerError)
        return
    }

    db.SavePayment(charge.ID, req.OrderID, req.Amount)
    json.NewEncoder(w).Encode(charge)
}

// WRONG — cache population without deduplication
var cache = make(map[string]*UserProfile)
var mu sync.RWMutex

func getProfile(ctx context.Context, userID string) (*UserProfile, error) {
    mu.RLock()
    if p, ok := cache[userID]; ok {
        mu.RUnlock()
        return p, nil
    }
    mu.RUnlock()

    // Multiple goroutines can reach here simultaneously for the same userID
    // They'll all hit the DB at once — thundering herd on cache miss
    profile, err := db.GetUserProfile(ctx, userID)
    if err != nil {
        return nil, err
    }

    mu.Lock()
    cache[userID] = profile
    mu.Unlock()

    return profile, nil
}

// RIGHT — idempotency key pattern for payment handler
func handleCreatePayment(w http.ResponseWriter, r *http.Request) {
    idempotencyKey := r.Header.Get("Idempotency-Key")
    if idempotencyKey == "" {
        http.Error(w, "Idempotency-Key header required", http.StatusBadRequest)
        return
    }

    var req PaymentRequest
    json.NewDecoder(r.Body).Decode(&req)

    // Check if we've already processed this key
    existing, err := db.GetPaymentByIdempotencyKey(r.Context(), idempotencyKey)
    if err == nil && existing != nil {
        // Already processed — return the same response as last time
        w.Header().Set("X-Idempotent-Replayed", "true")
        json.NewEncoder(w).Encode(existing)
        return
    }

    // Process atomically — the upsert handles the race between concurrent
    // requests with the same idempotency key
    charge, err := stripe.Charges.New(&stripe.ChargeParams{
        Amount:      stripe.Int64(req.Amount),
        Currency:    stripe.String("usd"),
        Source:      stripe.String(req.CardToken),
        IdempotencyKey: stripe.String(idempotencyKey), // Stripe supports this natively
    })
    if err != nil {
        http.Error(w, "payment failed", http.StatusInternalServerError)
        return
    }

    // Use INSERT ... ON CONFLICT to handle concurrent requests with the same key
    if err := db.SavePaymentIdempotent(r.Context(), idempotencyKey, charge); err != nil {
        // If it's a conflict error, another request already saved it — fetch and return
        existing, _ = db.GetPaymentByIdempotencyKey(r.Context(), idempotencyKey)
        json.NewEncoder(w).Encode(existing)
        return
    }

    json.NewEncoder(w).Encode(charge)
}

// RIGHT — singleflight deduplicates concurrent identical calls
import "golang.org/x/sync/singleflight"

var (
    profileGroup singleflight.Group
    cache        sync.Map
)

func getProfile(ctx context.Context, userID string) (*UserProfile, error) {
    // Check cache first (sync.Map is safe for concurrent reads)
    if cached, ok := cache.Load(userID); ok {
        return cached.(*UserProfile), nil
    }

    // singleflight ensures only one DB call happens per userID,
    // regardless of how many goroutines call getProfile concurrently.
    // All callers for the same key block and get the same result.
    result, err, _ := profileGroup.Do(userID, func() (interface{}, error) {
        // Only one goroutine executes this for a given userID at a time
        profile, err := db.GetUserProfile(ctx, userID)
        if err != nil {
            return nil, err
        }
        cache.Store(userID, profile)
        return profile, nil
    })
    if err != nil {
        return nil, err
    }
    return result.(*UserProfile), nil
}

// RIGHT — idempotent job completion using database upsert
func markJobComplete(ctx context.Context, db *sql.DB, jobID string, result JobResult) error {
    // ON CONFLICT DO UPDATE is idempotent — running it twice sets the same values
    _, err := db.ExecContext(ctx, `
        INSERT INTO job_results (job_id, status, result_data, completed_at)
        VALUES ($1, 'completed', $2, NOW())
        ON CONFLICT (job_id)
        DO UPDATE SET
            status = EXCLUDED.status,
            result_data = EXCLUDED.result_data,
            -- Don't update completed_at on replay — keep the original timestamp
            completed_at = job_results.completed_at
    `, jobID, result.Data)
    return err
}

// At-least-once consumer: this can safely run for the same job twice
func processJobMessage(ctx context.Context, db *sql.DB, msg Message) error {
    job, err := parseJob(msg)
    if err != nil {
        return err // malformed — don't ack, let dead letter queue handle it
    }

    // Idempotency check at the start — fast path for duplicates
    if done, _ := isJobComplete(ctx, db, job.ID); done {
        return nil // already done, ack the duplicate
    }

    result, err := executeJob(ctx, job)
    if err != nil {
        return err // processing failed — don't ack, allow retry
    }

    // Upsert the result — safe to call twice
    return markJobComplete(ctx, db, job.ID, result)
}