Nethereum.Mud.Repositories.Postgres 6.0.1

Nethereum.Mud.Repositories.Postgres

Nethereum.Mud.Repositories.Postgres provides a production-ready PostgreSQL implementation for persisting and querying MUD (Onchain Engine) table data. It includes database schema normalization, allowing you to automatically create typed database tables from MUD schemas.

Features

• PostgreSQL Optimized DbContext - Uses bytea for binary data instead of hex strings
• MUD Schema Normalizer - Automatically creates PostgreSQL tables from MUD schemas
• Binary Key Storage - Stores keys as bytea for efficient indexing and queries
• Background Sync Service - Process Store events and sync to PostgreSQL
• Chain State Validation - Ensures chainId matches stored state to prevent cross-chain corruption
• Reorg Buffer Support - Rewinds progress by N blocks to handle reorganizations
• Composite Indexes - Optimized indexes for Address + TableId + Key queries
• Typed Table Creation - Generate normalized tables with proper column types
• Snake Case Naming - Lowercase naming convention for PostgreSQL best practices
• Singleton Table Support - Handles tables without keys (configuration singletons)

Installation

dotnet add package Nethereum.Mud.Repositories.Postgres

Dependencies

• Npgsql.EntityFrameworkCore.PostgreSQL 8.0+
• EFCore.NamingConventions 8.0+
• Nethereum.Mud.Repositories.EntityFramework

Key Concepts

MudPostgresStoreRecordsDbContext

PostgreSQL-optimized DbContext with:

• bytea columns for binary data (TableId, Address, Key, StaticData, DynamicData, EncodedLengths)
• Composite primary key: (AddressBytes, TableIdBytes, KeyBytes)
• Indexes on individual keys (Key0Bytes, Key1Bytes, Key2Bytes, Key3Bytes)
• Lower case naming convention via UseLowerCaseNamingConvention()

MudPostgresStoreRecordsTableRepository

Extends MudEFTableRepository with PostgreSQL-specific binary operations:

• Queries using bytea comparisons (faster than hex string comparisons)
• Binary predicate builder for complex queries
• Batch processing with AsNoTracking optimization

MudPostgresStoreRecordsProcessingService

Background service for syncing MUD Store events to PostgreSQL:

• Processes Store_SetRecord, Store_DeleteRecord, Store_SpliceStaticData, Store_SpliceDynamicData events
• Tracks block progress to resume from last synced block
• Configurable batch size and retry logic
• Minimum block confirmations support
• ChainId validation against stored ChainStates to prevent cross-chain DB misuse
• Optional reorg buffer for safe log replays

MudPostgresStoreRecordsNormaliser

Converts raw StoredRecord entries into typed PostgreSQL tables:

• Reads MUD schema from on-chain Tables table
• Creates PostgreSQL tables dynamically with proper types
• Handles singleton tables (no keys) with auto-incrementing id
• Converts ABI types to PostgreSQL types (uint256 → NUMERIC, address → TEXT, etc.)
• Upserts records with conflict resolution

Usage Examples

Example 1: Setup PostgreSQL DbContext

Configure connection string and create DbContext:

using Microsoft.EntityFrameworkCore;
using Nethereum.Mud.Repositories.Postgres;

var connectionString = "Host=localhost;Database=muddb;Username=postgres;Password=password";

var optionsBuilder = new DbContextOptionsBuilder<MudPostgresStoreRecordsDbContext>();
optionsBuilder.UseNpgsql(connectionString);

using var context = new MudPostgresStoreRecordsDbContext(optionsBuilder.Options);

// Create database and tables
await context.Database.EnsureCreatedAsync();

Console.WriteLine("PostgreSQL database initialized");

Example 2: Apply Migrations

Generate and apply Entity Framework migrations:

# Add initial migration
dotnet ef migrations add InitialMudPostgres --context MudPostgresStoreRecordsDbContext

# Update database
dotnet ef database update --context MudPostgresStoreRecordsDbContext

Or run migrations in code:

using (var context = new MudPostgresStoreRecordsDbContext(optionsBuilder.Options))
{
    await context.Database.MigrateAsync();
    Console.WriteLine("Migrations applied successfully");
}

