Client → POST /orders → [network failure] → client retries
Message broker → delivers message → consumer crashes → broker redelivers
Scheduled job → runs twice (clock skew, pod restart) → duplicate processing

All three scenarios must produce the same outcome as a single execution.

Client generates a unique key for each operation (UUID).
Server stores the response for that key.
On retry with the same key: return the cached response — don't re-execute.

// Client sends Idempotency-Key header:
// POST /api/orders
// Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000
// { "customerId": 42, "items": [...] }

// Server: MediatR pipeline behaviour for idempotency
public class IdempotencyBehaviour<TRequest, TResponse>(
    IIdempotencyStore store,
    ILogger<IdempotencyBehaviour<TRequest, TResponse>> logger)
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IIdempotentCommand
{
    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var key = request.IdempotencyKey;

        // Check if we already processed this key
        var cached = await store.GetAsync<TResponse>(key, ct);
        if (cached is not null)
        {
            logger.LogInformation("Idempotent replay for key {Key}", key);
            return cached;
        }

        // Process the command
        var response = await next();

        // Store result — expire after 24 hours (clients retry within this window)
        await store.SetAsync(key, response, TimeSpan.FromHours(24), ct);
        return response;
    }
}

// Command marks itself as idempotent
public interface IIdempotentCommand
{
    string IdempotencyKey { get; }
}

public record PlaceOrderCommand(
    int    CustomerId,
    string IdempotencyKey,
    IReadOnlyList<OrderItem> Items)
    : IRequest<PlaceOrderResult>, IIdempotentCommand;

// Extract idempotency key from HTTP header in the controller
[HttpPost("orders")]
public async Task<IActionResult> PlaceOrder(
    PlaceOrderRequest request,
    [FromHeader(Name = "Idempotency-Key")] string? idempotencyKey,
    CancellationToken ct)
{
    if (string.IsNullOrEmpty(idempotencyKey))
        return BadRequest("Idempotency-Key header is required");

    if (!Guid.TryParse(idempotencyKey, out _))
        return BadRequest("Idempotency-Key must be a valid UUID");

    var command = new PlaceOrderCommand(request.CustomerId, idempotencyKey, request.Items);
    var result  = await mediator.Send(command, ct);
    return Ok(result);
}

public interface IIdempotencyStore
{
    Task<T?> GetAsync<T>(string key, CancellationToken ct = default);
    Task     SetAsync<T>(string key, T value, TimeSpan ttl, CancellationToken ct = default);
}

public class RedisIdempotencyStore(IDistributedCache cache) : IIdempotencyStore
{
    public async Task<T?> GetAsync<T>(string key, CancellationToken ct)
    {
        var raw = await cache.GetStringAsync(PrefixedKey(key), ct);
        return raw is null ? default : JsonSerializer.Deserialize<T>(raw);
    }

    public async Task SetAsync<T>(string key, T value, TimeSpan ttl, CancellationToken ct)
    {
        var json = JsonSerializer.Serialize(value);
        await cache.SetStringAsync(PrefixedKey(key), json,
            new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = ttl }, ct);
    }

    private static string PrefixedKey(string key) => $"idempotency:{key}";
}

// Idempotency table
// CREATE TABLE ProcessedCommands (
//     Key       VARCHAR(100) PRIMARY KEY,
//     Response  TEXT NOT NULL,
//     CreatedAt TIMESTAMP NOT NULL DEFAULT NOW()
// );

public class DatabaseIdempotencyStore(AppDbContext context) : IIdempotencyStore
{
    public async Task<T?> GetAsync<T>(string key, CancellationToken ct)
    {
        var record = await context.ProcessedCommands
            .FirstOrDefaultAsync(r => r.Key == key, ct);

        return record is null ? default : JsonSerializer.Deserialize<T>(record.Response);
    }

    public async Task SetAsync<T>(string key, T value, TimeSpan ttl, CancellationToken ct)
    {
        var record = new ProcessedCommand
        {
            Key      = key,
            Response = JsonSerializer.Serialize(value),
            ExpiresAt = DateTime.UtcNow.Add(ttl),
        };

        try
        {
            context.ProcessedCommands.Add(record);
            await context.SaveChangesAsync(ct);
        }
        catch (DbUpdateException)
        {
            // Unique constraint violation — another instance already stored it
            // This is fine — the response is already there
        }
    }
}

