Cleanup

src/Contrib.KafkaFlow.Outbox.Postgres/ConfigurationBuilderExtensions.cs

@@ -5,5 +5,7 @@ namespace KafkaFlow.Outbox.Postgres;
 public static class ConfigurationBuilderExtensions
 {
     public static IServiceCollection AddPostgresOutboxBackend(this IServiceCollection services) =>
-        services.AddSingleton<IOutboxBackend, PostgresOutboxBackend>();
+        services
+            .AddSingleton<IOutboxRepository, PostgresOutboxRepository>()
+            .AddSingleton<IOutboxBackend, OutboxBackend>();
 }

src/Contrib.KafkaFlow.Outbox.Postgres/Contrib.KafkaFlow.Outbox.Postgres.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>

src/Contrib.KafkaFlow.Outbox.Postgres/PostgresOutboxRepository.cs

@@ -0,0 +1,50 @@
+using Dapper;
+using Npgsql;
+
+namespace KafkaFlow.Outbox.Postgres;
+
+public class PostgresOutboxRepository(NpgsqlDataSource connectionPool) : IOutboxRepository
+{
+    private readonly NpgsqlDataSource _connectionPool = connectionPool;
+
+    public async ValueTask Store(OutboxTableRow outboxTableRow, CancellationToken token = default)
+    {
+        var sql = """
+            INSERT INTO outbox.outbox(topic_name, partition, message_key, message_headers, message_body)
+            VALUES (@topic_name, @partition, @message_key, @message_headers, @message_body)
+            """;
+
+        await using var conn = _connectionPool.CreateConnection();
+        await conn.ExecuteAsync(sql, new
+        {
+            topic_name = outboxTableRow.TopicName,
+            partition = outboxTableRow.Partition,
+            message_key = outboxTableRow.MessageKey,
+            message_headers = outboxTableRow.MessageHeaders,
+            message_body = outboxTableRow.MessageBody
+        }).ConfigureAwait(false);
+    }
+
+    public async Task<IEnumerable<OutboxTableRow>> Read(int batchSize, CancellationToken token = default)
+    {
+        var sql = """
+            DELETE FROM outbox.outbox
+            WHERE
+                sequence_id = ANY(ARRAY(
+                    SELECT sequence_id FROM outbox.outbox
+                    ORDER BY sequence_id
+                    LIMIT @batch_size
+                    FOR UPDATE
+                ))
+            RETURNING
+                sequence_id as "SequenceId",
+                topic_name as "TopicName",
+                partition as "Partition",
+                message_key as "MessageKey",
+                message_headers as "MessageHeaders",
+                message_body as "MessageBody"
+            """;
+        await using var conn = _connectionPool.CreateConnection();
+        return await conn.QueryAsync<OutboxTableRow>(sql, new { batch_size = batchSize }).ConfigureAwait(false);
+    }
+}

src/Contrib.KafkaFlow.Outbox.SqlServer/ConfigurationBuilderExtensions.cs

