Merge pull request #293 from Vulthil/fix/outbox-sync-capture-and-validation

Capture domain events and validate outbox delay options for sync SaveChanges callers

Author: Vulthil

src/Vulthil.SharedKernel.Outbox.EntityFrameworkCore/DomainEventsToOutboxMessageSaveChangesInterceptor.cs

@@ -15,16 +15,53 @@ public sealed class DomainEventsToOutboxMessageSaveChangesInterceptor(TimeProvid
 {
     private readonly TimeProvider _timeProvider = timeProvider;
 
+    /// <summary>
+    /// Captures domain events from tracked aggregate roots and stores them as outbox messages before persisting changes.
+    /// </summary>
+    public override InterceptionResult<int> SavingChanges(DbContextEventData eventData, InterceptionResult<int> result)
+    {
+        CaptureDomainEvents(eventData.Context);
+        return result;
+    }
+
     /// <summary>
     /// Captures domain events from tracked aggregate roots and stores them as outbox messages before persisting changes.
     /// </summary>
     public override ValueTask<InterceptionResult<int>> SavingChangesAsync(DbContextEventData eventData, InterceptionResult<int> result, CancellationToken cancellationToken = default)
     {
-        var dbContext = eventData.Context;
+        CaptureDomainEvents(eventData.Context);
+        return base.SavingChangesAsync(eventData, result, cancellationToken);
+    }
+
+    /// <summary>
+    /// Wakes the outbox relay after a save that committed outside an explicit transaction, so domain events captured
+    /// by a bare <c>SaveChanges</c> are relayed promptly instead of waiting for the next poll. When a transaction is
+    /// open the relay is signalled on commit by the transaction-commit interceptor instead, so this skips that case to
+    /// avoid waking the relay before the rows are committed and visible.
+    /// </summary>
+    public override int SavedChanges(SaveChangesCompletedEventData eventData, int result)
+    {
+        WakeRelayIfNeeded(eventData.Context);
+        return result;
+    }
+
+    /// <summary>
+    /// Wakes the outbox relay after a save that committed outside an explicit transaction, so domain events captured
+    /// by a bare <c>SaveChanges</c> are relayed promptly instead of waiting for the next poll. When a transaction is
+    /// open the relay is signalled on commit by the transaction-commit interceptor instead, so this skips that case to
+    /// avoid waking the relay before the rows are committed and visible.
+    /// </summary>
+    public override ValueTask<int> SavedChangesAsync(SaveChangesCompletedEventData eventData, int result, CancellationToken cancellationToken = default)
+    {
+        WakeRelayIfNeeded(eventData.Context);
+        return base.SavedChangesAsync(eventData, result, cancellationToken);
+    }
 
+    private void CaptureDomainEvents(DbContext? dbContext)
+    {
         if (dbContext is not ISaveOutboxMessages dbContextWithOutboxMessages)
         {
-            return base.SavingChangesAsync(eventData, result, cancellationToken);
+            return;
         }
 
         Activity? activity = null;
@@ -55,24 +92,14 @@ public override ValueTask<InterceptionResult<int>> SavingChangesAsync(DbContextE
             }).ToList();
 
         dbContextWithOutboxMessages.OutboxMessages.AddRange(outboxMessages);
-
-        return base.SavingChangesAsync(eventData, result, cancellationToken);
     }
 
-    /// <summary>
-    /// Wakes the outbox relay after a save that committed outside an explicit transaction, so domain events captured
-    /// by a bare <c>SaveChanges</c> are relayed promptly instead of waiting for the next poll. When a transaction is
-    /// open the relay is signalled on commit by the transaction-commit interceptor instead, so this skips that case to
-    /// avoid waking the relay before the rows are committed and visible.
-    /// </summary>
-    public override ValueTask<int> SavedChangesAsync(SaveChangesCompletedEventData eventData, int result, CancellationToken cancellationToken = default)
+    private void WakeRelayIfNeeded(DbContext? dbContext)
     {
-        if (ShouldWakeRelay(eventData.Context))
+        if (ShouldWakeRelay(dbContext))
         {
             signal.Notify();
         }
-
-        return base.SavedChangesAsync(eventData, result, cancellationToken);
     }
 
     private static bool ShouldWakeRelay(DbContext? dbContext) =>

src/Vulthil.SharedKernel.Outbox.EntityFrameworkCore/PublicAPI.Unshipped.txt

@@ -1 +1,3 @@
 #nullable enable
+override Vulthil.SharedKernel.Outbox.EntityFrameworkCore.DomainEventsToOutboxMessageSaveChangesInterceptor.SavedChanges(Microsoft.EntityFrameworkCore.Diagnostics.SaveChangesCompletedEventData! eventData, int result) -> int
+override Vulthil.SharedKernel.Outbox.EntityFrameworkCore.DomainEventsToOutboxMessageSaveChangesInterceptor.SavingChanges(Microsoft.EntityFrameworkCore.Diagnostics.DbContextEventData! eventData, Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<int> result) -> Microsoft.EntityFrameworkCore.Diagnostics.InterceptionResult<int>

src/Vulthil.SharedKernel.Outbox/OutboxEngineServiceCollectionExtensions.cs

@@ -39,6 +39,9 @@ public static IServiceCollection AddOutboxEngine(
                     && o.Retention.SweepInterval > TimeSpan.Zero
                     && o.Retention.BatchSize >= 1,
                 "Outbox retention requires RetentionPeriod and SweepInterval greater than zero and BatchSize of at least 1 when enabled.")
+            .Validate(
+                o => o.MaxDelaySeconds >= o.OutboxProcessingDelaySeconds,
+                "MaxDelaySeconds must be greater than or equal to OutboxProcessingDelaySeconds.")
             .ValidateDataAnnotations()
             .ValidateOnStart();
 

tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests/DomainEventsToOutboxMessageSaveChangesInterceptorTests.cs

@@ -37,6 +37,24 @@ public async Task NonTransactionalSaveCapturingDomainEventsWakesTheRelay()
         GetMock<IOutboxSignal>().Verify(signal => signal.Notify(), Times.Once());
     }
 