// Strategy 1: Check-before-execute using a processed message table
public class OrderCreatedConsumer(AppDbContext context, IMediator mediator)
{
    public async Task ConsumeAsync(OrderCreatedEvent @event, CancellationToken ct)
    {
        var messageId = @event.MessageId;   // unique per message

        // Check if already processed (database unique index on MessageId)
        var alreadyProcessed = await context.ProcessedMessages
            .AnyAsync(m => m.Id == messageId, ct);

        if (alreadyProcessed)
        {
            logger.LogInformation("Duplicate message {MessageId} — skipping", messageId);
            return;
        }

        // Process the message
        await mediator.Send(new HandleOrderCreatedCommand(@event.OrderId), ct);

        // Record as processed — within the same DB transaction as the work
        context.ProcessedMessages.Add(new ProcessedMessage
        {
            Id          = messageId,
            ProcessedAt = DateTime.UtcNow,
        });
        await context.SaveChangesAsync(ct);
    }
}

// Strategy 2: Upsert (natural idempotency in the operation itself)
public class InventoryConsumer(AppDbContext context)
{
    public async Task ConsumeAsync(StockUpdatedEvent @event, CancellationToken ct)
    {
        // UPSERT is naturally idempotent — running twice produces the same result
        await context.Database.ExecuteSqlInterpolatedAsync($"""
            INSERT INTO InventoryLevels (ProductId, Quantity, UpdatedAt)
            VALUES ({@event.ProductId}, {@event.Quantity}, {@event.Timestamp})
            ON CONFLICT (ProductId)
            DO UPDATE SET
                Quantity  = EXCLUDED.Quantity,
                UpdatedAt = EXCLUDED.UpdatedAt
            WHERE InventoryLevels.UpdatedAt < EXCLUDED.UpdatedAt
            """, ct);
    }
}

// Problem: saving the order and publishing the event are two separate operations
// If the app crashes between them, one succeeds and the other doesn't
await repo.SaveOrderAsync(order);       // succeeds
// CRASH HERE — event never published
await bus.PublishAsync(orderCreated);   // never runs

// Solution: Outbox Pattern — write event to the DB in the same transaction
public class CreateOrderHandler(AppDbContext context)
    : IRequestHandler<CreateOrderCommand, int>
{
    public async Task<int> Handle(CreateOrderCommand cmd, CancellationToken ct)
    {
        var order = Order.Create(cmd.CustomerId, cmd.Items);
        context.Orders.Add(order);

        // Write the outbox message in the SAME transaction as the order
        context.OutboxMessages.Add(new OutboxMessage
        {
            Id          = Guid.NewGuid(),
            Type        = nameof(OrderCreatedEvent),
            Payload     = JsonSerializer.Serialize(new OrderCreatedEvent(order.Id)),
            CreatedAt   = DateTime.UtcNow,
            ProcessedAt = null,
        });

        await context.SaveChangesAsync(ct);   // atomic: order + outbox message
        return order.Id;
    }
}

// Background worker: publish unprocessed outbox messages
public class OutboxWorker(AppDbContext context, IEventBus bus, ILogger<OutboxWorker> logger)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            var messages = await context.OutboxMessages
                .Where(m => m.ProcessedAt == null)
                .OrderBy(m => m.CreatedAt)
                .Take(20)
                .ToListAsync(ct);

            foreach (var msg in messages)
            {
                try
                {
                    await bus.PublishRawAsync(msg.Type, msg.Payload, ct);
                    msg.ProcessedAt = DateTime.UtcNow;
                }
                catch (Exception ex)
                {
                    logger.LogError(ex, "Failed to publish outbox message {Id}", msg.Id);
                }
            }

            if (messages.Count > 0)
                await context.SaveChangesAsync(ct);

            await Task.Delay(TimeSpan.FromSeconds(5), ct);
        }
    }
}