feat(pubsub): named-broker and broker-per-tenant support (GH-3306)

docs/guide/messaging/transports/gcp-pubsub/index.md

@@ -64,6 +64,92 @@ opts.UsePubsub("your-project-id")
 
 The credential manages its own token refresh lifecycle, so no additional background task is required. For more control over the underlying GCP client builders, see [Customisation](/guide/messaging/transports/gcp-pubsub/customisation).
 
+## Multiple / Named Brokers
+
+You can connect to more than one GCP Pub/Sub broker (typically a different GCP project) from a single Wolverine application by registering an additional, *named* broker alongside the default one. Endpoints are then pinned to the named broker with the `...OnNamedBroker` overloads:
+
+<!-- snippet: sample_named_pubsub_broker -->
+<a id='snippet-sample_named_pubsub_broker'></a>
+```cs
+var host = await Host.CreateDefaultBuilder()
+    .UseWolverine(opts =>
+    {
+        // The default / shared Pub/Sub broker
+        opts.UsePubsub("your-project-id").AutoProvision();
+
+        // An additional, independent Pub/Sub broker pointed at a different GCP project.
+        // The Wolverine Uri scheme for endpoints on this broker becomes the broker name
+        // ("americas"), e.g. americas://americas-project-id/colors
+        opts.AddNamedPubsubBroker(new BrokerName("americas"), "americas-project-id")
+            .AutoProvision();
+
+        // Pin specific endpoints to the named broker
+        opts.PublishMessage<ColorMessage>()
+            .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors");
+        opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors");
+    }).StartAsync();
+```
+<sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs#L67-L84' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_named_pubsub_broker' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+Note that the `Uri` scheme within Wolverine for any endpoint on a *named* GCP Pub/Sub broker is the broker name you supply, not `pubsub`. So in the example above you would see `Uri` values like `americas://americas-project-id/colors`.
+
+## Multi-Tenancy with a Broker Per Tenant
+
+Named brokers (above) are a *static* topology: you pin specific endpoints to a specific broker at configuration time. **Broker-per-tenant** is different — it is *runtime* routing. You declare one shared topic topology, and each tenant is served by its **own dedicated GCP project**. Which project a message goes to (and which project an inbound message came from) is decided at runtime by the message's [tenant id](/guide/handlers/multi-tenancy), typically set through `DeliveryOptions.TenantId`.
+
+Project-id-per-tenant is the natural isolation axis for Pub/Sub: the topic and subscription names embed the project id, so the *same* logical topic under a *different* project is already a physically distinct Pub/Sub resource — "shared by name, isolated by project".
+
+<!-- snippet: sample_pubsub_broker_per_tenant -->
+<a id='snippet-sample_pubsub_broker_per_tenant'></a>
+```cs
+var host = await Host.CreateDefaultBuilder()
+    .UseWolverine(opts =>
+    {
+        // The "default" / shared Pub/Sub connection on its own GCP project
+        opts.UsePubsub("shared-project-id")
+            .AutoProvision()
+
+            // How should Wolverine route a message whose TenantId is null or
+            // unknown? FallbackToDefault (the default) uses the shared project;
+            // TenantIdRequired throws; IgnoreUnknownTenants silently drops it.
+            .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault)
+
+            // Each tenant is served by its OWN dedicated GCP project, but shares
+            // the topic topology declared below. Project-id-per-tenant is the
+            // natural isolation axis: the same logical topic under a different
+            // project is a physically distinct Pub/Sub resource.
+            .AddTenant("tenant1", "tenant1-project-id")
+
+            // A tenant may also carry its own dedicated credentials by configuring
+            // its client builders (seeded from the parent transport otherwise):
+            .AddTenant("tenant2", "tenant2-project-id", tenant =>
+            {
+                tenant.ConfigurePublisherApiBuilder =
+                    builder => { /* builder.GoogleCredential = ...; */ return ValueTask.CompletedTask; };
+            });
+
+        // One shared topology; messages are routed to the right project at runtime
+        // by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant1" }).
+        opts.PublishMessage<ColorMessage>().ToPubsubTopic("colors");
+        opts.ListenToPubsubTopic("colors");
+    }).StartAsync();
+```
+<sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs#L91-L122' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_pubsub_broker_per_tenant' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+To route a specific message to a tenant's project, stamp the tenant id on the send:
+
+```csharp
+await bus.SendAsync(new ColorMessage("blue"), new DeliveryOptions { TenantId = "tenant1" });
+```
+
+Wolverine wraps the outbound endpoint in a `TenantedSender` that dispatches on `Envelope.TenantId`, and builds a compound listener that runs one listener per tenant project — each inbound envelope is stamped with the tenant id of the project it was consumed from. When `AutoProvision()` is enabled, Wolverine provisions the shared topology (topics and subscriptions) on **every** tenant project, not just the default one.
+
+::: tip The emulator caveat
+The GCP Pub/Sub emulator ignores credentials and accepts arbitrary project ids with no auth, so per-tenant *projects* are trivially testable on a single emulator (just use distinct project id strings). Per-tenant *credentials*, however, cannot be exercised against the emulator — test that path against real GCP.
+:::
+
 ## Request/Reply
 
 [Request/reply](https://www.enterpriseintegrationpatterns.com/patterns/messaging/RequestReply.html) mechanics (`IMessageBus.InvokeAsync<T>()`) are possible with the GCP Pub/Sub transport *if* Wolverine has the ability to auto-provision a specific response topic and subscription for each node. That topic and subscription would be named like `wlvrn.response.[application node id]` if you happen to notice that in your GCP Pub/Sub.

src/Transports/GCP/Wolverine.Pubsub.Tests/DocumentationSamples.cs

@@ -6,6 +6,7 @@
 using Microsoft.Extensions.Hosting;
 using Wolverine.ComplianceTests.Compliance;
 using Wolverine.Transports;
+using Wolverine.Transports.Sending;
 using Wolverine.Util;
 
 namespace Wolverine.Pubsub.Tests;
@@ -61,6 +62,68 @@ public async Task enable_system_endpoints()
         #endregion
     }
 
