Close outbox seam test-coverage gaps

tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs

@@ -1,8 +1,10 @@
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
+using Vulthil.Extensions.Testing;
 using Vulthil.IntegrationTests.Fixtures;
 using Vulthil.Messaging.Abstractions.Publishers;
 using Vulthil.Messaging.TestHarness;
+using Vulthil.Results;
 using Vulthil.SharedKernel.Outbox;
 using Vulthil.SharedKernel.Outbox.EntityFrameworkCore;
 using Vulthil.TestHost.Data;
@@ -15,6 +17,7 @@ public sealed class TransactionalOutboxIntegrationTests(TestHarnessWebApplicatio
     : BaseIntegrationTestCase<TestHarnessWebApplicationFactory, Program>(factory, testOutputHelper), IClassFixture<TestHarnessWebApplicationFactory>
 {
     private ITestHarness Harness => Factory.Services.GetRequiredService<ITestHarness>();
+    private IRequester Requester => Factory.Services.GetRequiredService<IRequester>();
 
     [Fact]
     public async Task PublishInsideATransactionIsCapturedIntoTheOutboxAndNotSentDirectly()
@@ -44,4 +47,72 @@ await dbContext.ExecuteInTransactionAsync(async cancellationToken =>
         capturedDuringTransaction.ShouldBeGreaterThan(0);
         sentDuringTransaction.ShouldBeFalse();
     }
+
+    [Fact]
+    public async Task CommittingATransactionRelaysTheMessageAndTheConsumerReceivesItWithTheStableMessageId()
+    {
+        // Arrange
+        var id = Guid.NewGuid();
+        var messageId = Guid.NewGuid().ToString();
+        var dbContext = ScopedServices.GetRequiredService<TestHostDbContext>();
+        var publisher = ScopedServices.GetRequiredService<IPublisher>();
+
+        // Act — publish inside a committed transaction, then wait for the real relay to dispatch it.
+        await dbContext.ExecuteInTransactionAsync(async cancellationToken =>
+        {
+            await publisher.PublishAsync(
+                new ProbeCreatedIntegrationEvent(id),
+                context =>
+                {
+                    context.SetMessageId(messageId);
+                    return ValueTask.CompletedTask;
+                },
+                cancellationToken);
+
+            return 0;
+        }, CancellationToken);
+
+        var polled = await Polling.WaitAsync(TimeSpan.FromSeconds(15), async () =>
+        {
+            var consumed = Harness.Consumed<ProbeCreatedIntegrationEvent>().FirstOrDefault(message => message.Message.Id == id);
+            return consumed is not null
+                ? Result.Success(consumed)
+                : Result.Failure<CapturedMessage<ProbeCreatedIntegrationEvent>>(
+                    Error.NotFound("Probe.NotYetRelayed", "The captured message has not been relayed and consumed yet."));
+        }, CancellationToken);
+
+        // Assert
+        polled.IsSuccess.ShouldBeTrue();
+        polled.Value.Envelope.MessageId.ShouldBe(messageId);
+
+        var sideEffects = await Requester.RequestAsync<GetProbeSideEffects, List<ProbeSideEffectDto>>(new GetProbeSideEffects(id), CancellationToken);
+        sideEffects.IsSuccess.ShouldBeTrue();
+        sideEffects.Value.ShouldContain(sideEffect => sideEffect.ProbeId == id);
+    }
+
+    [Fact]
+    public async Task RollingBackATransactionNeverRelaysTheCapturedMessage()
+    {
+        // Arrange
+        var id = Guid.NewGuid();
+        var dbContext = ScopedServices.GetRequiredService<TestHostDbContext>();
+        var publisher = ScopedServices.GetRequiredService<IPublisher>();
+
+        // Act — the operation reports failure, so ExecuteInTransactionAsync rolls back instead of committing.
+        await dbContext.ExecuteInTransactionAsync(
+            async cancellationToken =>
+            {
+                await publisher.PublishAsync(new ProbeCreatedIntegrationEvent(id), cancellationToken);
+                return false;
+            },
+            static operationSucceeded => operationSucceeded,
+            CancellationToken);
+        await Task.Delay(TimeSpan.FromSeconds(1), CancellationToken);
+
+        // Assert — fetch every row and filter in memory: Content is mapped to jsonb, which Npgsql cannot LIKE-match server-side.
+        var outboxMessages = await dbContext.OutboxMessages.AsNoTracking().ToListAsync(CancellationToken);
+        outboxMessages.ShouldNotContain(message => message.Content.Contains(id.ToString()));
+        Harness.Published<ProbeCreatedIntegrationEvent>().ShouldNotContain(message => message.Message.Id == id);
+        Harness.Consumed<ProbeCreatedIntegrationEvent>().ShouldNotContain(message => message.Message.Id == id);
+    }
 }

tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj

@@ -10,6 +10,7 @@
 	</ItemGroup>
 
 	<ItemGroup>
+		<ProjectReference Include="..\..\src\Vulthil.Extensions.Testing\Vulthil.Extensions.Testing.csproj" />
 		<ProjectReference Include="..\..\src\Vulthil.Messaging.Inbox.Cosmos\Vulthil.Messaging.Inbox.Cosmos.csproj" />
 		<ProjectReference Include="..\..\src\Vulthil.Messaging.TestHarness\Vulthil.Messaging.TestHarness.csproj" />
 		<ProjectReference Include="..\..\src\Vulthil.SharedKernel.Infrastructure.Cosmos\Vulthil.SharedKernel.Infrastructure.Cosmos.csproj" />

tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs

@@ -1,3 +1,6 @@
+using System.Text.Json;
+using Vulthil.Messaging.Abstractions.Publishers;
+using Vulthil.Messaging.Transport;
 using Vulthil.SharedKernel.Outbox;
 using Vulthil.xUnit;
 
@@ -26,4 +29,123 @@ public async Task DispatchAsyncThrowsDescriptiveErrorWhenMessageTypeCannotBeReso
         // Assert
         exception.Message.ShouldContain("Vulthil.Nonexistent.PhantomMessage");
     }