If you are upgrading an existing database without migrations, apply the chain state script:

-- src/Nethereum.Mud.Repositories.Postgres/Sql/MudPostgresStoreRecordsDbContext.ChainStates.sql

Example 3: Background Sync Service

Use the processing service to sync MUD events:

using Microsoft.Extensions.Logging;
using Nethereum.Mud.Repositories.Postgres;

var logger = LoggerFactory.Create(builder => builder.AddConsole()).CreateLogger("MudSync");

using var context = new MudPostgresStoreRecordsDbContext(optionsBuilder.Options);

var processingService = new MudPostgresStoreRecordsProcessingService(context, logger)
{
    Address = "0xWorldContractAddress",
    RpcUrl = "https://rpc.mud.game",
    StartAtBlockNumberIfNotProcessed = 0,
    NumberOfBlocksToProcessPerRequest = 1000,
    RetryWeight = 50,
    MinimumNumberOfConfirmations = 12,
    ReorgBuffer = 25
};

// Start syncing (blocks until cancelled)
await processingService.ExecuteAsync(CancellationToken.None);

Example 4: Query StoredRecords by Binary Keys

Query using bytea for performance:

using Nethereum.Hex.HexConvertors.Extensions;
using Microsoft.EntityFrameworkCore;

var tableIdBytes = "0x7462000000000000000000000000000000000000000000000000000000000000".HexToByteArray();
var keyBytes = "0x000000000000000000000000000000000000000000000000000000000000002a".HexToByteArray();

using var context = new MudPostgresStoreRecordsDbContext(optionsBuilder.Options);

var record = await context.StoredRecords
    .AsNoTracking()
    .FirstOrDefaultAsync(r =>
        r.TableIdBytes == tableIdBytes &&
        r.KeyBytes == keyBytes &&
        !r.IsDeleted
    );

if (record != null)
{
    Console.WriteLine($"Found record at block {record.BlockNumber}");
    Console.WriteLine($"Static data length: {record.StaticData?.Length ?? 0}");
    Console.WriteLine($"Dynamic data length: {record.DynamicData?.Length ?? 0}");
}

Example 5: Query with TablePredicate

Use predicates for complex queries:

using Nethereum.Mud.TableRepository;
using Nethereum.Mud.Repositories.Postgres;

var repository = new MudPostgresStoreRecordsTableRepository(context);

// Build predicate
var predicate = new TablePredicate
{
    Conditions = new List<TableCondition>
    {
        new TableCondition
        {
            TableId = "0x7462...",
            Address = "0xWorldAddress",
            Key = "key0",
            ComparisonOperator = ">",
            HexValue = "0x0000000000000000000000000000000000000000000000000000000000000000",
            UnionOperator = "AND"
        }
    }
};

var records = await repository.GetRecordsAsync(predicate);
Console.WriteLine($"Found {records.Count} records matching predicate");

Example 6: Convert to Typed TableRecords

Retrieve strongly-typed MUD table records:

using Nethereum.Mud;
using Nethereum.Mud.TableRepository;

// Assume PlayerTableRecord is a generated MUD table record
var playerTableResource = new Resource("Game", "Player");
var tableIdHex = playerTableResource.ResourceIdEncoded.ToHex(true);

var repository = new MudPostgresStoreRecordsTableRepository(context);
var playerRecords = await repository.GetTableRecordsAsync<PlayerTableRecord>(tableIdHex);

foreach (var player in playerRecords)
{
    Console.WriteLine($"Player ID: {player.Keys.PlayerId}");
    Console.WriteLine($"Name: {player.Values.Name}");
    Console.WriteLine($"Level: {player.Values.Level}");
}

Example 7: Schema Normalization - Create Typed Tables

Use the normalizer to create typed PostgreSQL tables:

using Npgsql;
using Nethereum.Mud.Repositories.Postgres.StoreRecordsNormaliser;
using Nethereum.Mud.Contracts.Store;
using Nethereum.Web3;
using Microsoft.Extensions.Logging;

var logger = LoggerFactory.Create(builder => builder.AddConsole()).CreateLogger("Normalizer");

