Idempotency in Event-Driven Systems: Outbox Pattern & Distributed Locks

// PSEUDOCODE - DO NOT USE IN PRODUCTION
func handleMessage(msg Message) {
    idempotencyKey := msg.ID

    // 1. Check if we've seen this key
    isProcessed, err := db.CheckIfProcessed(idempotencyKey)
    if err != nil { /* handle error */ }
    if isProcessed {
        ack(msg) // Already done, just acknowledge
        return
    }

    // 2. Process the message
    processBusinessLogic(msg.Data)

    // 3. Mark as processed
    db.MarkAsProcessed(idempotencyKey)

    ack(msg)
}

CREATE TABLE outbox_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_id VARCHAR(255) NOT NULL, -- ID of the entity that was changed (e.g., order_id)
    aggregate_type VARCHAR(255) NOT NULL, -- Type of entity (e.g., 'order')
    event_type VARCHAR(255) NOT NULL, -- e.g., 'OrderCreated', 'OrderShipped'
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ NULL -- Null until the relay process publishes it
);

-- Index for the relay process to efficiently find unpublished events
CREATE INDEX idx_outbox_events_unpublished ON outbox_events (published_at) WHERE published_at IS NULL;

package orderservice

import (
	"context"
	"encoding/json"
	"time"

	"github.com/google/uuid"
	"github.com/jmoiron/sqlx"
)

// Order represents our business entity
type Order struct {
	ID        string    `db:"id"`
	CustomerID string    `db:"customer_id"`
	Amount    int64     `db:"amount"`
	CreatedAt time.Time `db:"created_at"`
}

// OrderCreatedEvent is the payload for our event
type OrderCreatedEvent struct {
	OrderID    string `json:"order_id"`
	CustomerID string `json:"customer_id"`
	Amount     int64  `json:"amount"`
}

// Service handles order creation
type Service struct {
	DB *sqlx.DB
}

func (s *Service) CreateOrder(ctx context.Context, customerID string, amount int64) (*Order, error) {
	// Begin a new database transaction
	tx, err := s.DB.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	// Defer a rollback in case of error. The transaction will be committed on success.
	defer tx.Rollback() // This is a no-op if Commit() was called

	// 1. Create the Order entity
	order := &Order{
		ID:         uuid.New().String(),
		CustomerID: customerID,
		Amount:     amount,
		CreatedAt:  time.Now(),
	}

	orderSQL := `INSERT INTO orders (id, customer_id, amount, created_at) VALUES (:id, :customer_id, :amount, :created_at)`
	_, err = tx.NamedExecContext(ctx, orderSQL, order)
	if err != nil {
		return nil, err
	}

	// 2. Create the Outbox Event within the same transaction
	eventPayload := OrderCreatedEvent{
		OrderID:    order.ID,
		CustomerID: order.CustomerID,
		Amount:     order.Amount,
	}
	payloadBytes, err := json.Marshal(eventPayload)
	if err != nil {
		return nil, err
	}

	outboxSQL := `INSERT INTO outbox_events (aggregate_id, aggregate_type, event_type, payload) VALUES ($1, $2, $3, $4)`
	_, err = tx.ExecContext(ctx, outboxSQL, order.ID, "order", "OrderCreated", payloadBytes)
	if err != nil {
		return nil, err
	}

	// 3. Commit the transaction
	if err := tx.Commit(); err != nil {
		return nil, err
	}

	return order, nil
}

package relay

import (
	"context"
	"log"
	"time"

	"github.com/jmoiron/sqlx"
	// Assume a kafkaProducer client exists
)

type OutboxEvent struct {
	ID           uuid.UUID `db:"id"`
	EventType    string    `db:"event_type"`
	Payload      []byte    `db:"payload"`
	PublishedAt  *time.Time `db:"published_at"`
}

func RunRelay(db *sqlx.DB, producer kafkaProducer) {
	for {
		select {
		case <-time.After(5 * time.Second): // Poll every 5 seconds
			processOutbox(db, producer)
		}
	}
}

func processOutbox(db *sqlx.DB, producer kafkaProducer) {
	// Use a transaction to ensure we lock the rows we're about to process
	tx, err := db.BeginTxx(context.Background(), nil)
	if err != nil {
		log.Printf("Error starting transaction: %v", err)
		return
	}
	defer tx.Rollback()

	// Select and lock a batch of unpublished events
	var events []OutboxEvent
	query := `SELECT * FROM outbox_events WHERE published_at IS NULL ORDER BY created_at ASC LIMIT 100 FOR UPDATE SKIP LOCKED`
	err = tx.Select(&events, query)
	if err != nil {
		log.Printf("Error selecting events: %v", err)
		return
	}

	if len(events) == 0 {
		return
	}

	for _, event := range events {
		if err := producer.Publish("order_events", event.EventType, event.Payload); err != nil {
			log.Printf("Failed to publish event %s: %v", event.ID, err)
			// If publishing fails, the transaction will roll back and we'll retry later
			return
		}
	}

	// Mark all events in this batch as published
	ids := make([]uuid.UUID, len(events))
	for i, e := range events {
		ids[i] = e.ID
	}
	query, args, err := sqlx.In(`UPDATE outbox_events SET published_at = NOW() WHERE id IN (?)`, ids)
	if err != nil {
		log.Printf("Error building update query: %v", err)
		return
	}
	query = db.Rebind(query)
	_, err = tx.Exec(query, args...)
	if err != nil {
		log.Printf("Error marking events as published: %v", err)
		return
	}

	if err := tx.Commit(); err != nil {
		log.Printf("Error committing transaction: %v", err)
	}

	log.Printf("Successfully relayed %d events", len(events))
}