+    [Fact]
+    public async Task SyncSaveCapturingDomainEventsWakesTheRelay()
+    {
+        // Arrange
+        await using var context = NewContext();
+        var aggregate = new TestAggregate(Guid.NewGuid());
+        aggregate.RaiseSomething();
+        context.Aggregates.Add(aggregate);
+
+        // Act
+        SaveChangesSynchronously(context);
+
+        // Assert
+        GetMock<IOutboxSignal>().Verify(signal => signal.Notify(), Times.Once());
+        var captured = await context.OutboxMessages.SingleAsync(CancellationToken);
+        captured.Type.ShouldBe(typeof(TestDomainEvent).FullName);
+    }
+
     [Fact]
     public async Task SaveInsideAnExplicitTransactionDoesNotWakeTheRelay()
     {
@@ -113,6 +131,8 @@ private TestDbContext NewContext(bool withInterceptor = true)
         return new TestDbContext(builder.Options);
     }
 
+    private static void SaveChangesSynchronously(TestDbContext context) => context.SaveChanges();
+
     public sealed class TestDbContext(DbContextOptions<TestDbContext> options) : DbContext(options), ISaveOutboxMessages
     {
         public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs

@@ -0,0 +1,43 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using Vulthil.xUnit;
+
+namespace Vulthil.SharedKernel.Outbox.Tests;
+
+public sealed class OutboxEngineServiceCollectionExtensionsTests : BaseUnitTestCase
+{
+    [Fact]
+    public void MaxDelaySecondsLessThanOutboxProcessingDelaySecondsFailsValidationAtStartup()
+    {
+        // Arrange
+        var services = new ServiceCollection();
+        services.AddOutboxEngine(o =>
+        {
+            o.OutboxProcessingDelaySeconds = 100;
+            o.MaxDelaySeconds = 1;
+        });
+        using var provider = services.BuildServiceProvider();
+
+        // Act & Assert
+        Should.Throw<OptionsValidationException>(() => provider.GetRequiredService<IOptions<OutboxProcessingOptions>>().Value);
+    }
+
+    [Fact]
+    public void MaxDelaySecondsEqualToOutboxProcessingDelaySecondsPassesValidation()
+    {
+        // Arrange
+        var services = new ServiceCollection();
+        services.AddOutboxEngine(o =>
+        {
+            o.OutboxProcessingDelaySeconds = 5;
+            o.MaxDelaySeconds = 5;
+        });
+        using var provider = services.BuildServiceProvider();
+
+        // Act
+        var options = provider.GetRequiredService<IOptions<OutboxProcessingOptions>>().Value;
+
+        // Assert
+        options.MaxDelaySeconds.ShouldBe(5);
+    }
+}