Idempotency Layers for Asynchronous Event-Driven Systems

CREATE TYPE idempotency_status AS ENUM ('started', 'completed', 'failed');

CREATE TABLE idempotency_keys (
    -- The client-provided idempotency key.
    idempotency_key VARCHAR(255) PRIMARY KEY,

    -- The current status of the operation.
    status idempotency_status NOT NULL,

    -- The user or tenant this key belongs to, crucial for multi-tenancy and indexing.
    user_id UUID NOT NULL,

    -- The HTTP status code and response body to return on subsequent requests.
    -- Stored as JSONB for flexibility.
    response_code INT,
    response_body JSONB,

    -- Timestamps for lifecycle management and debugging.
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- A lock timestamp to prevent orphaned 'started' records from blocking forever.
    locked_at TIMESTAMPTZ
);

-- A multi-column index is critical for lookup performance.
CREATE INDEX idx_idempotency_keys_user_id ON idempotency_keys (user_id);

graph TD
    subgraph Time
        direction LR
        T1 --> T2 --> T3 --> T4
    end

    subgraph Process A
        A1(T1: SELECT key 'xyz') --> A2(T2: Not found)
        A2 --> A3(T4: INSERT key 'xyz')
    end

    subgraph Process B
        B1(T1: SELECT key 'xyz') --> B2(T3: Not found)
        B2 --> B3(T4: INSERT key 'xyz')
    end

    A3 --> F1(FAIL: Unique constraint violation)
    B3 --> F2(FAIL: Unique constraint violation)

package main

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

	_ "github.com/lib/pq"
)

// IdempotencyStatus defines the state of an idempotency key.
type IdempotencyStatus string

const (
	StatusStarted   IdempotencyStatus = "started"
	StatusCompleted IdempotencyStatus = "completed"
	StatusFailed    IdempotencyStatus = "failed"
)

// IdempotencyRecord holds the state for a given key.
type IdempotencyRecord struct {
	Key          string
	Status       IdempotencyStatus
	ResponseCode int
	ResponseBody []byte
}

// Pre-defined errors for clear control flow.
var (
	ErrConflict      = errors.New("request with this idempotency key is already in progress")
	ErrAlreadyExists = errors.New("request with this idempotency key has already been completed")
)

// IdempotencyManager handles the idempotency logic.
type IdempotencyManager struct {
	DB *sql.DB
}

// NewIdempotencyManager creates a new manager.
func NewIdempotencyManager(db *sql.DB) *IdempotencyManager {
	return &IdempotencyManager{DB: db}
}

// BeginOperation starts the idempotency check within a transaction.
// It returns the existing record if found, or nil if this is a new operation.
func (m *IdempotencyManager) BeginOperation(ctx context.Context, tx *sql.Tx, key string, userID string) (*IdempotencyRecord, error) {
	// 1. Atomic INSERT ... ON CONFLICT
	// We use RETURNING to see if we actually inserted a row.
	query := `
        INSERT INTO idempotency_keys (idempotency_key, user_id, status, locked_at)
        VALUES ($1, $2, 'started', NOW())
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING idempotency_key
    `
	var insertedKey string
	err := tx.QueryRowContext(ctx, query, key, userID).Scan(&insertedKey)

	if err == nil {
		// Success! We inserted the key, this is a new operation.
		return nil, nil
	} else if err != sql.ErrNoRows {
		// Some unexpected database error occurred.
		return nil, fmt.Errorf("failed to insert idempotency key: %w", err)
	}

	// 2. If we reach here, it means ON CONFLICT was triggered. The key exists.
	// We need to fetch the existing record to decide what to do.
	query = `
        SELECT status, response_code, response_body, locked_at
        FROM idempotency_keys
        WHERE idempotency_key = $1
    `
	var status string
	var responseCode sql.NullInt64
	var responseBody sql.NullString
	var lockedAt sql.NullTime

	if err := tx.QueryRowContext(ctx, query, key).Scan(&status, &responseCode, &responseBody, &lockedAt); err != nil {
		return nil, fmt.Errorf("failed to select existing idempotency key: %w", err)
	}

	// 3. Handle existing key states
	switch IdempotencyStatus(status) {
	case StatusCompleted:
		// Operation is already done. Return the cached response.
		return &IdempotencyRecord{
			Key:          key,
			Status:       StatusCompleted,
			ResponseCode: int(responseCode.Int64),
			ResponseBody: []byte(responseBody.String),
		}, ErrAlreadyExists

	case StatusStarted:
		// Another process is working on this. Check the lock timeout.
		// For a production system, this timeout should be configurable.
		if lockedAt.Valid && time.Since(lockedAt.Time) < 5*time.Minute {
			return nil, ErrConflict
		}
		// The lock has expired. We can take over.
		fallthrough // Fall through to the FAILED case logic to retry.

	case StatusFailed:
		// Previous attempt failed or lock expired, we can retry.
		// Update the lock time and set status back to 'started'.
		updateQuery := `
            UPDATE idempotency_keys
            SET status = 'started', locked_at = NOW(), updated_at = NOW()
            WHERE idempotency_key = $1
        `
		_, err := tx.ExecContext(ctx, updateQuery, key)
		if err != nil {
			return nil, fmt.Errorf("failed to relock idempotency key: %w", err)
		}
		return nil, nil // Proceed with business logic

	default:
		return nil, fmt.Errorf("unknown idempotency status: %s", status)
	}
}