+
+    [Fact]
+    public async Task DispatchAsyncPublishesThePayloadWithTheCapturedMessageIdCorrelationIdRoutingKeyAndHeaders()
+    {
+        // Arrange
+        var jsonOptions = new JsonSerializerOptions();
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
+        Func<IPublishContext, ValueTask>? capturedConfigure = null;
+        GetMock<ITransportPublisher>()
+            .Setup(publisher => publisher.PublishAsync(It.IsAny<TestMessage>(), It.IsAny<Func<IPublishContext, ValueTask>>(), It.IsAny<CancellationToken>()))
+            .Callback<TestMessage, Func<IPublishContext, ValueTask>, CancellationToken>((_, configure, _) => capturedConfigure = configure)
+            .Returns(Task.CompletedTask);
+        var metadata = new BrokerOutboxMetadata
+        {
+            MessageId = "stable-message-id",
+            CorrelationId = "correlation-1",
+            RoutingKey = "routing-key-1",
+            Headers = new Dictionary<string, object?> { ["tenant"] = "acme" },
+        };
+        var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
+        var message = new OutboxMessageData(
+            Id: Guid.NewGuid(),
+            Type: typeof(TestMessage).FullName!,
+            Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
+            TraceParent: null,
+            TraceState: null,
+            Destination: OutboxDestination.Publish,
+            Metadata: JsonSerializer.Serialize(metadata, jsonOptions));
+
+        // Act
+        await dispatcher.DispatchAsync(message, CancellationToken);
+
+        // Assert
+        GetMock<ITransportPublisher>().Verify(
+            publisher => publisher.PublishAsync(It.Is<TestMessage>(m => m.Value == "hello"), It.IsAny<Func<IPublishContext, ValueTask>>(), CancellationToken),
+            Times.Once);
+        capturedConfigure.ShouldNotBeNull();
+        var appliedContext = new PublishContext();
+        await capturedConfigure!(appliedContext);
+        appliedContext.MessageId.ShouldBe("stable-message-id");
+        appliedContext.CorrelationId.ShouldBe("correlation-1");
+        appliedContext.RoutingKey.ShouldBe("routing-key-1");
+        appliedContext.Headers["tenant"].ShouldBe("acme");
+    }
+
+    [Fact]
+    public async Task DispatchAsyncSendsThePayloadToTheCapturedDestinationAddressWithFidelity()
+    {
+        // Arrange
+        var jsonOptions = new JsonSerializerOptions();
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
+        var sendEndpoint = new Mock<ISendEndpoint>();
+        Func<IPublishContext, ValueTask>? capturedConfigure = null;
+        sendEndpoint
+            .Setup(endpoint => endpoint.SendAsync(It.IsAny<TestMessage>(), It.IsAny<Func<IPublishContext, ValueTask>>(), It.IsAny<CancellationToken>()))
+            .Callback<TestMessage, Func<IPublishContext, ValueTask>, CancellationToken>((_, configure, _) => capturedConfigure = configure)
+            .Returns(Task.CompletedTask);
+        var destinationAddress = new Uri("queue:order-commands");
+        GetMock<ITransportSendEndpointProvider>()
+            .Setup(provider => provider.GetSendEndpointAsync(destinationAddress, It.IsAny<CancellationToken>()))
+            .ReturnsAsync(sendEndpoint.Object);
+        var metadata = new BrokerOutboxMetadata
+        {
+            MessageId = "stable-send-id",
+            DestinationAddress = destinationAddress.ToString(),
+            Headers = new Dictionary<string, object?> { ["tenant"] = "acme" },
+        };
+        var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
+        var message = new OutboxMessageData(
+            Id: Guid.NewGuid(),
+            Type: typeof(TestMessage).FullName!,
+            Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
+            TraceParent: null,
+            TraceState: null,
+            Destination: OutboxDestination.Send,
+            Metadata: JsonSerializer.Serialize(metadata, jsonOptions));
+
+        // Act
+        await dispatcher.DispatchAsync(message, CancellationToken);
+
+        // Assert
+        sendEndpoint.Verify(
+            endpoint => endpoint.SendAsync(It.Is<TestMessage>(m => m.Value == "hello"), It.IsAny<Func<IPublishContext, ValueTask>>(), CancellationToken),
+            Times.Once);
+        capturedConfigure.ShouldNotBeNull();
+        var appliedContext = new PublishContext();
+        await capturedConfigure!(appliedContext);
+        appliedContext.MessageId.ShouldBe("stable-send-id");
+        appliedContext.Headers["tenant"].ShouldBe("acme");
+    }
+
+    [Fact]
+    public async Task DispatchAsyncThrowsDescriptiveErrorWhenASendMessageHasNoDestinationAddress()
+    {
+        // Arrange
+        var jsonOptions = new JsonSerializerOptions();
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions);
+        var dispatcher = CreateInstance<BrokerOutboxDispatcher>();
+        var message = new OutboxMessageData(
+            Id: Guid.NewGuid(),
+            Type: typeof(TestMessage).FullName!,
+            Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions),
+            TraceParent: null,
+            TraceState: null,
+            Destination: OutboxDestination.Send,
+            Metadata: null);
+
+        // Act
+        var exception = await Should.ThrowAsync<InvalidOperationException>(
+            () => dispatcher.DispatchAsync(message, CancellationToken));
+
+        // Assert
+        exception.Message.ShouldContain("missing its destination address");
+        GetMock<ITransportSendEndpointProvider>().Verify(
+            provider => provider.GetSendEndpointAsync(It.IsAny<Uri>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    private sealed record TestMessage(string Value);
 }

tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs

@@ -0,0 +1,84 @@
+using Vulthil.Messaging.Abstractions.Consumers;
+using Vulthil.Messaging.Transport;
+using Vulthil.SharedKernel.Application.Data;
+using Vulthil.xUnit;
+
+namespace Vulthil.Messaging.Outbox.Tests;
+
+public sealed class TransactionalConsumeFilterTests : BaseUnitTestCase
+{
+    [Fact]
+    public async Task ConsumeAsyncRunsTheNextDelegateInsideUnitOfWorkExecuteInTransactionAsync()
+    {
+        // Arrange
+        SetUpUnitOfWorkToInvokeTheOperation();
+        var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
+        var context = NewContext();
+        var nextInvoked = false;
+        ConsumeDelegate<TestMessage> next = receivedContext =>
+        {
+            nextInvoked = ReferenceEquals(receivedContext, context);
+            return Task.CompletedTask;
+        };
+
+        // Act
+        await filter.ConsumeAsync(context, next);
+
+        // Assert
+        nextInvoked.ShouldBeTrue();
+        GetMock<IUnitOfWork>().Verify(
+            unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), context.CancellationToken),
+            Times.Once);
+    }
+
+    [Fact]
+    public async Task ConsumeAsyncNeverRunsTheNextDelegateWhenTheUnitOfWorkDoesNotInvokeTheOperation()
+    {
+        // Arrange
+        GetMock<IUnitOfWork>()
+            .Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(false);
+        var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
+        var context = NewContext();
+        var nextInvoked = false;
+        ConsumeDelegate<TestMessage> next = _ =>
+        {
+            nextInvoked = true;
+            return Task.CompletedTask;
+        };
+
+        // Act
+        await filter.ConsumeAsync(context, next);
+
+        // Assert
+        nextInvoked.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task ConsumeAsyncPropagatesAnExceptionThrownByTheNextDelegate()
+    {
+        // Arrange
+        SetUpUnitOfWorkToInvokeTheOperation();
+        var filter = CreateInstance<TransactionalConsumeFilter<TestMessage>>();
+        var context = NewContext();
+        ConsumeDelegate<TestMessage> next = _ => throw new InvalidOperationException("handler failed");
+
+        // Act & Assert
+        await Should.ThrowAsync<InvalidOperationException>(() => filter.ConsumeAsync(context, next));
+    }
+
+    private void SetUpUnitOfWorkToInvokeTheOperation() =>
+        GetMock<IUnitOfWork>()
+            .Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny<Func<CancellationToken, Task<bool>>>(), It.IsAny<CancellationToken>()))
+            .Returns((Func<CancellationToken, Task<bool>> operation, CancellationToken token) => operation(token));
+
+    private static MessageContext<TestMessage> NewContext() => new()
+    {
+        Message = new TestMessage("hello"),
+        RoutingKey = string.Empty,
+        Headers = new Dictionary<string, object?>(),
+        CancellationToken = CancellationToken,
+    };
+
+    private sealed record TestMessage(string Value);
+}

tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Text.Json;
 using System.Transactions;
 using Microsoft.Extensions.Options;
@@ -88,6 +89,73 @@ public async Task PublishingInsideAnEntityFrameworkTransactionCapturesTheMessage
         GetMock<IOutboxStore>().Verify(store => store.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
     }
 
+    [Fact]
+    public async Task EnableTracingCapturesTheCurrentActivityOntoTheCapturedRow()
+    {
+        // Arrange
+        GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
+        Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions { EnableTracing = true }));
+        Use(TimeProvider.System);
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
+        var filter = CreateInstance<TransactionalPublishFilter>();
+        var context = NewContext();
+        OutboxMessage? captured = null;
+        GetMock<IOutboxStore>()
+            .Setup(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()))
+            .Callback<OutboxMessage>(message => captured = message);
+        using var activity = new Activity("test-publish");
+        activity.TraceStateString = "congo=t61rcWkgMzE";
+
+        // Act
+        activity.Start();
+        try
+        {
+            await filter.PublishAsync(context, _ => Task.CompletedTask);
+        }
+        finally
+        {
+            activity.Stop();
+        }
+
+        // Assert
+        captured.ShouldNotBeNull();
+        captured.TraceParent.ShouldBe(activity.Id);
+        captured.TraceState.ShouldBe(activity.TraceStateString);
+    }
+
+    [Fact]
+    public async Task EnableTracingDisabledLeavesTraceFieldsNullEvenWithAnActiveActivity()
+    {
+        // Arrange
+        GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
+        Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions { EnableTracing = false }));
+        Use(TimeProvider.System);
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
+        var filter = CreateInstance<TransactionalPublishFilter>();
+        var context = NewContext();
+        OutboxMessage? captured = null;
+        GetMock<IOutboxStore>()
+            .Setup(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()))
+            .Callback<OutboxMessage>(message => captured = message);
+        using var activity = new Activity("test-publish");
+
+        // Act
+        activity.Start();
+        try
+        {
+            await filter.PublishAsync(context, _ => Task.CompletedTask);
+        }
+        finally
+        {
+            activity.Stop();
+        }
+
+        // Assert
+        captured.ShouldNotBeNull();
+        captured.TraceParent.ShouldBeNull();
+        captured.TraceState.ShouldBeNull();
+    }
+
     private static PublishFilterContext NewContext() => new()
     {
         Message = new TestMessage("hello"),