Map Wolverine TenantId from incoming MassTransit messages

docs/guide/messaging/transports/rabbitmq/interoperability.md

@@ -221,8 +221,34 @@ Wolverine = await Host.CreateDefaultBuilder().UseWolverine(opts =>
         // Tell Wolverine to make this endpoint interoperable with MassTransit
         .UseMassTransitInterop(mt =>
         {
-            // optionally customize the inner JSON serialization
+            // optionally customize the inner JSON serialization, or map the Wolverine
+            // tenant id from each incoming MassTransit message (see below)
         })
         .DefaultIncomingMessage<ResponseMessage>().UseForReplies();
 }).StartAsync();
 ```
+
+### Mapping the Tenant Id
+
+When consuming messages from MassTransit, you can derive Wolverine's tenant id for each incoming
+message from either the message body or the MassTransit envelope metadata (headers, addresses,
+correlation ids). Register one or more `MapTenantIdFrom<T>` mappers inside `UseMassTransitInterop`:
+
+```cs
+opts.ListenToRabbitQueue("orders")
+    .UseMassTransitInterop(mt =>
+    {
+        // Pull the tenant id straight off the strongly-typed message body
+        mt.MapTenantIdFrom<OrderPlaced>(env => env.Message?.TenantId);
+
+        // ...or from a MassTransit header carried on the envelope
+        mt.MapTenantIdFrom<OrderShipped>(env =>
+            env.Headers.TryGetValue("tenant-id", out var value) ? value?.ToString() : null);
+    });
+```
+
+The lambda receives the strongly-typed MassTransit envelope (`IMassTransitEnvelope<T>`), which exposes
+the deserialized `Message` alongside the MassTransit metadata. Each registration applies only to its own
+message type, and returning `null` or an empty string leaves the tenant id untouched. Tenant mapping
+affects only the inbound (deserialization) path, and works on any MassTransit-interop listener
+(Rabbit MQ, Azure Service Bus, Amazon SQS).

src/Testing/CoreTests/Runtime/Interop/MassTransitTenantMappingTests.cs

@@ -0,0 +1,109 @@
+using NSubstitute;
+using Wolverine.Runtime.Interop.MassTransit;
+using Xunit;
+
+namespace CoreTests.Runtime.Interop;
+
+public class MassTransitTenantMappingTests
+{
+    private readonly IMassTransitInteropEndpoint theEndpoint = Substitute.For<IMassTransitInteropEndpoint>();
+
+    public MassTransitTenantMappingTests()
+    {
+        // The serializer round-trips a response address through the envelope; give the fake
+        // endpoint a valid reply Uri so the write path doesn't emit an empty address.
+        theEndpoint.MassTransitReplyUri().Returns(new Uri("rabbitmq://localhost/responses"));
+    }
+
+    // Round-trips a message through the MassTransit serializer: writes it in the
+    // MassTransit envelope format, then reads it back the way an incoming message would be
+    // deserialized. Returns the incoming Envelope so tests can assert on mapped metadata.
+    private Envelope readIncoming(MassTransitJsonSerializer serializer, object message)
+    {
+        var outgoing = new Envelope { Id = Guid.NewGuid(), Message = message };
+        var data = serializer.Write(outgoing);
+
+        var incoming = new Envelope { Data = data };
+        serializer.ReadFromData(message.GetType(), incoming);
+        return incoming;
+    }
+
+    [Fact]
+    public void maps_tenant_id_from_the_incoming_message_body()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer.MapTenantIdFrom<TenantMessage>(env => env.Message?.TenantId);
+
+        var incoming = readIncoming(serializer, new TenantMessage("acme", "hello"));
+
+        incoming.TenantId.ShouldBe("acme");
+    }
+
+    [Fact]
+    public void maps_tenant_id_from_a_masstransit_header()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer.MapTenantIdFrom<TenantMessage>(env =>
+            env.Headers.TryGetValue("tenant-id", out var value) ? value?.ToString() : null);
+
+        var outgoing = new Envelope { Id = Guid.NewGuid(), Message = new TenantMessage("ignored", "hello") };
+        outgoing.Headers["tenant-id"] = "globex";
+        var data = serializer.Write(outgoing);
+
+        var incoming = new Envelope { Data = data };
+        serializer.ReadFromData(typeof(TenantMessage), incoming);
+
+        incoming.TenantId.ShouldBe("globex");
+    }
+
+    [Fact]
+    public void leaves_tenant_id_untouched_for_an_unregistered_message_type()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer.MapTenantIdFrom<TenantMessage>(env => env.Message?.TenantId);
+
+        var incoming = readIncoming(serializer, new OtherMessage("no tenant here"));
+
+        incoming.TenantId.ShouldBeNull();
+    }
+
+    [Fact]
+    public void does_not_overwrite_tenant_id_when_the_source_returns_empty()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer.MapTenantIdFrom<TenantMessage>(_ => "");
+
+        var incoming = readIncoming(serializer, new TenantMessage("acme", "hello"));
+
+        incoming.TenantId.ShouldBeNull();
+    }
+
+    [Fact]
+    public void supports_multiple_registered_message_types()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer
+            .MapTenantIdFrom<TenantMessage>(env => env.Message?.TenantId)
+            .MapTenantIdFrom<AnotherTenantMessage>(env => env.Message?.Org);
+
+        readIncoming(serializer, new TenantMessage("acme", "hello")).TenantId.ShouldBe("acme");
+        readIncoming(serializer, new AnotherTenantMessage("globex", "hi")).TenantId.ShouldBe("globex");
+    }
+
+    [Fact]
+    public void falls_through_the_whole_mapper_chain_for_an_unregistered_type()
+    {
+        var serializer = new MassTransitJsonSerializer(theEndpoint);
+        serializer
+            .MapTenantIdFrom<TenantMessage>(env => env.Message?.TenantId)
+            .MapTenantIdFrom<AnotherTenantMessage>(env => env.Message?.Org);
+
+        var incoming = readIncoming(serializer, new OtherMessage("matches nothing"));
+
+        incoming.TenantId.ShouldBeNull();
+    }
+}
+
+public record TenantMessage(string TenantId, string Body);
+public record AnotherTenantMessage(string Org, string Body);
+public record OtherMessage(string Body);

src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_mapper_tenant_mapping.cs

@@ -0,0 +1,40 @@
+using Shouldly;
+using Wolverine.AmazonSqs.Internal;
+using Wolverine.Runtime.Interop.MassTransit;
+using Xunit;
+
+namespace Wolverine.AmazonSqs.Tests;
+
+public class masstransit_mapper_tenant_mapping
+{
+    // The SQS MassTransit mapper must apply the UseMassTransitInterop(configure) lambda to its
+    // serializer, otherwise MapTenantIdFrom (and any other serializer customization) silently
+    // no-ops on the Amazon SQS listener.
+    [Fact]
+    public void mapper_applies_the_configure_lambda_to_its_serializer()
+    {
+        var endpoint = new FakeMassTransitEndpoint();
+
+        var mapper = new MassTransitMapper(endpoint,
+            mt => mt.MapTenantIdFrom<SqsTenantMessage>(env => env.Message?.TenantId));
+
+        var serializer = mapper.Serializer;
+
+        var outgoing = new Envelope { Id = Guid.NewGuid(), Message = new SqsTenantMessage("acme", "hello") };
+        var data = serializer.Write(outgoing);
+
+        var incoming = new Envelope { Data = data };
+        serializer.ReadFromData(typeof(SqsTenantMessage), incoming);
+
+        incoming.TenantId.ShouldBe("acme");
+    }
+
+    private sealed class FakeMassTransitEndpoint : IMassTransitInteropEndpoint
+    {
+        public Uri? MassTransitUri() => null;
+        public Uri? MassTransitReplyUri() => new Uri("sqs://responses");
+        public Uri? TranslateMassTransitToWolverineUri(Uri uri) => null;
+    }
+
+    public record SqsTenantMessage(string TenantId, string Body);
+}

src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsListenerConfiguration.cs

@@ -156,7 +156,7 @@ public AmazonSqsListenerConfiguration UseNServiceBusInterop(string? replyQueueNa
 
     public AmazonSqsListenerConfiguration UseMassTransitInterop(Action<IMassTransitInterop>? configure = null)
     {
-        add(e => e.Mapper = new MassTransitMapper((Endpoint as IMassTransitInteropEndpoint)!));
+        add(e => e.Mapper = new MassTransitMapper((Endpoint as IMassTransitInteropEndpoint)!, configure));
         return this;
     }
 

src/Transports/AWS/Wolverine.AmazonSqs/Internal/MassTransitMapper.cs

@@ -12,10 +12,11 @@ internal class MassTransitMapper : ISqsEnvelopeMapper
     private readonly IMassTransitInteropEndpoint _endpoint;
     private MassTransitJsonSerializer _serializer;
 
-    public MassTransitMapper(IMassTransitInteropEndpoint endpoint)
+    public MassTransitMapper(IMassTransitInteropEndpoint endpoint, Action<IMassTransitInterop>? configure = null)
     {
         _endpoint = endpoint;
         _serializer = new MassTransitJsonSerializer(endpoint);
+        configure?.Invoke(_serializer);
     }
 
     public override string ToString() => "MassTransit Interop";

src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/masstransit_interop_map_tenant_id.cs

@@ -0,0 +1,78 @@
+using System.Text;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.ComplianceTests;
+using Wolverine.RabbitMQ.Internal;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+using Wolverine.Util;
+using Xunit;
+
+namespace Wolverine.RabbitMQ.Tests;
+
+public class masstransit_interop_map_tenant_id
+{
+    // UseMassTransitInterop(configure) must thread the configure lambda all the way down to
+    // the MassTransit serializer so MapTenantIdFrom takes effect on the Rabbit MQ listener.
+    // Uses stubbed transports (no live broker) and drives the deserialization path directly.
+    [Fact]
+    public async Task map_tenant_id_from_is_applied_on_the_rabbitmq_listener()
+    {
+        using var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.DisableConventionalDiscovery();
+                opts.IncludeType<TenantOrderPlacedHandler>();
+
+                // Bogus host name on purpose: stubbed transports must never connect.
+                opts.UseRabbitMq(x => x.HostName = Guid.NewGuid().ToString());
+                opts.ListenToRabbitQueue("orders")
+                    .UseMassTransitInterop(mt => mt.MapTenantIdFrom<TenantOrderPlaced>(env => env.Message?.Tenant));
+
+                opts.StubAllExternalTransports();
+            }).StartAsync();
+
+        var runtime = host.GetRuntime();
+
+        var endpoint = runtime.Endpoints.EndpointFor(new Uri("rabbitmq://queue/orders"))
+            .ShouldNotBeNull()
+            .ShouldBeAssignableTo<RabbitMqEndpoint>();
+
+        // Applies UseMassTransitInterop and wires the MassTransit serializer onto the endpoint
+        // (a live listener does this at startup; stubbed transports skip listener startup).
+        endpoint.BuildMapper(runtime);
+
+        // The wire payload a MassTransit producer sends: the real message lives under "message".
+        var json = $$"""
+        {
+            "messageId": "{{Guid.NewGuid()}}",
+            "messageType": ["urn:message:Orders:TenantOrderPlaced"],
+            "message": { "orderId": 92883, "tenant": "acme" }
+        }
+        """;
+
+        var incoming = new Envelope
+        {
+            Data = Encoding.UTF8.GetBytes(json),
+            ContentType = "application/vnd.masstransit+json",
+            MessageType = typeof(TenantOrderPlaced).ToMessageTypeName(),
+            Destination = endpoint.Uri
+        };
+
+        await runtime.Pipeline.TryDeserializeEnvelope(incoming);
+
+        incoming.Message.ShouldBeOfType<TenantOrderPlaced>().OrderId.ShouldBe(92883);
+        incoming.TenantId.ShouldBe("acme");
+    }
+
+    public record TenantOrderPlaced(int OrderId, string Tenant);
+
+    public class TenantOrderPlacedHandler
+    {
+        // Presence registers TenantOrderPlaced in the HandlerGraph so the pipeline can
+        // resolve its message type during deserialization.
+        public void Handle(TenantOrderPlaced order)
+        {
+        }
+    }
+}

src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqEndpoint.MassTransit.cs

@@ -54,6 +54,6 @@ public abstract partial class RabbitMqEndpoint : IMassTransitInteropEndpoint
 
     public void UseMassTransitInterop(Action<IMassTransitInterop>? configure = null)
     {
-        customizeMapping((m, _) => m.InteropWithMassTransit());
+        customizeMapping((m, _) => m.InteropWithMassTransit(configure));
     }
 }

src/Wolverine/Runtime/Interop/MassTransit/IMassTransitInterop.cs

@@ -10,6 +10,23 @@ public interface IMassTransitInterop
     /// <param name="configuration"></param>
     void UseSystemTextJsonForSerialization(Action<JsonSerializerOptions>? configuration = null);
 
+    /// <summary>
+    ///     Derive the Wolverine <see cref="Envelope.TenantId" /> for incoming MassTransit messages of
+    ///     type <typeparamref name="T" /> from the message itself or its MassTransit metadata. The
+    ///     supplied lambda receives the strongly-typed MassTransit envelope and returns the tenant id
+    ///     (or <c>null</c> / empty to leave the tenant id untouched). This only affects the inbound
+    ///     (deserialization) path. Registering multiple message types is supported — each registration
+    ///     applies only to its own <typeparamref name="T" />. Registering the same type more than once
+    ///     replaces the previous mapping for that type.
+    /// </summary>
+    /// <param name="tenantIdSource">
+    ///     Maps the incoming MassTransit envelope to a tenant id, e.g.
+    ///     <c>env =&gt; env.Message?.TenantId</c> or
+    ///     <c>env =&gt; env.Headers.TryGetValue("tenant-id", out var v) ? v?.ToString() : null</c>.
+    /// </param>
+    /// <typeparam name="T">The Wolverine message type to extract the tenant id from.</typeparam>
+    IMassTransitInterop MapTenantIdFrom<T>(Func<IMassTransitEnvelope<T>, string?> tenantIdSource) where T : class;
+
     // Newtonsoft.Json variant moved to WolverineFx.Newtonsoft as the
     // UseNewtonsoftForSerialization(this IMassTransitInterop, ...)
     // extension method. Install WolverineFx.Newtonsoft to opt in.

src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs

@@ -10,7 +10,39 @@ internal interface IMassTransitEnvelope
     void TransferData(Envelope envelope);
 }
 
-internal class MassTransitEnvelope<T> : IMassTransitEnvelope where T : class
+/// <summary>
+///     Read-only, strongly-typed view over an incoming MassTransit "envelope" message. Exposed to
+///     user code through hooks such as <see cref="IMassTransitInterop.MapTenantIdFrom{T}" /> so that
+///     Wolverine metadata (e.g. the tenant id) can be derived either from the deserialized message
+///     body or from the surrounding MassTransit transport metadata (headers, addresses, ids).
+/// </summary>
+/// <typeparam name="T">The Wolverine message type carried by the MassTransit envelope.</typeparam>
+public interface IMassTransitEnvelope<T> where T : class
+{
+    /// <summary>The deserialized message body.</summary>
+    T? Message { get; }
+
+    string? MessageId { get; }
+    string? RequestId { get; }
+    string? CorrelationId { get; }
+    string? ConversationId { get; }
+    string? InitiatorId { get; }
+
+    string? SourceAddress { get; }
+    string? ResponseAddress { get; }
+    string? DestinationAddress { get; }
+    string? FaultAddress { get; }
+
+    string[]? MessageType { get; }
+
+    DateTime? SentTime { get; }
+    DateTime? ExpirationTime { get; }
+
+    /// <summary>The MassTransit message headers, as deserialized from the envelope.</summary>
+    IReadOnlyDictionary<string, object?> Headers { get; }
+}
+
+internal class MassTransitEnvelope<T> : IMassTransitEnvelope, IMassTransitEnvelope<T> where T : class
 {
     public MassTransitEnvelope()
     {
@@ -68,6 +100,8 @@ public MassTransitEnvelope(Envelope envelope)
 
     public Dictionary<string, object?> Headers { get; set; } = new();
 
+    IReadOnlyDictionary<string, object?> IMassTransitEnvelope<T>.Headers => Headers;
+
     // Wolverine doesn't care about this, so don't bother deserializing it
     // ReSharper disable once UnusedMember.Global
     public BusHostInfo Host => BusHostInfo.Instance;