// Connect to PostgreSQL
var connectionString = "Host=localhost;Database=muddb;Username=postgres;Password=password";
using var connection = new NpgsqlConnection(connectionString);

// Connect to MUD World for schema retrieval
var web3 = new Web3("https://rpc.mud.game");
var worldAddress = "0xWorldAddress";
var storeNamespace = new StoreNamespace(web3, worldAddress);

// Create normalizer (worldAddress is optional, used to prefix table names)
var normalizer = new MudPostgresStoreRecordsNormaliser(connection, storeNamespace, logger, worldAddress);

// Get table schema from on-chain
var playerTableId = new Resource("Game", "Player").ResourceIdEncoded;
var schema = await normalizer.GetTableSchemaAsync(playerTableId);

Console.WriteLine($"Table: {schema.Namespace}:{schema.Name}");
Console.WriteLine($"Keys: {string.Join(", ", schema.SchemaKeys.Select(k => k.Name))}");
Console.WriteLine($"Values: {string.Join(", ", schema.SchemaValues.Select(v => v.Name))}");

// Table is automatically created in PostgreSQL
// Example: "game_player" table with columns (playerid NUMERIC, name TEXT, level SMALLINT, health INTEGER, ...)

Example 8: Normalize Records to Typed Tables

Convert StoredRecords to typed table rows:

using Nethereum.Mud.EncodingDecoding;

// Get StoredRecord from database
var storedRecord = await context.StoredRecords
    .FirstOrDefaultAsync(r => r.TableIdBytes == playerTableId && !r.IsDeleted);

if (storedRecord != null)
{
    // Convert to EncodedTableRecord
    var encodedTableRecord = new EncodedTableRecord
    {
        TableId = storedRecord.TableIdBytes,
        Key = storedRecord.KeyBytes.SplitBytes(),
        EncodedValues = storedRecord
    };

    // Upsert to normalized table (e.g., "game_player")
    await normalizer.UpsertRecordAsync(encodedTableRecord);

    Console.WriteLine("Record normalized to typed table");
}

Example 9: Continuous Normalization Service

Background service to continuously normalize records:

using Nethereum.Mud.Repositories.Postgres.StoreRecordsNormaliser;

var repository = new MudPostgresStoreRecordsTableRepository(context);

var normaliserProcessingService = new MudPostgresNormaliserProcessingService(
    repository,
    connection,
    logger
)
{
    RpcUrl = "https://rpc.mud.game",
    Address = "0xWorldAddress",
    PageSize = 100
};

// Process all stored records in batches
await normaliserProcessingService.ExecuteAsync(CancellationToken.None);

Console.WriteLine("All records normalized to typed tables");

Example 10: Production Sync + Normalize Pipeline

Complete pipeline: sync events → store raw records → normalize to typed tables:

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

public class MudPipelineBackgroundService : BackgroundService
{
    private readonly ILogger<MudPipelineBackgroundService> _logger;
    private readonly MudPostgresStoreRecordsDbContext _context;
    private readonly string _rpcUrl;
    private readonly string _worldAddress;
    private readonly string _postgresConnectionString;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Starting MUD Pipeline: Sync + Normalize");

        // Task 1: Sync Store events to StoredRecords table
        var syncTask = Task.Run(async () =>
        {
            var syncService = new MudPostgresStoreRecordsProcessingService(_context, _logger)
            {
                Address = _worldAddress,
                RpcUrl = _rpcUrl,
                StartAtBlockNumberIfNotProcessed = 0,
                NumberOfBlocksToProcessPerRequest = 1000
            };

            await syncService.ExecuteAsync(stoppingToken);
        }, stoppingToken);