+    public async Task named_pubsub_brokers()
+    {
+        #region sample_named_pubsub_broker
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                // The default / shared Pub/Sub broker
+                opts.UsePubsub("your-project-id").AutoProvision();
+
+                // An additional, independent Pub/Sub broker pointed at a different GCP project.
+                // The Wolverine Uri scheme for endpoints on this broker becomes the broker name
+                // ("americas"), e.g. americas://americas-project-id/colors
+                opts.AddNamedPubsubBroker(new BrokerName("americas"), "americas-project-id")
+                    .AutoProvision();
+
+                // Pin specific endpoints to the named broker
+                opts.PublishMessage<ColorMessage>()
+                    .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors");
+                opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors");
+            }).StartAsync();
+
+        #endregion
+    }
+
+    public async Task broker_per_tenant()
+    {
+        #region sample_pubsub_broker_per_tenant
+        var host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                // The "default" / shared Pub/Sub connection on its own GCP project
+                opts.UsePubsub("shared-project-id")
+                    .AutoProvision()
+
+                    // How should Wolverine route a message whose TenantId is null or
+                    // unknown? FallbackToDefault (the default) uses the shared project;
+                    // TenantIdRequired throws; IgnoreUnknownTenants silently drops it.
+                    .TenantIdBehavior(TenantedIdBehavior.FallbackToDefault)
+
+                    // Each tenant is served by its OWN dedicated GCP project, but shares
+                    // the topic topology declared below. Project-id-per-tenant is the
+                    // natural isolation axis: the same logical topic under a different
+                    // project is a physically distinct Pub/Sub resource.
+                    .AddTenant("tenant1", "tenant1-project-id")
+
+                    // A tenant may also carry its own dedicated credentials by configuring
+                    // its client builders (seeded from the parent transport otherwise):
+                    .AddTenant("tenant2", "tenant2-project-id", tenant =>
+                    {
+                        tenant.ConfigurePublisherApiBuilder =
+                            builder => { /* builder.GoogleCredential = ...; */ return ValueTask.CompletedTask; };
+                    });
+
+                // One shared topology; messages are routed to the right project at runtime
+                // by Envelope.TenantId (e.g. new DeliveryOptions { TenantId = "tenant1" }).
+                opts.PublishMessage<ColorMessage>().ToPubsubTopic("colors");
+                opts.ListenToPubsubTopic("colors");
+            }).StartAsync();
+
+        #endregion
+    }
+
     public async Task configuring_listeners()
     {
         #region sample_listen_to_pubsub_topic
@@ -470,4 +533,5 @@ public async ValueTask<GoogleCredential> GetAsync()
         => throw new NotImplementedException();
 }
 
