Refactored EventStoreDB subscriptions making it more scalable

Core.EntityFramework.Tests/Core.EntityFramework.Tests.csproj

@@ -0,0 +1,48 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net8.0</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <None Update="xunit.runner.json;appsettings.json">
+            <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+        </None>
+    </ItemGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\Core.EntityFramework\Core.EntityFramework.csproj"/>
+        <ProjectReference Include="..\Core.Testing\Core.Testing.csproj"/>
+    </ItemGroup>
+
+    <ItemGroup>
+        <PackageReference Include="FluentAssertions" Version="6.12.0"/>
+        <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.9.0"/>
+        <PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" Version="8.5.0"/>
+        <PackageReference Include="GitHubActionsTestLogger" Version="2.3.3">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+        <PackageReference Include="xunit" Version="2.8.0"/>
+        <PackageReference Include="xunit.runner.visualstudio" Version="2.8.0">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
+        </PackageReference>
+        <!-- For PostgreSQL checkpointing tests -->
+        <PackageReference Include="Testcontainers.PostgreSql" Version="3.8.0"/>
+        <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="8.0.4">
+            <PrivateAssets>all</PrivateAssets>
+            <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+        </PackageReference>
+    </ItemGroup>
+
+    <ItemGroup>
+      <Content Update="appsettings.json">
+        <ExcludeFromSingleFile>true</ExcludeFromSingleFile>
+        <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+        <CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
+      </Content>
+    </ItemGroup>
+
+
+</Project>

Core.EntityFramework.Tests/EntityFrameworkProjectionTests.cs