        // Task 2: Normalize StoredRecords to typed tables
        var normalizeTask = Task.Run(async () =>
        {
            await Task.Delay(10000, stoppingToken); // Wait for initial sync

            using var connection = new NpgsqlConnection(_postgresConnectionString);
            var repository = new MudPostgresStoreRecordsTableRepository(_context);
            var normaliserService = new MudPostgresNormaliserProcessingService(
                repository,
                connection,
                _logger
            )
            {
                RpcUrl = _rpcUrl,
                Address = _worldAddress
            };

            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    await normaliserService.ExecuteAsync(stoppingToken);
                    await Task.Delay(5000, stoppingToken); // Check for new records every 5s
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Normalization error");
                    await Task.Delay(30000, stoppingToken);
                }
            }
        }, stoppingToken);

        await Task.WhenAll(syncTask, normalizeTask);
    }
}

// Register in Program.cs
services.AddDbContext<MudPostgresStoreRecordsDbContext>(options =>
    options.UseNpgsql(postgresConnectionString));

services.AddHostedService<MudPipelineBackgroundService>();

Core Classes

MudPostgresStoreRecordsDbContext

public class MudPostgresStoreRecordsDbContext : DbContext, IMudStoreRecordsDbSets
{
    public DbSet<StoredRecord> StoredRecords { get; set; }
    public DbSet<BlockProgress> BlockProgress { get; set; }
    public DbSet<ChainState> ChainStates { get; set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        // Composite primary key using bytea
        modelBuilder.Entity<StoredRecord>()
            .HasKey(r => new { r.AddressBytes, r.TableIdBytes, r.KeyBytes });

        // Indexes for queries
        modelBuilder.Entity<StoredRecord>()
            .HasIndex(r => new { r.AddressBytes, r.TableIdBytes, r.Key0Bytes });

        // Map to bytea
        modelBuilder.Entity<StoredRecord>()
            .Property(e => e.TableIdBytes)
            .HasColumnName("tableid")
            .HasColumnType("bytea");

        // BigInteger to numeric conversion
        modelBuilder.Entity<StoredRecord>()
            .Property(r => r.BlockNumber)
            .HasConversion(
                v => (decimal)v,
                v => (BigInteger)v
            )
            .HasColumnType("numeric(1000, 0)");
    }
}

MudPostgresStoreRecordsTableRepository

public class MudPostgresStoreRecordsTableRepository : MudEFTableRepository<MudPostgresStoreRecordsDbContext>
{
    // Query using bytea
    public override async Task<StoredRecord> GetRecordAsync(string tableIdHex, string keyHex)
    {
        var tableIdBytes = tableIdHex.HexToByteArray();
        var keyBytes = keyHex.HexToByteArray();

        return await Context.StoredRecords
            .AsNoTracking()
            .FirstOrDefaultAsync(r => r.TableIdBytes == tableIdBytes && r.KeyBytes == keyBytes);
    }

    // SQL predicate using bytea
    public override Task<List<StoredRecord>> GetRecordsAsync(TablePredicate predicate)
    {
        var builder = new MudPostgresStoreRecordsSqlByteaPredicateBuilder();
        var sqlPredicate = builder.BuildSql(predicate);
        string sqlQuery = $"SELECT * FROM storedrecords WHERE {sqlPredicate.Sql}";
        return Context.StoredRecords.FromSqlRaw(sqlQuery, sqlPredicate.GetParameterValues()).ToListAsync();
    }
}

MudPostgresStoreRecordsProcessingService

public class MudPostgresStoreRecordsProcessingService
{
    public string Address { get; set; }                     // World contract address
    public string RpcUrl { get; set; }                      // Ethereum RPC URL
    public BigInteger StartAtBlockNumberIfNotProcessed { get; set; } = 0;
    public int NumberOfBlocksToProcessPerRequest { get; set; } = 1000;
    public int RetryWeight { get; set; } = 50;
    public uint MinimumNumberOfConfirmations { get; set; } = 0;
    public int ReorgBuffer { get; set; } = 0;

    public async Task ExecuteAsync(CancellationToken cancellationToken = default);
}

MudPostgresStoreRecordsNormaliser

public class MudPostgresStoreRecordsNormaliser
{
    // Retrieve and cache MUD schema
    public async Task<TableSchema> GetTableSchemaAsync(byte[] tableId);

    // Create PostgreSQL table from MUD schema
    public async Task CreateTableIfNotExistsAsync(TableSchema schema);