// CompleteOperation marks an operation as successful.
func (m *IdempotencyManager) CompleteOperation(ctx context.Context, tx *sql.Tx, key string, code int, body interface{}) error {
	bodyJSON, err := json.Marshal(body)
	if err != nil {
		return fmt.Errorf("failed to marshal response body: %w", err)
	}

	query := `
        UPDATE idempotency_keys
        SET status = 'completed', response_code = $2, response_body = $3, updated_at = NOW(), locked_at = NULL
        WHERE idempotency_key = $1
    `
	_, err = tx.ExecContext(ctx, query, key, code, bodyJSON)
	return err
}

// FailOperation marks an operation as failed.
func (m *IdempotencyManager) FailOperation(ctx context.Context, tx *sql.Tx, key string) error {
	query := `
        UPDATE idempotency_keys
        SET status = 'failed', updated_at = NOW(), locked_at = NULL
        WHERE idempotency_key = $1
    `
	_, err := tx.ExecContext(ctx, query, key)
	return err
}

// PaymentService handles payment processing.
type PaymentService struct {
	DB         *sql.DB
	IdemManager *IdempotencyManager
}

// ChargeRequest represents the input for a charge.
type ChargeRequest struct {
	Amount   int    `json:"amount"`
	Currency string `json:"currency"`
}

// ChargeResponse represents the output.
type ChargeResponse struct {
	TransactionID string `json:"transactionId"`
	Status        string `json:"status"`
}

func (s *PaymentService) CreateCharge(ctx context.Context, idempotencyKey, userID string, req ChargeRequest) (*ChargeResponse, int, error) {
	tx, err := s.DB.BeginTx(ctx, nil)
	if err != nil {
		return nil, 500, fmt.Errorf("could not begin transaction: %w", err)
	}
	defer tx.Rollback() // Rollback is a no-op if the transaction is committed.

	// 1. Begin idempotency check
	existingRecord, err := s.IdemManager.BeginOperation(ctx, tx, idempotencyKey, userID)
	if err != nil {
		if errors.Is(err, ErrConflict) {
			return nil, 409, err // 409 Conflict
		} 
        if errors.Is(err, ErrAlreadyExists) {
			// Return the cached response
			var cachedResp ChargeResponse
			if unmarshalErr := json.Unmarshal(existingRecord.ResponseBody, &cachedResp); unmarshalErr != nil {
				return nil, 500, fmt.Errorf("failed to unmarshal cached response: %w", unmarshalErr)
			}
			return &cachedResp, existingRecord.ResponseCode, nil // e.g., 201 Created
		}
		// Any other error is a server error
		return nil, 500, fmt.Errorf("idempotency check failed: %w", err)
	}

	// 2. Execute business logic (only if it's a new operation)
	// This is a placeholder for your actual payment gateway integration.
	transactionID, err := s.processPaymentInGateway(ctx, req.Amount, req.Currency)
	if err != nil {
		// Mark operation as failed and roll back
		s.IdemManager.FailOperation(ctx, tx, idempotencyKey)
		tx.Commit() // Commit the 'failed' state!
		return nil, 500, fmt.Errorf("payment processing failed: %w", err)
	}

	// Also, persist the charge in our own database.
	// This happens inside the same transaction.
	_, err = tx.ExecContext(ctx, "INSERT INTO charges (id, user_id, amount) VALUES ($1, $2, $3)", transactionID, userID, req.Amount)
	if err != nil {
		// No need to call FailOperation here, the defer tx.Rollback() will handle it.
		return nil, 500, fmt.Errorf("failed to save charge record: %w", err)
	}

	// 3. Complete the operation successfully
	resp := &ChargeResponse{TransactionID: transactionID, Status: "succeeded"}
	if err := s.IdemManager.CompleteOperation(ctx, tx, idempotencyKey, 201, resp); err != nil {
		return nil, 500, fmt.Errorf("failed to complete idempotency record: %w", err)
	}

	// 4. Atomically commit everything
	if err := tx.Commit(); err != nil {
		return nil, 500, fmt.Errorf("failed to commit transaction: %w", err)
	}

	return resp, 201, nil
}

func (s *PaymentService) processPaymentInGateway(ctx context.Context, amount int, currency string) (string, error) {
	// In a real system, this would make an RPC/HTTP call to a payment provider.
	// For this example, we'll just simulate success.
	fmt.Printf("Processing payment of %d %s...\n", amount, currency)
	time.Sleep(100 * time.Millisecond)
	return fmt.Sprintf("txn_%d", time.Now().UnixNano()), nil
}

CREATE TABLE idempotency_keys_partitioned (
    -- columns as before
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);

-- Create partitions for today and tomorrow
CREATE TABLE idempotency_keys_y2023m10d26 PARTITION OF idempotency_keys_partitioned
    FOR VALUES FROM ('2023-10-26 00:00:00+00') TO ('2023-10-27 00:00:00+00');

CREATE TABLE idempotency_keys_y2023m10d27 PARTITION OF idempotency_keys_partitioned
    FOR VALUES FROM ('2023-10-27 00:00:00+00') TO ('2023-10-28 00:00:00+00');

-- A cron job would create new partitions and drop old ones.
-- DROP TABLE idempotency_keys_y2023m10d25;

UPDATE idempotency_keys
SET status = 'failed', locked_at = NULL, updated_at = NOW()
WHERE status = 'started' AND locked_at < NOW() - INTERVAL '5 minutes';