@@ -0,0 +1,124 @@
+using Core.EntityFramework.Projections;
+using Core.EntityFramework.Tests.Fixtures;
+using Core.Events;
+using FluentAssertions;
+using Microsoft.EntityFrameworkCore;
+using Xunit;
+
+namespace Core.EntityFramework.Tests;
+
+public record Opened(Guid ShoppingCartId, Guid ClientId);
+
+public record ProductItemAdded(Guid ShoppingCartId, int Count);
+
+public record Cancelled(Guid ShoppingCartId);
+
+public class ShoppingCartProjection: EntityFrameworkProjection<ShoppingCart, Guid, TestDbContext>
+{
+    public ShoppingCartProjection()
+    {
+        ViewId(c => c.Id);
+
+        Creates<Opened>(Create);
+        Projects<ProductItemAdded>(e => e.ShoppingCartId, Apply);
+        Deletes<Cancelled>(e => e.ShoppingCartId);
+    }
+
+    private ShoppingCart Create(Opened opened) =>
+        new() { Id = opened.ShoppingCartId, ClientId = opened.ClientId, ProductCount = 0 };
+
+
+    private ShoppingCart Apply(ShoppingCart cart, ProductItemAdded added)
+    {
+        cart.ProductCount += added.Count;
+        return cart;
+    }
+}
+
+public class EntityFrameworkProjectionTests(EfCorePostgresContainerFixture<TestDbContext> fixture)
+    : IClassFixture<EfCorePostgresContainerFixture<TestDbContext>>
+{
+    private readonly TestDbContext context = fixture.DbContext;
+
+    [Fact]
+    public async Task Applies_Works_Separately()
+    {
+        var projection = new ShoppingCartProjection { DbContext = context };
+
+        var cartId = Guid.NewGuid();
+        var clientId = Guid.NewGuid();
+
+        await projection.Handle([EventEnvelope.From(new Opened(cartId, clientId))], CancellationToken.None);
+        await context.SaveChangesAsync();
+
+        var savedEntity = await context.ShoppingCarts.Where(e => e.Id == cartId).FirstOrDefaultAsync();
+        savedEntity.Should().NotBeNull();
+        savedEntity.Should().BeEquivalentTo(new ShoppingCart { Id = cartId, ClientId = clientId, ProductCount = 0 });
+
+        await projection.Handle([EventEnvelope.From(new ProductItemAdded(cartId, 2))], CancellationToken.None);
+        await context.SaveChangesAsync();
+
+        savedEntity = await context.ShoppingCarts.Where(e => e.Id == cartId).FirstOrDefaultAsync();
+        savedEntity.Should().NotBeNull();
+        savedEntity.Should().BeEquivalentTo(new ShoppingCart { Id = cartId, ClientId = clientId, ProductCount = 2 });
+
+
+        await projection.Handle([EventEnvelope.From(new Cancelled(cartId))], CancellationToken.None);
+        await context.SaveChangesAsync();
+
+        savedEntity = await context.ShoppingCarts.Where(e => e.Id == cartId).FirstOrDefaultAsync();
+        savedEntity.Should().BeNull();
+    }
+
+    [Fact]
+    public async Task Applies_Works_In_Batch()
+    {
+        var projection = new ShoppingCartProjection { DbContext = context };
+
+        var cartId = Guid.NewGuid();
+        var clientId = Guid.NewGuid();
+
+        await projection.Handle([
+            EventEnvelope.From(new Opened(cartId, clientId)),
+            EventEnvelope.From(new ProductItemAdded(cartId, 2)),
+            EventEnvelope.From(new ProductItemAdded(cartId, 3)),
+            EventEnvelope.From(new ProductItemAdded(cartId, 5))
+        ], CancellationToken.None);
+        await context.SaveChangesAsync();
+
+        var savedEntity = await context.ShoppingCarts.Where(e => e.Id == cartId).FirstOrDefaultAsync();
+        savedEntity.Should().NotBeNull();
+        savedEntity.Should().BeEquivalentTo(new ShoppingCart { Id = cartId, ClientId = clientId, ProductCount = 10 });
+    }
+
+    [Fact]
+    public async Task Applies_Works_In_Batch_With_AddAndDeleteOnTheSameRecord()
+    {
+        var projection = new ShoppingCartProjection { DbContext = context };
+
+        var cartId = Guid.NewGuid();
+        var clientId = Guid.NewGuid();
+
+        await projection.Handle([
+            EventEnvelope.From(new Opened(cartId, clientId)),
+            EventEnvelope.From(new Cancelled(cartId))
+        ], CancellationToken.None);
+        await context.SaveChangesAsync();
+
+        var savedEntity = await context.ShoppingCarts.Where(e => e.Id == cartId).FirstOrDefaultAsync();
+        savedEntity.Should().BeNull();
+    }
+
+    [Fact]
+    public async Task SmokeTest()
+    {
+        var entity = new ShoppingCart { Id = Guid.NewGuid(), ProductCount = 2, ClientId = Guid.NewGuid() };
+        context.ShoppingCarts.Add(entity);
+        await context.SaveChangesAsync();
+
+        var savedEntity = await context.ShoppingCarts.FindAsync(entity.Id);
+        Assert.NotNull(savedEntity);
+        savedEntity.Should().NotBeNull();
+        savedEntity.Should().BeEquivalentTo(entity);
+    }
+}

Core.EntityFramework.Tests/Fixtures/EfCorePostgresContainerFixture.cs

@@ -0,0 +1,31 @@
+using Microsoft.EntityFrameworkCore;
+using Core.Testing.Fixtures;
+using Xunit;
+
+namespace Core.EntityFramework.Tests.Fixtures;
+
+public class EfCorePostgresContainerFixture<TContext>: IAsyncLifetime where TContext : DbContext
+{
+    private readonly PostgresContainerFixture postgresContainerFixture = new();
+
+    public TContext DbContext { get; private set; } = default!;
+
+    public async Task InitializeAsync()
+    {
+        await postgresContainerFixture.InitializeAsync().ConfigureAwait(false);
+
+        var optionsBuilder = new DbContextOptionsBuilder<TContext>()
+            .UseNpgsql(postgresContainerFixture.DataSource);
+
+        DbContext = (TContext)Activator.CreateInstance(typeof(TContext), optionsBuilder.Options)!;
+
+        await DbContext.Database.MigrateAsync().ConfigureAwait(false);
+    }
+
+    public async Task DisposeAsync()
+    {
+        await DbContext.DisposeAsync().ConfigureAwait(false);
+        await postgresContainerFixture.DisposeAsync().ConfigureAwait(false);
+    }
+}
+