    // Upsert record to normalized table
    public async Task UpsertRecordAsync(EncodedTableRecord encodedTableRecord);
    public async Task UpsertRecordAsync(TableSchema schema, List<FieldValue> fieldValues);

    // Delete record from normalized table
    public async Task DeleteRecordAsync(EncodedTableRecord encodedTableRecord);
    public async Task DeleteRecordAsync(TableSchema schema, List<FieldValue> fieldValues);
}

NormalisedTableQueryService

Query normalized tables with filtering, pagination, and schema introspection:

public class NormalisedTableQueryService
{
    public NormalisedTableQueryService(NpgsqlConnection connection, ILogger<NormalisedTableQueryService> logger = null);

    // Query rows with optional predicate, pagination
    public Task<List<Dictionary<string, object>>> QueryAsync(
        string tableName, TablePredicate predicate = null, int pageSize = 100, int page = 0);

    // Count rows matching a predicate
    public Task<int> CountAsync(string tableName, TablePredicate predicate = null);

    // List all normalized tables (filtered by prefix, default "w_")
    public Task<List<string>> GetAvailableTablesAsync(string prefix = "w_");

    // Get column metadata for a normalized table
    public Task<List<NormalisedColumnInfo>> GetTableColumnsAsync(string tableName);
}

Usage example:

using Nethereum.Mud.Repositories.Postgres.StoreRecordsNormaliser;

var queryService = new NormalisedTableQueryService(connection, logger);

// List all normalized tables
var tables = await queryService.GetAvailableTablesAsync("w_");

// Inspect columns
var columns = await queryService.GetTableColumnsAsync("w_abc123_game_player");

// Query with predicate
var predicate = new TablePredicate();
predicate.Conditions.Add(new KeyValueOperator
{
    Name = "level",
    IsValueField = true,
    ComparisonOperator = ">",
    RawValue = 10,
    AbiType = "uint8"
});

var players = await queryService.QueryAsync("w_abc123_game_player", predicate, pageSize: 50, page: 0);
var totalCount = await queryService.CountAsync("w_abc123_game_player", predicate);

Type Conversion Mapping

MUD ABI types are mapped to PostgreSQL types:

Advanced Topics

PostgreSQL Performance Optimization

1. Use bytea for keys and binary data:

// Fast: Binary comparison
r.TableIdBytes == tableIdBytes

// Slow: String comparison
r.TableId == tableIdHex

2. Leverage composite indexes:

CREATE INDEX ix_address_tableid_key0 ON storedrecords (addressbytes, tableidbytes, key0bytes);

3. Use NUMERIC for BigInteger:

// Handles uint256 values without overflow
modelBuilder.Entity<StoredRecord>()
    .Property(r => r.BlockNumber)
    .HasColumnType("numeric(1000, 0)");

Handling Singleton Tables

Singleton tables (no keys) are handled with auto-incrementing id:

-- Generated for singleton table
CREATE TABLE config (
    id SERIAL PRIMARY KEY,
    maxplayers NUMERIC(78, 0),
    ispaused BOOLEAN
);

-- Upsert always updates id=1
INSERT INTO config (id, maxplayers, ispaused)
VALUES (1, 100, false)
ON CONFLICT (id)
DO UPDATE SET maxplayers = 100, ispaused = false;

Custom Table Schemas

Override GetTableName to customize table naming:

public class CustomNormaliser : MudPostgresStoreRecordsNormaliser
{
    protected override string GetTableName(TableSchema schema)
    {
        // Custom naming: prefix all tables with "mud_"
        return $"mud_{base.GetTableName(schema)}";
    }
}

Query Normalized Tables Directly

Once normalized, query typed tables directly with SQL:

-- Query game_player table
SELECT playerid, name, level, health, experience
FROM game_player
WHERE level > 10
ORDER BY experience DESC;

-- Join multiple MUD tables
SELECT p.playerid, p.name, i.itemid, i.quantity
FROM game_player p
JOIN game_inventory i ON p.playerid = i.playerid
WHERE p.level > 5;

Production Patterns

1. Two-Phase Sync: Raw + Normalized

Store raw records first, then normalize asynchronously:

