Idempotent Sagas: Exactly-Once Processing in Event-Driven Systems

// DO NOT DO THIS - RACE CONDITION
func (s *PaymentService) handleDebitEvent(event OrderCreated) {
    // 1. Check if payment exists
    exists, _ := s.db.PaymentExistsForOrder(event.OrderID)
    if exists {
        return // Assume already processed
    }

    // 2. Create payment
    _ = s.db.CreatePayment(event.OrderID, event.Amount)
}

-- Status of the idempotent operation
CREATE TYPE idempotency_status AS ENUM ('pending', 'completed', 'failed');

-- Table to store idempotency keys and their results
CREATE TABLE idempotency_keys (
    -- The key itself, provided by the client/producer
    key VARCHAR(255) PRIMARY KEY,

    -- The name of the user/actor performing the action, for locking and scoping
    -- In our event-driven case, this could be the consumer group ID or a static service name.
    actor_id VARCHAR(255) NOT NULL,

    -- The current status of the operation
    status idempotency_status NOT NULL DEFAULT 'pending',

    -- The result of the operation, stored as JSON to be flexible
    response_payload JSONB,

    -- The HTTP status code or an equivalent result code for the operation
    response_code INT,

    -- Timestamp for when the operation was locked for processing
    locked_at TIMESTAMPTZ,

    -- Timestamps for creation and updates
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Crucial index for fast lookups
CREATE INDEX idx_idempotency_keys_key ON idempotency_keys(key);

-- Optional: Index for cleanup jobs
CREATE INDEX idx_idempotency_keys_created_at ON idempotency_keys(created_at);

CREATE TABLE payments (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_id UUID NOT NULL UNIQUE,
    amount_cents INT NOT NULL,
    status VARCHAR(50) NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/google/uuid"
	_ "github.com/lib/pq"
)

// Represents the stored state of an idempotent operation
type IdempotencyKey struct {
	Key             string          `db:"key"`
	ActorID         string          `db:"actor_id"`
	Status          string          `db:"status"`
	ResponsePayload json.RawMessage `db:"response_payload"`
	ResponseCode    int             `db:"response_code"`
	LockedAt        sql.NullTime    `db:"locked_at"`
	CreatedAt       time.Time       `db:"created_at"`
	UpdatedAt       time.Time       `db:"updated_at"`
}

// A generic response structure to store in the idempotency table
type HandlerResponse struct {
	StatusCode int
	Body       []byte
}

// The signature for our actual business logic handlers
type IdempotentHandler func(ctx context.Context, tx *sql.Tx, eventPayload []byte) (*HandlerResponse, error)

// The middleware that wraps the business logic handler
func IdempotencyWrapper(db *sql.DB, handler IdempotentHandler) func(ctx context.Context, idempotencyKey string, eventPayload []byte) (*HandlerResponse, error) {
	return func(ctx context.Context, idempotencyKey string, eventPayload []byte) (*HandlerResponse, error) {
		// 1. Begin Transaction
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			return nil, fmt.Errorf("failed to begin transaction: %w", err)
		}
		defer tx.Rollback() // Rollback on any error

		// 2. Lock and Check for Existing Key. THIS IS THE CRITICAL STEP.
		// `FOR UPDATE` acquires a row-level lock, preventing concurrent processors
		// from interfering with this specific key.
		var keyRecord IdempotencyKey
		row := tx.QueryRowContext(ctx, `SELECT key, status, response_payload, response_code FROM idempotency_keys WHERE key = $1 FOR UPDATE`, idempotencyKey)
		err = row.Scan(&keyRecord.Key, &keyRecord.Status, &keyRecord.ResponsePayload, &keyRecord.ResponseCode)

		if err != nil && !errors.Is(err, sql.ErrNoRows) {
			return nil, fmt.Errorf("failed to select idempotency key: %w", err)
		}

		// Case A: Key Found - Operation was already processed or is in progress
		if err == nil {
			if keyRecord.Status == "completed" {
				log.Printf("Request with key %s already completed. Returning stored response.", idempotencyKey)
				return &HandlerResponse{
					StatusCode: keyRecord.ResponseCode,
					Body:       keyRecord.ResponsePayload,
				}, nil
			} else if keyRecord.Status == "pending" {
                // This could indicate a crash during a previous attempt.
                // We can decide to either fail or proceed, depending on business requirements.
                // For this example, we will proceed, as the transaction was rolled back.
                log.Printf("Request with key %s found in pending state. Retrying operation.", idempotencyKey)
            }
		} else { // Case B: Key Not Found - First time seeing this request
			log.Printf("First time seeing key %s. Inserting new record.", idempotencyKey)
			_, err = tx.ExecContext(ctx,
				`INSERT INTO idempotency_keys (key, actor_id, status, locked_at) VALUES ($1, $2, 'pending', NOW())`,
				idempotencyKey, "payment-service-v1")
			if err != nil {
				return nil, fmt.Errorf("failed to insert idempotency key: %w", err)
			}
		}

		// 3. Execute Core Business Logic
		response, err := handler(ctx, tx, eventPayload)
		if err != nil {
			// Business logic failed. We can optionally mark the key as 'failed'
			// to prevent retries for non-transient errors.
			_, updateErr := tx.ExecContext(ctx, `UPDATE idempotency_keys SET status = 'failed', updated_at = NOW() WHERE key = $1`, idempotencyKey)
			if updateErr != nil {
				log.Printf("CRITICAL: Failed to mark key %s as failed after business logic error: %v", idempotencyKey, updateErr)
			}
			return nil, fmt.Errorf("handler logic failed: %w", err)
		}

		// 4. Persist the successful result
		_, err = tx.ExecContext(ctx,
			`UPDATE idempotency_keys SET status = 'completed', response_payload = $1, response_code = $2, updated_at = NOW() WHERE key = $3`,
			response.Body, response.StatusCode, idempotencyKey)
		if err != nil {
			return nil, fmt.Errorf("failed to update idempotency key to completed: %w", err)
		}

		// 5. Commit the entire transaction
		if err := tx.Commit(); err != nil {
			return nil, fmt.Errorf("failed to commit transaction: %w", err)
		}

		log.Printf("Successfully processed request with key %s.", idempotencyKey)
		return response, nil
	}
}

