Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Eventuous/eventuous/llms.txt

Use this file to discover all available pages before exploring further.

The Eventuous.Postgresql package lets you use a standard PostgreSQL database as a fully-featured Eventuous event store. It stores streams and events in a schema managed by Eventuous itself and provides polling-based subscriptions for both all-stream and single-stream reads, plus a base class for writing SQL projections.

​
Installation

dotnet add package Eventuous.Postgresql
The package depends on Npgsql for all database access. Make sure your application uses a compatible version.

​
Schema

Eventuous manages its own schema inside a configurable PostgreSQL schema (default: eventuous). The schema includes:
ObjectPurpose
{schema}.append_eventsStored function to append events atomically
{schema}.read_stream_forwardsRead a stream from a given position
{schema}.read_stream_backwardsRead a stream backwards
{schema}.read_all_forwardsRead all events globally, ordered by position
{schema}.read_stream_subSubscription read for a single stream
{schema}.check_streamValidate expected version
{schema}.streamsStream metadata table
{schema}.checkpointsSubscription checkpoint table

​
Schema initialisation

SchemaInitializer is a hosted service that creates (or re-creates) the schema on startup when PostgresStoreOptions.InitializeDatabase is set to true. 
// In DI setup:
services.Configure<PostgresStoreOptions>(opts => {
    opts.Schema             = "eventuous"; // default
    opts.ConnectionString   = connectionString;
    opts.InitializeDatabase = true;        // create schema on startup
});
services.AddHostedService<SchemaInitializer>();
For production you may prefer to run migrations out-of-band and leave InitializeDatabase as false.

​
Event Store

PostgresStore implements IEventStore on top of an NpgsqlDataSource. 
var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
var store = new PostgresStore(dataSource, new PostgresStoreOptions());
PostgresStoreOptions properties:
PropertyDefaultDescription
Schema"eventuous"PostgreSQL schema name
ConnectionString—Connection string (used by SchemaInitializer)
InitializeDatabasefalseCreate schema automatically on startup

​
Dependency injection

var connectionString = builder.Configuration.GetConnectionString("Postgres")!;

// Register the NpgsqlDataSource as a singleton
builder.Services.AddSingleton(
    new NpgsqlDataSourceBuilder(connectionString).Build()
);

// Configure store options (schema name, etc.)
builder.Services.Configure<PostgresStoreOptions>(opts => {
    opts.Schema             = "eventuous";
    opts.ConnectionString   = connectionString;
    opts.InitializeDatabase = true;
});

// Register the event store
builder.Services.AddEventStore<PostgresStore>();

// Register schema initializer to run on startup
builder.Services.AddHostedService<SchemaInitializer>();

​
Subscriptions

Both subscription types poll the database on a configurable interval, making them compatible with any standard PostgreSQL instance without requiring logical replication.

​
PostgresAllStreamSubscription

Reads every event in the database in global-position order, advancing a persisted checkpoint after each page.
  • Options type: PostgresAllStreamSubscriptionOptions
services.AddSubscription<PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions>(
    "AllEventsProjection",
    builder => builder
        .UseCheckpointStore<PostgresCheckpointStore>()
        .AddEventHandler<ReadModelProjection>()
);

​
PostgresStreamSubscription

Reads events from a single named stream, tracking position per-stream.
  • Options type: PostgresStreamSubscriptionOptions
  • Key option: Stream — the StreamName to subscribe to
services.AddSubscription<PostgresStreamSubscription, PostgresStreamSubscriptionOptions>(
    "OrderStream",
    builder => builder
        .Configure(x => x.Stream = new StreamName("Order"))
        .UseCheckpointStore<PostgresCheckpointStore>()
        .AddEventHandler<OrderProjection>()
);

​
Checkpoint store

PostgresCheckpointStore persists subscription positions back into the {schema}.checkpoints table, so you can keep everything in the same PostgreSQL instance.

​
Projections

PostgresProjector is an abstract base class for event handlers that write read models into PostgreSQL tables. Override it and use On<TEvent> to register SQL commands per event type. 
public class BookingProjector : PostgresProjector {
    public BookingProjector(NpgsqlDataSource dataSource) : base(dataSource) {
        On<BookingRegistered>(
            (connection, ctx) => Project(
                connection,
                "INSERT INTO bookings (id, room_id, checkin, checkout) VALUES (@id, @room, @in, @out)",
                ("id",   ctx.Message.BookingId),
                ("room", ctx.Message.RoomId),
                ("in",   ctx.Message.CheckIn.ToString()),
                ("out",  ctx.Message.CheckOut.ToString())
            )
        );

        On<BookingCancelled>(
            (connection, ctx) => Project(
                connection,
                "UPDATE bookings SET cancelled = TRUE WHERE id = @id",
                ("id", ctx.Message.BookingId)
            )
        );
    }
}
The static Project helper builds an NpgsqlCommand from a SQL string and either NpgsqlParameter objects or (Name, Value) tuples, cutting down on boilerplate. Register the projector as a subscription event handler: 
services.AddSubscription<PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions>(
    "BookingProjections",
    builder => builder
        .UseCheckpointStore<PostgresCheckpointStore>()
        .AddEventHandler<BookingProjector>()
);

Build docs developers (and LLMs) love

Get started for freeTalk to us