package locking

import (
	"context"
	"time"

	"github.com/go-redis/redis/v8"
)

// DistributedLockManager provides an interface for acquiring and releasing locks.
type DistributedLockManager interface {
	AcquireLock(ctx context.Context, key string, ttl time.Duration) (bool, error)
	ReleaseLock(ctx context.comtext, key string) error
}

// redisLockManager implements the lock manager using Redis.
type redisLockManager struct {
	client *redis.Client
}

// NewRedisLockManager creates a new Redis-backed lock manager.
func NewRedisLockManager(client *redis.Client) DistributedLockManager {
	return &redisLockManager{client: client}
}

// AcquireLock attempts to acquire a lock for a given key with a TTL.
// It uses the SET command with NX (Not Exists) and PX (millisecond TTL) options for atomicity.
func (r *redisLockManager) AcquireLock(ctx context.Context, key string, ttl time.Duration) (bool, error) {
	// A value of "1" is arbitrary. The existence of the key is what matters.
	result, err := r.client.SetNX(ctx, key, 1, ttl).Result()
	if err != nil {
		return false, err
	}
	return result, nil
}

// ReleaseLock releases a lock. This uses a Lua script to ensure atomicity.
// We only delete the key if its current value matches what we expect, preventing
// a consumer from deleting a lock that was acquired by another process after its own expired.
// For simplicity in this example, we'll just use a simple DEL, but a production system
// should use a more robust script with a unique lock token.
func (r *redisLockManager) ReleaseLock(ctx context.Context, key string) error {
	// WARNING: Simple DEL is not perfectly safe. See edge cases section.
	_, err := r.client.Del(ctx, key).Err()
	return err
}

CREATE TABLE processed_messages (
    idempotency_key VARCHAR(255) PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

package consumer

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

	"github.com/jmoiron/sqlx"
	"my-project/locking"
)

// Assume message format from Kafka/RabbitMQ
type Message struct {
	ID      string // Unique ID from the outbox table
	Payload json.RawMessage
}

// OrderCreatedEvent is the structure of our event payload
type OrderCreatedEvent struct {
	OrderID    string `json:"order_id"`
	CustomerID string `json:"customer_id"`
	Amount     int64  `json:"amount"`
}

type Consumer struct {
	DB          *sqlx.DB
	LockManager locking.DistributedLockManager
}

func (c *Consumer) HandleOrderCreated(ctx context.Context, msg Message) error {
	// The outbox event ID is a perfect, unique idempotency key.
	idempotencyKey := msg.ID
	lockKey := fmt.Sprintf("lock:order_created:%s", idempotencyKey)
	lockTTL := 30 * time.Second // The max expected processing time

	// 1. Attempt to acquire the distributed lock
	acquired, err := c.LockManager.AcquireLock(ctx, lockKey, lockTTL)
	if err != nil {
		log.Printf("Error acquiring lock for key %s: %v", lockKey, err)
		return err // Return error to trigger redelivery
	}

	if !acquired {
		log.Printf("Could not acquire lock for key %s, another consumer is processing.", lockKey)
		// Acknowledge the message. We don't want to process it, and another consumer
		// is already handling it. If that one fails, the broker will redeliver and
		// another instance (maybe this one) can try again later.
		return nil
	}
	// Ensure the lock is released when the function returns
	defer c.LockManager.ReleaseLock(ctx, lockKey)

	// 2. Within the lock, check if the message has already been successfully processed.
	// This handles retries after a crash post-processing but before acknowledgement.
	var keyExists string
	err = c.DB.GetContext(ctx, &keyExists, "SELECT idempotency_key FROM processed_messages WHERE idempotency_key = $1", idempotencyKey)
	if err == nil {
		log.Printf("Message %s already processed, skipping.", idempotencyKey)
		return nil // Acknowledge and exit
	} else if err != sql.ErrNoRows {
		log.Printf("Error checking processed_messages table: %v", err)
		return err // Redeliver on DB error
	}

	// 3. Begin a transaction for the business logic and idempotency key storage
	tx, err := c.DB.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// 4. Execute the actual business logic
	var event OrderCreatedEvent
	if err := json.Unmarshal(msg.Payload, &event); err != nil {
		log.Printf("Error unmarshalling payload for message %s: %v", idempotencyKey, err)
		// This is likely a poison pill. Send to DLQ and acknowledge.
		return nil
	}

	// Example: Update a customer's lifetime value
	_, err = tx.ExecContext(ctx, 
        "UPDATE customers SET lifetime_value = lifetime_value + $1 WHERE id = $2", 
        event.Amount, event.CustomerID)
	if err != nil {
		log.Printf("Business logic failed for message %s: %v", idempotencyKey, err)
		return err // Redeliver
	}

	// 5. Atomically mark the message as processed
	_, err = tx.ExecContext(ctx, "INSERT INTO processed_messages (idempotency_key) VALUES ($1)", idempotencyKey)
	if err != nil {
		log.Printf("Failed to mark message %s as processed: %v", idempotencyKey, err)
		return err // Redeliver
	}

	// 6. Commit the transaction
	if err := tx.Commit(); err != nil {
		return err
	}

	log.Printf("Successfully processed message %s", idempotencyKey)
	return nil // Acknowledge the message
}

-- Lua script for safe lock release
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end