// Represents the payload of our OrderCreated event
type OrderCreatedEvent struct {
	OrderID    string `json:"order_id"`
	CustomerID string `json:"customer_id"`
	Amount     int    `json:"amount_cents"`
}

// processDebit is our core business logic.
func processDebit(ctx context.Context, tx *sql.Tx, eventPayload []byte) (*HandlerResponse, error) {
	var event OrderCreatedEvent
	if err := json.Unmarshal(eventPayload, &event); err != nil {
		return nil, fmt.Errorf("invalid event payload: %w", err)
	}

	// Business logic: create a payment record.
	// In a real system, this would also call out to a payment gateway, etc.
	// All database operations MUST use the provided `tx` object.
	_, err := tx.ExecContext(ctx,
		`INSERT INTO payments (order_id, amount_cents, status) VALUES ($1, $2, 'completed')`,
		event.OrderID, event.Amount)

	if err != nil {
        // Check for unique constraint violation on order_id to be extra safe
        // This can happen if the idempotency key logic somehow fails.
		return nil, fmt.Errorf("failed to insert payment: %w", err)
	}

	responseBody, _ := json.Marshal(map[string]string{"status": "success", "order_id": event.OrderID})
	return &HandlerResponse{
		StatusCode: 200,
		Body:       responseBody,
	}, nil
}

// Main function to simulate a Kafka consumer
func main() {
	// Database connection setup
	dbURL := "postgres://user:password@localhost:5432/idempotency_db?sslmode=disable"
	db, err := sql.Open("postgres", dbURL)
	if err != nil {
		log.Fatalf("Could not connect to database: %v", err)
	}
	defer db.Close()

	// Wrap our business logic with the idempotency middleware
	debitHandler := IdempotencyWrapper(db, processDebit)

	// --- Simulate receiving two identical Kafka messages ---
	orderID := uuid.New().String()
	eventPayload, _ := json.Marshal(OrderCreatedEvent{
		OrderID:    orderID,
		CustomerID: uuid.New().String(),
		Amount:     10000, // 100.00
	})

	// The producer of the event generates the idempotency key
	idempotencyKey := uuid.New().String()

	log.Println("--- First attempt --- ")
	_, err = debitHandler(context.Background(), idempotencyKey, eventPayload)
	if err != nil {
		log.Fatalf("First attempt failed: %v", err)
	}

	log.Println("--- Second attempt (duplicate) --- ")
	_, err = debitHandler(context.Background(), idempotencyKey, eventPayload)
	if err != nil {
		log.Fatalf("Second attempt failed: %v", err)
	}
}

    DELETE FROM idempotency_keys WHERE created_at < NOW() - INTERVAL '48 hours';

// Inside the IdempotencyWrapper
response, err := handler(ctx, tx, eventPayload)
if err != nil {
    var insufficientFundsErr *InsufficientFundsError
    if errors.As(err, &insufficientFundsErr) {
        // This is a terminal failure. Mark it as such.
        _, updateErr := tx.ExecContext(ctx, `UPDATE idempotency_keys SET status = 'failed', response_payload = $1, response_code = 400 WHERE key = $2`, insufficientFundsErr.Error(), idempotencyKey)
        // ... handle updateErr ...
        // We still commit the transaction to save the 'failed' state!
        tx.Commit()
        return nil, err // Return the original error to the consumer loop
    }
    // For other errors, we let the defer tx.Rollback() handle it.
    return nil, err
}