// Phase 1: Fast sync to StoredRecords (critical path)
var syncService = new MudPostgresStoreRecordsProcessingService(context, logger);
await syncService.ExecuteAsync(cancellationToken);

// Phase 2: Normalize to typed tables (background)
var normalizeService = new MudPostgresNormaliserProcessingService(...);
await normalizeService.ExecuteAsync(cancellationToken);

2. Read from Typed Tables, Write to StoredRecords

• Writes: Always go through MUD contracts → Store events → StoredRecords
• Reads: Query normalized typed tables for fast SQL queries

// Write: Via MUD contracts
await worldService.CallFromRequestAndWaitForReceiptAsync(systemId, callData);

// Read: From normalized table
var players = await connection.QueryAsync<PlayerInfo>(
    "SELECT * FROM game_player WHERE level > @minLevel",
    new { minLevel = 10 }
);

3. Horizontal Scaling with Read Replicas

// Primary: Writes to StoredRecords
var primaryConnectionString = "Host=primary.db;Database=mud;...";

// Replica: Read-only queries on typed tables
var replicaConnectionString = "Host=replica.db;Database=mud;...";

// Use replica for queries
using var replicaConnection = new NpgsqlConnection(replicaConnectionString);
var players = await replicaConnection.QueryAsync<PlayerInfo>(...);

4. Connection Pooling Configuration

var connectionString = "Host=localhost;Database=muddb;Username=postgres;Password=password;Pooling=true;MinPoolSize=5;MaxPoolSize=100;ConnectionLifetime=300";

services.AddDbContext<MudPostgresStoreRecordsDbContext>(options =>
    options.UseNpgsql(connectionString));

Database Schema

StoredRecords Table (Raw Storage)

CREATE TABLE storedrecords (
    rowid BIGSERIAL,
    addressbytes BYTEA NOT NULL,
    tableidbytes BYTEA NOT NULL,
    keybytes BYTEA NOT NULL,
    key0bytes BYTEA,
    key1bytes BYTEA,
    key2bytes BYTEA,
    key3bytes BYTEA,
    static_data BYTEA,
    encoded_lengths BYTEA,
    dynamic_data BYTEA,
    isdeleted BOOLEAN NOT NULL DEFAULT false,
    blocknumber NUMERIC(1000, 0),
    logindex INTEGER,
    PRIMARY KEY (addressbytes, tableidbytes, keybytes)
);

CREATE INDEX ix_rowid ON storedrecords (rowid);
CREATE INDEX ix_address_tableid_key0 ON storedrecords (addressbytes, tableidbytes, key0bytes);

ChainStates Table (Chain Validation)

CREATE TABLE chainstates (
    rowindex BIGSERIAL PRIMARY KEY,
    lastcanonicalblocknumber TEXT,
    lastcanonicalblockhash TEXT,
    finalizedblocknumber TEXT,
    chainid INTEGER,
    creationdate TIMESTAMP,
    lastupdatedate TIMESTAMP
);

CREATE INDEX ix_chainstates_lastcanonicalblocknumber ON chainstates (lastcanonicalblocknumber);

Example Normalized Table

-- Generated from MUD Player table schema
CREATE TABLE game_player (
    playerid NUMERIC(78, 0),         -- uint256
    name TEXT,                       -- string
    level SMALLINT,                  -- uint8
    health INTEGER,                  -- uint16
    experience NUMERIC(78, 0),       -- uint256
    isactive BOOLEAN,                -- bool
    PRIMARY KEY (playerid)
);

Related Packages

Dependencies

• Nethereum.Mud.Repositories.EntityFramework - Base EF Core abstractions
• Nethereum.Mud.Contracts - Store contract interactions
• Nethereum.Mud - Core MUD types
• Npgsql.EntityFrameworkCore.PostgreSQL - PostgreSQL EF provider

Complementary

• Dapper - Can be used to query normalized tables directly
• Npgsql - Direct PostgreSQL access for advanced queries

Additional Resources

• MUD Documentation
• Npgsql Documentation
• PostgreSQL NUMERIC Type
• Nethereum MUD Console Tests

Support

• Nethereum Discord
• GitHub Issues