Koto.Messaging.Wolverine.Postgres

Durable PostgreSQL idempotency store so consumer dedup survives restarts.

Why

IntegrationEventConsumerBase<TEvent> deduplicates events through IProcessedMessageStore. The default in-memory store loses its state on every restart — and with Kafka's at-least-once delivery that means real duplicate processing in production. This package keeps processed message ids in PostgreSQL instead, so the deduplication window survives a redeploy.

Usage

services.AddKotoWolverine();
services.AddPostgresProcessedMessageStore(
    builder.Configuration.GetConnectionString("db")!);

Call order relative to AddKotoWolverine does not matter.

Options

services.AddPostgresProcessedMessageStore(connectionString, o =>
{
    o.Schema = "koto";                          // default
    o.Table = "processed_messages";             // default
    o.AutoCreateSchema = true;                  // default; disable with external migrations
    o.CleanupInterval = TimeSpan.FromHours(1);  // default
});

• The deduplication window comes from KotoWolverineOptions.IdempotencyWindow (default 24 h) — the same option the in-memory store uses.
• The schema, table, and index are created on first use (CREATE ... IF NOT EXISTS). Set AutoCreateSchema = false to manage them with your own migrations:

CREATE SCHEMA IF NOT EXISTS koto;
CREATE TABLE IF NOT EXISTS koto.processed_messages (
    message_id   uuid PRIMARY KEY,
    processed_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS ix_processed_messages_processed_at
    ON koto.processed_messages (processed_at);

• A background service deletes entries older than the idempotency window every CleanupInterval, so the table does not grow without bound.

Semantics

Delivery stays at-least-once. The store marks an event as processed after your ConsumeAsync completes, outside a shared transaction — a crash in between redelivers the event. For strict business-level deduplication use a deterministic operation id with a unique constraint in the consumer's own storage (see the Koto ledger patterns).

Durable outbox defaults

The same connection string can back the Wolverine durable outbox — envelopes in Postgres, EF transactions, and domain-event scraping in one place:

builder.Host.UseWolverine(opts =>
{
    opts.UseKotoKafka(kafka, typeof(SomeHandler).Assembly)
        .PublishIntegrationEvents(typeof(OrderPlacedV1).Assembly)
        .UseKotoDurableOutbox(connectionString);
});