Implement PostgreSQL event store for feature parity with SQL Server (#47)

Author: Copilot

src/BbQ.Events.PostgreSql/BbQ.Events.PostgreSql.csproj

@@ -7,8 +7,8 @@
     <Version>0.1.0</Version>
     <Authors>Jean</Authors>
     <Company>JM Mbouma</Company>
-    <Description>PostgreSQL implementation for BbQ.Events, providing IProjectionCheckpointStore for projection checkpoints. Features durable checkpoint persistence with upsert semantics, atomic operations, and thread-safe parallel processing support.</Description>
-    <PackageTags>events postgresql projections checkpoint persistence bbq cqrs</PackageTags>
+    <Description>PostgreSQL implementation for BbQ.Events, providing both IEventStore for event sourcing and IProjectionCheckpointStore for projection checkpoints. Features durable event persistence with sequential positions, atomic operations, JSON serialization, and thread-safe parallel processing support.</Description>
+    <PackageTags>events postgresql event-sourcing projections checkpoint persistence event-store bbq cqrs</PackageTags>
     <RepositoryUrl>https://github.com/JeanMarcMbouma/Outcome</RepositoryUrl>
     <PackageProjectUrl>https://github.com/JeanMarcMbouma/Outcome</PackageProjectUrl>
     <RepositoryType>git</RepositoryType>

src/BbQ.Events.PostgreSql/Configuration/ServiceCollectionExtensions.cs

@@ -1,7 +1,9 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using BbQ.Events.Checkpointing;
+using BbQ.Events.Events;
 using BbQ.Events.PostgreSql.Checkpointing;
+using BbQ.Events.PostgreSql.Events;
 
 namespace BbQ.Events.PostgreSql.Configuration;
 
@@ -46,4 +48,85 @@ public static IServiceCollection UsePostgreSqlCheckpoints(
 
         return services;
     }
+
+    /// <summary>
+    /// Registers the PostgreSQL event store for event sourcing.
+    /// </summary>
+    /// <param name="services">The service collection to register with</param>
+    /// <param name="connectionString">The PostgreSQL connection string</param>
+    /// <returns>The service collection for chaining</returns>
+    /// <remarks>
+    /// This method registers IEventStore as a singleton using PostgreSqlEventStore.
+    /// 
+    /// Prerequisites:
+    /// - PostgreSQL database must be accessible
+    /// - bbq_events table must be created (see Schema/CreateEventsTable.sql)
+    /// - bbq_streams table must be created (see Schema/CreateStreamsTable.sql)
+    /// 
+    /// Example usage:
+    /// <code>
+    /// services.UsePostgreSqlEventStore("Host=localhost;Database=mydb;Username=myuser;Password=mypass");
+    /// </code>
+    /// </remarks>
+    public static IServiceCollection UsePostgreSqlEventStore(
+        this IServiceCollection services, 
+        string connectionString)
+    {
+        if (string.IsNullOrWhiteSpace(connectionString))
+        {
+            throw new ArgumentNullException(nameof(connectionString));
+        }
+
+        return services.UsePostgreSqlEventStore(options =>
+        {
+            options.ConnectionString = connectionString;
+        });
+    }
+
+    /// <summary>
+    /// Registers the PostgreSQL event store for event sourcing with custom options.
+    /// </summary>
+    /// <param name="services">The service collection to register with</param>
+    /// <param name="configureOptions">Action to configure event store options</param>
+    /// <returns>The service collection for chaining</returns>
+    /// <remarks>
+    /// This method registers IEventStore as a singleton using PostgreSqlEventStore.
+    /// 
+    /// Prerequisites:
+    /// - PostgreSQL database must be accessible
+    /// - bbq_events table must be created (see Schema/CreateEventsTable.sql)
+    /// - bbq_streams table must be created (see Schema/CreateStreamsTable.sql)
+    /// 
+    /// Example usage:
+    /// <code>
+    /// services.UsePostgreSqlEventStore(options =>
+    /// {
+    ///     options.ConnectionString = "Host=localhost;Database=mydb;Username=myuser;Password=mypass";
+    ///     options.IncludeMetadata = true;
+    /// });
+    /// </code>
+    /// </remarks>
+    public static IServiceCollection UsePostgreSqlEventStore(
+        this IServiceCollection services,
+        Action<PostgreSqlEventStoreOptions> configureOptions)
+    {
+        if (configureOptions == null)
+        {
+            throw new ArgumentNullException(nameof(configureOptions));
+        }
+
+        var options = new PostgreSqlEventStoreOptions();
+        configureOptions(options);
+
+        if (string.IsNullOrWhiteSpace(options.ConnectionString))
+        {
+            throw new ArgumentException("Connection string must be configured via the configureOptions action.", nameof(configureOptions));
+        }
+
+        // Replace any existing IEventStore registration
+        services.Replace(ServiceDescriptor.Singleton<IEventStore>(
+            _ => new PostgreSqlEventStore(options)));
+
+        return services;
+    }
 }

