WalFlow.Stores.Postgres 0.1.0-alpha.alpha.20260708090623

WalFlow.Stores.Postgres

PostgreSQL-based store for WalFlow CDC processor. Persists replication state in a PostgreSQL table with automatic schema and table creation.

Features

• Automatic Table Creation: Automatically creates schema and table on first use
• JSONB Storage: Stores state as JSONB for flexibility and queryability
• Singleton Pattern: Thread-safe initialization with double-check locking
• Upsert Support: Uses ON CONFLICT for atomic state updates
• Monitoring Support: Includes indexed updated_at column for monitoring queries

Installation

dotnet add package WalFlow.Stores.Postgres

Configuration

appsettings.json

{
  "PostgresStore": {
    "ConnectionString": "Host=localhost;Database=mydb;Username=user;Password=pass",
    "SchemaName": "public",
    "TableName": "walflow_state",
    "AutoCreateTable": true
  }
}

Options

Usage

Basic Registration

services.Configure<PostgresStoreOptions>(
    configuration.GetSection(PostgresStoreOptions.SectionName));

services.AddSingleton<IReplicationStore, PostgresStore>();

Manual Configuration

services.AddSingleton<IReplicationStore>(sp =>
{
    var options = Options.Create(new PostgresStoreOptions
    {
        ConnectionString = "Host=localhost;Database=cdc;Username=user;Password=pass",
        SchemaName = "cdc",
        TableName = "replication_state"
    });
    
    var logger = sp.GetRequiredService<ILogger<PostgresStore>>();
    return new PostgresStore(options, logger);
});

Table Schema

The state table is automatically created with the following schema:

CREATE TABLE IF NOT EXISTS "public"."walflow_state" (
    slot_name TEXT PRIMARY KEY,
    state_json JSONB NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_walflow_state_updated_at 
ON "public"."walflow_state" (updated_at DESC);

Monitoring

Query current replication state:

-- View all slots
SELECT 
    slot_name,
    state_json->>'LastProcessedLsn' as lsn,
    state_json->>'MessagesProcessed' as messages,
    updated_at
FROM walflow_state
ORDER BY updated_at DESC;

-- Check if snapshot is completed
SELECT 
    slot_name,
    state_json->>'SnapshotCompleted' as snapshot_done,
    state_json->'CompletedSnapshotTables' as completed_tables
FROM walflow_state;

-- Monitor lag (requires joining with pg_replication_slots)
SELECT 
    ws.slot_name,
    (state_json->>'LastProcessedLsn')::pg_lsn as last_processed,
    prs.confirmed_flush_lsn,
    pg_current_wal_lsn() - (state_json->>'LastProcessedLsn')::pg_lsn as lag_bytes
FROM walflow_state ws
JOIN pg_replication_slots prs ON prs.slot_name = ws.slot_name;

Error Handling

• Connection Failures: Throws connection exceptions on failure
• JSON Corruption: Logs error and returns null on deserialization failure
• Schema/Table Creation Failures: Throws exception if AutoCreateTable=true and creation fails
• Concurrent Updates: Uses PostgreSQL's ON CONFLICT for safe concurrent updates

Thread Safety

The PostgresStore is thread-safe:

• Singleton initialization uses SemaphoreSlim with double-check locking
• Each operation uses its own connection (no shared state)
• Table creation is guarded by initialization lock

Performance

• Read Operations: Single SELECT query with primary key lookup
• Write Operations: Single UPSERT with ON CONFLICT
• Connection Pooling: Leverages Npgsql's built-in connection pooling
• No Long-Lived Connections: Each operation opens and closes connections

License

See the LICENSE file in the repository root.