-#endregion
+#endregion
+public record ColorMessage(string Color);

src/Transports/GCP/Wolverine.Pubsub.Tests/NamedBrokerComplianceTests.cs

@@ -0,0 +1,120 @@
+using Google.Api.Gax;
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.Runtime;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.Pubsub.Tests;
+
+public class NamedBrokerComplianceTests
+{
+    // A named broker is a second, independent PubsubTransport whose Protocol (and therefore URI scheme) is the
+    // broker name, and which carries its own project id. No broker needed.
+    [Fact]
+    public void named_broker_is_a_distinct_transport_with_its_own_protocol_and_project()
+    {
+        var options = new WolverineOptions();
+
+        options.UsePubsub("wolverine");
+        options.AddNamedPubsubBroker(new BrokerName("americas"), "wolverine2");
+
+        var defaultTransport = options.Transports.GetOrCreate<PubsubTransport>();
+        var named = options.Transports.OfType<PubsubTransport>().Single(x => x.Protocol == "americas");
+
+        named.ShouldNotBeSameAs(defaultTransport);
+        defaultTransport.Protocol.ShouldBe(PubsubTransport.ProtocolName);
+        defaultTransport.ProjectId.ShouldBe("wolverine");
+        named.Protocol.ShouldBe("americas");
+        named.ProjectId.ShouldBe("wolverine2");
+    }
+
+    [Fact]
+    public void named_broker_endpoint_uri_scheme_is_the_broker_name()
+    {
+        var options = new WolverineOptions();
+
+        options.AddNamedPubsubBroker(new BrokerName("americas"), "wolverine2");
+        options.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "colors");
+
+        var named = options.Transports.OfType<PubsubTransport>().Single(x => x.Protocol == "americas");
+        var endpoint = named.Topics["colors"];
+
+        endpoint.Uri.Scheme.ShouldBe("americas");
+        endpoint.Uri.ShouldBe(new Uri("americas://wolverine2/colors"));
+        named.ResourceUri.ShouldBe(new Uri("americas://wolverine2"));
+    }
+}
+
+// Integration: round-trip a message over a second project on the same emulator, addressed exclusively through the
+// named broker. Skip-guarded when the emulator/Docker is unavailable.
+public class NamedBrokerRoundTripTests : IAsyncLifetime
+{
+    private IHost? _host;
+    private bool _skip;
+
+    public async Task InitializeAsync()
+    {
+        _skip = !await TestingExtensions.IsEmulatorAvailable();
+        if (_skip)
+        {
+            return;
+        }
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                // Default/shared broker on project "wolverine"...
+                opts.UsePubsubTesting().AutoProvision().AutoPurgeOnStartup();
+
+                // ...plus a named broker on a second project "wolverine2" on the same emulator.
+                opts.AddNamedPubsubBrokerTesting(new BrokerName("americas"), "wolverine2").AutoProvision()
+                    .AutoPurgeOnStartup();
+
+                opts.PublishMessage<NamedBrokerMessage>()
+                    .ToPubsubTopicOnNamedBroker(new BrokerName("americas"), "named-broker-colors");
+                opts.ListenToPubsubTopicOnNamedBroker(new BrokerName("americas"), "named-broker-colors");
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        if (_host is not null)
+        {
+            await _host.StopAsync();
+            _host.Dispose();
+        }
+    }
+
+    [Fact]
+    public async Task round_trips_a_message_through_the_named_broker()
+    {
+        if (_skip)
+        {
+            return;
+        }
+
+        var message = new NamedBrokerMessage("blue");
+
+        var session = await _host!.TrackActivity()
+            .IncludeExternalTransports()
+            .Timeout(1.Minutes())
+            .SendMessageAndWaitAsync(message);
+
+        var received = session.Received.SingleEnvelope<NamedBrokerMessage>();
+        received.Message.ShouldBeOfType<NamedBrokerMessage>().Color.ShouldBe("blue");
+
+        // The receiving endpoint lives on the named broker, so its Uri scheme is the broker name.
+        received.Destination!.Scheme.ShouldBe("americas");
+    }
+}
+
+public record NamedBrokerMessage(string Color);
+
+public static class NamedBrokerMessageHandler
+{
+    public static void Handle(NamedBrokerMessage message)
+    {
+    }
+}