src/BbQ.Events.PostgreSql/Events/PostgreSqlEventStore.cs

@@ -0,0 +1,185 @@
+using System.Runtime.CompilerServices;
+using System.Text.Json;
+using BbQ.Events.Events;
+using BbQ.Events.PostgreSql.Internal;
+using Npgsql;
+
+namespace BbQ.Events.PostgreSql.Events;
+
+/// <summary>
+/// PostgreSQL implementation of IEventStore.
+/// </summary>
+/// <remarks>
+/// This implementation provides:
+/// - Durable event persistence in PostgreSQL
+/// - Sequential position tracking per stream
+/// - Atomic append operations
+/// - Efficient event replay via streaming reads
+/// - JSON serialization of event data
+/// - Support for event metadata
+/// 
+/// Connection handling:
+/// - Each operation opens a new connection (connection pooling is handled by Npgsql)
+/// - Operations are fully async for optimal scalability
+/// - Connections are properly disposed in all code paths
+/// 
+/// Prerequisites:
+/// - bbq_events table must exist (see Schema/CreateEventsTable.sql)
+/// - bbq_streams table must exist (see Schema/CreateStreamsTable.sql)
+/// </remarks>
+public class PostgreSqlEventStore : IEventStore
+{
+    private readonly PostgreSqlEventStoreOptions _options;
+    private readonly JsonSerializerOptions _jsonOptions;
+    private static readonly string MachineName = Environment.MachineName;
+
+    /// <summary>
+    /// Creates a new PostgreSQL event store.
+    /// </summary>
+    /// <param name="options">Configuration options</param>
+    /// <exception cref="ArgumentNullException">Thrown when options is null</exception>
+    /// <exception cref="ArgumentException">Thrown when connection string is null or empty</exception>
+    public PostgreSqlEventStore(PostgreSqlEventStoreOptions options)
+    {
+        _options = options ?? throw new ArgumentNullException(nameof(options));
+        
+        if (string.IsNullOrWhiteSpace(_options.ConnectionString))
+        {
+            throw new ArgumentException("Connection string cannot be null or empty", nameof(options));
+        }
+
+        _jsonOptions = _options.JsonSerializerOptions ?? new JsonSerializerOptions
+        {
+            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
+            WriteIndented = false
+        };
+    }
+
+    /// <summary>
+    /// Appends an event to a stream.
+    /// </summary>
+    /// <typeparam name="TEvent">The type of event</typeparam>
+    /// <param name="stream">The stream name</param>
+    /// <param name="event">The event to append</param>
+    /// <param name="ct">Cancellation token</param>
+    /// <returns>The position of the appended event</returns>
+    /// <exception cref="ArgumentException">Thrown when stream name is null or empty</exception>
+    /// <exception cref="ArgumentNullException">Thrown when event is null</exception>
+    public async Task<long> AppendAsync<TEvent>(string stream, TEvent @event, CancellationToken ct = default)
+    {
+        if (string.IsNullOrWhiteSpace(stream))
+        {
+            throw new ArgumentException("Stream name cannot be null or empty", nameof(stream));
+        }
+
+        if (@event == null)
+        {
+            throw new ArgumentNullException(nameof(@event));
+        }
+
+        await using var connection = new NpgsqlConnection(_options.ConnectionString);
+        await connection.OpenAsync(ct);
+
+        await using var command = connection.CreateCommand();
+        command.CommandText = PostgreSqlConstants.AppendEventSqlSimplified;
+
+        var eventType = typeof(TEvent).FullName ?? typeof(TEvent).Name;
+        var eventData = PostgreSqlHelpers.SerializeToJson(@event, _jsonOptions);
+        
+        command.AddParameter("@stream_name", stream);
+        command.AddParameter("@event_type", eventType);
+        command.AddParameter("@event_data", eventData);
+        command.AddParameter("@metadata", _options.IncludeMetadata ? CreateMetadata() : null);
+
+        var result = await command.ExecuteScalarAsync(ct);
+        return Convert.ToInt64(result);
+    }
+
+    /// <summary>
+    /// Reads events from a stream starting at a given position.
+    /// </summary>
+    /// <typeparam name="TEvent">The type of events to read</typeparam>
+    /// <param name="stream">The stream name</param>
+    /// <param name="fromPosition">The position to start reading from (inclusive)</param>
+    /// <param name="ct">Cancellation token</param>
+    /// <returns>An async enumerable of events with their positions</returns>
+    /// <exception cref="ArgumentException">Thrown when stream name is null or empty</exception>
+    public async IAsyncEnumerable<StoredEvent<TEvent>> ReadAsync<TEvent>(
+        string stream, 
+        long fromPosition = 0,
+        [EnumeratorCancellation] CancellationToken ct = default)
+    {
+        if (string.IsNullOrWhiteSpace(stream))
+        {
+            throw new ArgumentException("Stream name cannot be null or empty", nameof(stream));
+        }
+
+        await using var connection = new NpgsqlConnection(_options.ConnectionString);
+        await connection.OpenAsync(ct);
+
+        await using var command = connection.CreateCommand();
+        command.CommandText = PostgreSqlConstants.ReadEventsSql;
+        command.AddParameter("@stream_name", stream);
+        command.AddParameter("@from_position", fromPosition);
+
+        await using var reader = await command.ExecuteReaderAsync(ct);
+
+        while (await reader.ReadAsync(ct))
+        {
+            var position = reader.GetLong(PostgreSqlConstants.Position);
+            var eventType = reader.GetString(reader.GetOrdinal(PostgreSqlConstants.EventType));
+            var eventData = reader.GetString(reader.GetOrdinal(PostgreSqlConstants.EventData));
+
+            // Only deserialize if the event type matches
+            // This allows for type filtering when reading from streams with multiple event types
+            var expectedType = typeof(TEvent).FullName ?? typeof(TEvent).Name;
+            if (eventType == expectedType)
+            {
+                var @event = PostgreSqlHelpers.DeserializeFromJson<TEvent>(eventData, _jsonOptions);
+                yield return new StoredEvent<TEvent>(position, @event);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Gets the current position (last event position) in a stream.
+    /// </summary>
+    /// <param name="stream">The stream name</param>
+    /// <param name="ct">Cancellation token</param>
+    /// <returns>The current position, or null if the stream doesn't exist</returns>
+    /// <exception cref="ArgumentException">Thrown when stream name is null or empty</exception>
+    public async Task<long?> GetStreamPositionAsync(string stream, CancellationToken ct = default)
+    {
+        if (string.IsNullOrWhiteSpace(stream))
+        {
+            throw new ArgumentException("Stream name cannot be null or empty", nameof(stream));
+        }
+
+        await using var connection = new NpgsqlConnection(_options.ConnectionString);
+        await connection.OpenAsync(ct);
+
+        await using var command = connection.CreateCommand();
+        command.CommandText = PostgreSqlConstants.GetStreamPositionSql;
+        command.AddParameter("@stream_name", stream);
+
+        var result = await command.ExecuteScalarAsync(ct);
+        
+        return result == null || result == DBNull.Value 
+            ? null 
+            : Convert.ToInt64(result);
+    }
+
+    /// <summary>
+    /// Creates metadata for an event.
+    /// </summary>
+    private string CreateMetadata()
+    {
+        var metadata = new Dictionary<string, object>
+        {
+            ["timestamp"] = DateTime.UtcNow,
+            ["server"] = MachineName
+        };
+
+        return PostgreSqlHelpers.SerializeToJson(metadata, _jsonOptions);
+    }
+}

src/BbQ.Events.PostgreSql/Events/PostgreSqlEventStoreOptions.cs

@@ -0,0 +1,31 @@
+using System.Text.Json;
+
+namespace BbQ.Events.PostgreSql.Events;
+
+/// <summary>
+/// Configuration options for PostgreSQL event store.
+/// </summary>
+public class PostgreSqlEventStoreOptions
+{
+    /// <summary>
+    /// Gets or sets the connection string for PostgreSQL.
+    /// </summary>
+    public string ConnectionString { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the JSON serializer options for event data.
+    /// </summary>
+    /// <remarks>
+    /// If not provided, defaults to camelCase property naming.
+    /// </remarks>
+    public JsonSerializerOptions? JsonSerializerOptions { get; set; }
+
+    /// <summary>
+    /// Gets or sets whether to include metadata in stored events.
+    /// </summary>
+    /// <remarks>
+    /// Metadata can include correlation IDs, causation IDs, timestamps, etc.
+    /// Default is false.
+    /// </remarks>
+    public bool IncludeMetadata { get; set; } = false;
+}

src/BbQ.Events.PostgreSql/Internal/PostgreSqlConstants.cs

@@ -0,0 +1,62 @@
+namespace BbQ.Events.PostgreSql.Internal;
+
+/// <summary>
+/// PostgreSQL constants for BbQ.Events.
+/// </summary>
+internal static class PostgreSqlConstants
+{
+    // Table names
+    public const string EventsTable = "bbq_events";
+    public const string StreamsTable = "bbq_streams";
+    public const string CheckpointsTable = "bbq_projection_checkpoints";
+
+    // Column names - Events
+    public const string EventId = "event_id";
+    public const string StreamName = "stream_name";
+    public const string Position = "position";
+    public const string EventType = "event_type";
+    public const string EventData = "event_data";
+    public const string Metadata = "metadata";
+    public const string CreatedUtc = "created_utc";
+
+    // Column names - Streams
+    public const string CurrentPosition = "current_position";
+    public const string Version = "version";
+    public const string LastUpdatedUtc = "last_updated_utc";
+
+    // Column names - Checkpoints
+    public const string ProjectionName = "projection_name";
+    public const string PartitionKey = "partition_key";
+
+    // SQL queries - Events
+    // Uses CTE (Common Table Expression) for atomic upsert and event insertion
+    public const string AppendEventSqlSimplified = @"
+        WITH updated_stream AS (
+            INSERT INTO bbq_streams (stream_name, current_position, version, created_utc, last_updated_utc)
+            VALUES (@stream_name, 0, 1, (NOW() AT TIME ZONE 'UTC'), (NOW() AT TIME ZONE 'UTC'))
+            ON CONFLICT (stream_name)
+            DO UPDATE SET 
+                current_position = bbq_streams.current_position + 1,
+                version = bbq_streams.version + 1,
+                last_updated_utc = (NOW() AT TIME ZONE 'UTC')
+            RETURNING current_position
+        ),
+        inserted_event AS (
+            INSERT INTO bbq_events (stream_name, position, event_type, event_data, metadata, created_utc)
+            SELECT @stream_name, current_position, @event_type, @event_data, @metadata, (NOW() AT TIME ZONE 'UTC')
+            FROM updated_stream
+            RETURNING position
+        )
+        SELECT position FROM inserted_event;";
+
+    public const string ReadEventsSql = @"
+        SELECT event_id, stream_name, position, event_type, event_data, metadata, created_utc
+        FROM bbq_events
+        WHERE stream_name = @stream_name AND position >= @from_position
+        ORDER BY position";
+
+    public const string GetStreamPositionSql = @"
+        SELECT current_position 
+        FROM bbq_streams 
+        WHERE stream_name = @stream_name";
+}