Idempotent Handlers

using Microsoft.EntityFrameworkCore;
using Npgsql;

public sealed class ProcessedMessage
{
    public string Consumer { get; set; } = "";
    public Guid MessageId { get; set; }
    public DateTimeOffset ProcessedAt { get; set; }
}

public sealed class StockReservation
{
    public int Id { get; set; }
    public Guid OrderId { get; set; }
    public string Sku { get; set; } = "";
    public int Quantity { get; set; }
}

public sealed class InventoryDb(DbContextOptions<InventoryDb> options) : DbContext(options)
{
    public DbSet<ProcessedMessage> ProcessedMessages => Set<ProcessedMessage>();
    public DbSet<StockReservation> Reservations => Set<StockReservation>();

    protected override void OnModelCreating(ModelBuilder b) =>
        b.Entity<ProcessedMessage>().HasKey(m => new { m.Consumer, m.MessageId }).HasName("pk_processed_messages");
}

public sealed class ReserveStockHandler(InventoryDb db, TimeProvider time)
{
    public async Task<bool> HandleAsync(Guid messageId, Guid orderId, string sku, int quantity, CancellationToken ct)
    {
        db.ProcessedMessages.Add(new ProcessedMessage { Consumer = "reserve-stock", MessageId = messageId, ProcessedAt = time.GetUtcNow() });
        db.Reservations.Add(new StockReservation { OrderId = orderId, Sku = sku, Quantity = quantity });
        try
        {
            await db.SaveChangesAsync(ct); // one transaction: marker and reservation together
            return true;
        }
        // Only the marker's key means "duplicate"; any other unique violation is a real failure.
        catch (DbUpdateException ex) when (ex.InnerException is PostgresException
            { SqlState: PostgresErrorCodes.UniqueViolation, ConstraintName: "pk_processed_messages" })
        {
            db.ChangeTracker.Clear();
            return false; // already processed: ack and move on
        }
    }
}

CREATE TABLE idempotency_keys (
    client_id        text        NOT NULL,
    idempotency_key  text        NOT NULL,
    request_hash     char(64)    NOT NULL,
    status_code      integer     NULL,      -- NULL while in progress
    response_body    jsonb       NULL,
    locked_until     timestamptz NULL,      -- lease on the in-progress state
    created_at       timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (client_id, idempotency_key)
);

-- Claim the key. Zero rows returned means another request already holds it.
INSERT INTO idempotency_keys (client_id, idempotency_key, request_hash, locked_until)
VALUES ('mobile-app', '6f1c2a9e-8f3b-4d1e-9a51-0c4b7e2d9f10', 'A1B2C3D4E5F60718293A4B5C6D7E8F90A1B2C3D4E5F60718293A4B5C6D7E8F90',
        now() + interval '30 seconds')
ON CONFLICT (client_id, idempotency_key) DO NOTHING
RETURNING created_at;

-- Reclaim an abandoned claim: still in progress, lease expired (the holder crashed).
UPDATE idempotency_keys
SET locked_until = now() + interval '30 seconds'
WHERE client_id = 'mobile-app' AND idempotency_key = '6f1c2a9e-8f3b-4d1e-9a51-0c4b7e2d9f10'
  AND status_code IS NULL AND locked_until < now()
RETURNING created_at;

-- On a 5xx or unhandled exception, release the claim so a retry can run.
DELETE FROM idempotency_keys
WHERE client_id = 'mobile-app' AND idempotency_key = '6f1c2a9e-8f3b-4d1e-9a51-0c4b7e2d9f10'
  AND status_code IS NULL;

-- Natural keys stop duplicates even when a client sends no header.
CREATE UNIQUE INDEX ux_orders_client_reference ON orders (customer_id, client_reference);
CREATE UNIQUE INDEX ux_payments_provider_charge ON payments (provider_charge_id);

-- Expire old keys daily.
DELETE FROM idempotency_keys WHERE created_at < now() - interval '24 hours';

using Microsoft.EntityFrameworkCore;

public sealed class Payment
{
    public Guid Id { get; set; }
    public string Status { get; set; } = "Authorized";
    public DateTimeOffset? CapturedAt { get; set; }
}

public sealed class PaymentsDb(DbContextOptions<PaymentsDb> options) : DbContext(options)
{
    public DbSet<Payment> Payments => Set<Payment>();
}

public enum CaptureResult { Captured, AlreadyCaptured, NotFound }

public sealed class CapturePaymentHandler(PaymentsDb db, TimeProvider time)
{
    public async Task<CaptureResult> HandleAsync(Guid paymentId, CancellationToken ct)
    {
        var now = time.GetUtcNow();
        var updated = await db.Payments
            .Where(p => p.Id == paymentId && p.Status == "Authorized")
            .ExecuteUpdateAsync(s => s
                .SetProperty(p => p.Status, "Captured")
                .SetProperty(p => p.CapturedAt, now), ct);

        if (updated == 1) return CaptureResult.Captured;
        return await db.Payments.AnyAsync(p => p.Id == paymentId, ct)
            ? CaptureResult.AlreadyCaptured
            : CaptureResult.NotFound;
    }
}