@@ -7,7 +7,9 @@ namespace KafkaFlow.Outbox.SqlServer;
 public static class ConfigurationBuilderExtensions
 {
     public static IServiceCollection AddSqlServerOutboxBackend(this IServiceCollection services) =>
-        services.AddSingleton<IOutboxBackend, SqlServerOutboxBackend>();
+        services
+            .AddSingleton<IOutboxRepository, SqlServerOutboxRepository>()
+            .AddSingleton<IOutboxBackend, OutboxBackend>();
 
     public static IServiceCollection AddSqlServerOutboxBackend(this IServiceCollection services, string connectionString)
     {

src/Contrib.KafkaFlow.Outbox.SqlServer/Contrib.KafkaFlow.Outbox.SqlServer.csproj

@@ -17,7 +17,6 @@
   <ItemGroup>
     <PackageReference Include="Dapper" Version="2.1.35" />
     <PackageReference Include="System.Data.SqlClient" Version="4.8.6" />
-    <PackageReference Include="System.Text.Json" Version="8.0.3" />
   </ItemGroup>
 
 </Project>

src/Contrib.KafkaFlow.Outbox.SqlServer/SqlServerOutboxRepository.cs

@@ -0,0 +1,50 @@
+using Dapper;
+using KafkaFlow.SqlServer;
+using Microsoft.Extensions.Options;
+using System.Data.SqlClient;
+
+namespace KafkaFlow.Outbox.SqlServer;
+
+public class SqlServerOutboxRepository(IOptions<SqlServerBackendOptions> options) : IOutboxRepository
+{
+    private readonly SqlServerBackendOptions _options = options?.Value ?? throw new ArgumentNullException(nameof(options));
+
+    public async ValueTask Store(OutboxTableRow outboxTableRow, CancellationToken token = default)
+    {
+        var sql = """
+            INSERT INTO [outbox].[outbox] ([topic_name], [partition], [message_key], [message_headers], [message_body])
+            VALUES (@topic_name, @partition, @message_key, @message_headers, @message_body);
+            """;
+
+        using var conn = new SqlConnection(_options.ConnectionString);
+        await conn.ExecuteAsync(sql, new
+        {
+            topic_name = outboxTableRow.TopicName,
+            partition = outboxTableRow.Partition,
+            message_key = outboxTableRow.MessageKey,
+            message_headers = outboxTableRow.MessageHeaders,
+            message_body = outboxTableRow.MessageBody
+        }).ConfigureAwait(false);
+    }
+
+    public async Task<IEnumerable<OutboxTableRow>> Read(int batchSize, CancellationToken token = default)
+    {
+        var sql = """
+            DELETE FROM [outbox].[outbox]
+            OUTPUT [DELETED].[sequence_id] as [SequenceId],
+                [DELETED].[topic_name] as [TopicName],
+                [DELETED].[partition] as [Partition],
+                [DELETED].[message_key] as [MessageKey],
+                [DELETED].[message_headers] as [MessageHeaders],
+                [DELETED].[message_body] as [MessageBody]
+            WHERE
+                [sequence_id] IN (
+                    SELECT TOP (@batch_size) [sequence_id] FROM [outbox].[outbox]
+                    ORDER BY [sequence_id]
+                );
+            """;
+
+        using var conn = new SqlConnection(_options.ConnectionString);
+        return await conn.QueryAsync<OutboxTableRow>(sql, new { batch_size = batchSize }).ConfigureAwait(false);
+    }
+}

src/Contrib.KafkaFlow.Outbox/ConfigurationExtensions.cs

@@ -3,8 +3,6 @@
 
 namespace KafkaFlow.Outbox;
 
-public interface IOutboxDispatcher {}
-
 public static class ConfigurationExtensions
 {
     public static IServiceCollection AddOutboxBackend(this IServiceCollection services, Func<IServiceProvider, IOutboxBackend> factory) =>

src/Contrib.KafkaFlow.Outbox/Contrib.KafkaFlow.Outbox.csproj

@@ -20,6 +20,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+    <PackageReference Include="System.Text.Json" Version="8.0.3" />
   </ItemGroup>
 
 </Project>

src/Contrib.KafkaFlow.Outbox/IOutboxBackend.cs

@@ -1,8 +1,6 @@
 using Confluent.Kafka;
 
 namespace KafkaFlow.Outbox;
-
-public sealed record OutboxRecord(TopicPartition TopicPartition, Message<byte[], byte[]> Message);
 public interface IOutboxBackend
 {
     ValueTask Store(TopicPartition topicPartition, Message<byte[], byte[]> message, CancellationToken token2 = default);

src/Contrib.KafkaFlow.Outbox/IOutboxDispatcher.cs

@@ -0,0 +1,3 @@
+namespace KafkaFlow.Outbox;
+
+public interface IOutboxDispatcher {}