Core.EntityFramework.Tests/Migrations/20240520164705_InitialCreate.cs

@@ -0,0 +1,35 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace Core.EntityFramework.Tests.Migrations
+{
+    /// <inheritdoc />
+    public partial class InitialCreate : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "ShoppingCarts",
+                columns: table => new
+                {
+                    Id = table.Column<Guid>(type: "uuid", nullable: false),
+                    ClientId = table.Column<Guid>(type: "uuid", nullable: false),
+                    ProductCount = table.Column<int>(type: "integer", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_ShoppingCarts", x => x.Id);
+                });
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "ShoppingCarts");
+        }
+    }
+}

Core.EntityFramework.Tests/Migrations/TestDbContextModelSnapshot.cs

@@ -0,0 +1,44 @@
+// <auto-generated />
+using System;
+using Core.EntityFramework.Tests;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
+
+#nullable disable
+
+namespace Core.EntityFramework.Tests.Migrations
+{
+    [DbContext(typeof(TestDbContext))]
+    partial class TestDbContextModelSnapshot : ModelSnapshot
+    {
+        protected override void BuildModel(ModelBuilder modelBuilder)
+        {
+#pragma warning disable 612, 618
+            modelBuilder
+                .HasAnnotation("ProductVersion", "8.0.4")
+                .HasAnnotation("Relational:MaxIdentifierLength", 63);
+
+            NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
+
+            modelBuilder.Entity("Core.EntityFramework.Tests.ShoppingCart", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<Guid>("ClientId")
+                        .HasColumnType("uuid");
+
+                    b.Property<int>("ProductCount")
+                        .HasColumnType("integer");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("ShoppingCarts");
+                });
+#pragma warning restore 612, 618
+        }
+    }
+}

Core.EntityFramework.Tests/TestDbContext.cs

@@ -0,0 +1,51 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Design;
+using Microsoft.Extensions.Configuration;
+
+namespace Core.EntityFramework.Tests;
+
+public class ShoppingCart
+{
+    public Guid Id { get; set; }
+    public Guid ClientId { get; set; }
+    public int ProductCount { get; set; }
+}
+
+public class TestDbContext(DbContextOptions<TestDbContext> options): DbContext(options)
+{
+    public DbSet<ShoppingCart> ShoppingCarts { get; set; } = default!;
+
+    protected override void OnModelCreating(ModelBuilder modelBuilder) =>
+        modelBuilder.Entity<ShoppingCart>();
+}
+
+public class TestDbContextFactory: IDesignTimeDbContextFactory<TestDbContext>
+{
+    public TestDbContext CreateDbContext(params string[] args)
+    {
+        var optionsBuilder = new DbContextOptionsBuilder<TestDbContext>();
+
+        if (optionsBuilder.IsConfigured)
+            return new TestDbContext(optionsBuilder.Options);
+
+        //Called by parameterless ctor Usually Migrations
+        var environmentName = Environment.GetEnvironmentVariable("EnvironmentName") ?? "Development";
+
+        var connectionString =
+            new ConfigurationBuilder()
+                .SetBasePath(AppContext.BaseDirectory)
+                .AddJsonFile("appsettings.json")
+                .AddJsonFile($"appsettings.{environmentName}.json", optional: true, reloadOnChange: false)
+                .AddEnvironmentVariables()
+                .Build()
+                .GetConnectionString("TestDb");
+
+        optionsBuilder.UseNpgsql(connectionString);
+
+        return new TestDbContext(optionsBuilder.Options);
+    }
+
+    public static TestDbContext Create()
+        => new TestDbContextFactory().CreateDbContext();
+}
+