Implement Polecat primary-store database-per-tenant (GH-3445)

Directory.Packages.props

@@ -45,7 +45,7 @@
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="16.0.0" />
     <PackageVersion Include="Marten" Version="9.15.1" />
     <PackageVersion Include="Microsoft.Data.SqlClient" Version="6.1.3" />
-    <PackageVersion Include="Polecat" Version="[4.8.0,6.0.0)" />
+    <PackageVersion Include="Polecat" Version="[5.1.0,6.0.0)" />
     <PackageVersion Include="Microsoft.Azure.Cosmos" Version="3.46.1" />
     <PackageVersion Include="Marten.AspNetCore" Version="9.15.1" />
     <PackageVersion Include="Marten.Newtonsoft" Version="9.15.1" />
@@ -124,13 +124,13 @@
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="9.0.5" />
     <PackageVersion Include="System.Net.NameResolution" Version="4.3.0" />
     <PackageVersion Include="System.Threading.Tasks.Dataflow" Version="9.0.5" />
-    <PackageVersion Include="Weasel.Core" Version="9.16.4" />
-    <PackageVersion Include="Weasel.EntityFrameworkCore" Version="9.16.4" />
-    <PackageVersion Include="Weasel.MySql" Version="9.16.4" />
-    <PackageVersion Include="Weasel.Oracle" Version="9.16.4" />
-    <PackageVersion Include="Weasel.Postgresql" Version="9.16.4" />
-    <PackageVersion Include="Weasel.SqlServer" Version="9.16.4" />
-    <PackageVersion Include="Weasel.Sqlite" Version="9.16.4" />
+    <PackageVersion Include="Weasel.Core" Version="9.17.0" />
+    <PackageVersion Include="Weasel.EntityFrameworkCore" Version="9.17.0" />
+    <PackageVersion Include="Weasel.MySql" Version="9.17.0" />
+    <PackageVersion Include="Weasel.Oracle" Version="9.17.0" />
+    <PackageVersion Include="Weasel.Postgresql" Version="9.17.0" />
+    <PackageVersion Include="Weasel.SqlServer" Version="9.17.0" />
+    <PackageVersion Include="Weasel.Sqlite" Version="9.17.0" />
     <PackageVersion Include="xunit" Version="2.9.3" />
     <PackageVersion Include="xunit.assemblyfixture" Version="2.2.0" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="2.8.2" />

docs/guide/durability/polecat/distribution.md

@@ -58,7 +58,15 @@ The `Uri` structure for event subscriptions or projections is:
 event-subscriptions://[event store type]/[event store name]/[database server].[database name]/[relative path of the shard]
 ```
 
-For example: `event-subscriptions://polecat/main/localhost.mydb/day/all`
+For an example from the tests: `event-subscriptions://sqlserver/main/localhost.mydb/day/all` where:
+
+* "sqlserver" is the event store *type*. Polecat reports its `IEventStore.Identity.Type` as its underlying
+  storage engine — `"SqlServer"`, lowercased by the URI — rather than "polecat", so this is where the
+  authority differs from the Marten page's `marten` example
+* "main" refers to this projection being in the primary Polecat store added from `AddPolecat()`. Otherwise
+  this value would be the type name of an ancillary store type in all lower case
+* "localhost" is the database server and "mydb" is the name of the database
+* "day/all" refers to a projection with the `ShardName` of "Day:All"
 
 ## Requirements
 

docs/guide/durability/polecat/multi-tenancy.md

@@ -20,9 +20,9 @@ builder.Services.AddPolecat(m =>
     {
         m.MultiTenantedDatabases(tenancy =>
         {
-            tenancy.AddSingleTenantDatabase("Server=localhost;Database=tenant1;...", "tenant1");
-            tenancy.AddSingleTenantDatabase("Server=localhost;Database=tenant2;...", "tenant2");
-            tenancy.AddSingleTenantDatabase("Server=localhost;Database=tenant3;...", "tenant3");
+            tenancy.AddTenant("tenant1", "Server=localhost;Database=tenant1;...");
+            tenancy.AddTenant("tenant2", "Server=localhost;Database=tenant2;...");
+            tenancy.AddTenant("tenant3", "Server=localhost;Database=tenant3;...");
         });
     })
     .IntegrateWithWolverine(x => x.MainDatabaseConnectionString = connectionString);

src/Persistence/PolecatTests/MultiTenancy/primary_store_database_per_tenant.cs

@@ -0,0 +1,108 @@
+using IntegrationTests;
+using JasperFx.Resources;
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Polecat;
+using Shouldly;
+using Wolverine;
+using Wolverine.Persistence.Durability;
+using Wolverine.Polecat;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Agents;
+using Wolverine.Tracking;
+
+namespace PolecatTests.MultiTenancy;
+
+// GH-3445: the PRIMARY Polecat store honors database-per-tenant. Before the fix, IntegrateWithWolverine()
+// unconditionally built a single non-tenanted SqlServerMessageStore and never read
+// MainDatabaseConnectionString, so tenant envelopes silently landed in the wrong database. Mirrors the
+// ancillary coverage (ancillary_stores_use_different_databases) and the Marten primary MultiTenancyFixture.
+public class primary_store_database_per_tenant : IAsyncLifetime
+{
+    private IHost theHost = null!;
+    private string tenant1ConnectionString = null!;
+    private string tenant2ConnectionString = null!;
+
+    public async Task InitializeAsync()
+    {
+        await using (var conn = new SqlConnection(Servers.SqlServerConnectionString))
+        {
+            await conn.OpenAsync();
+            tenant1ConnectionString = await CreateDatabaseIfNotExists(conn, "polecat_primary_t1");
+            tenant2ConnectionString = await CreateDatabaseIfNotExists(conn, "polecat_primary_t2");
+        }
+
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.Policies.AutoApplyTransactions();
+
+                opts.Services.AddPolecat(m =>
+                    {
+                        // Polecat still needs a base/default-tenant connection string even with
+                        // separate per-tenant databases configured.
+                        m.ConnectionString = Servers.SqlServerConnectionString;
+                        m.DatabaseSchemaName = "primary_mt";
+                        m.MultiTenantedDatabases(tenancy =>
+                        {
+                            tenancy.AddTenant("tenant1", tenant1ConnectionString);
+                            tenancy.AddTenant("tenant2", tenant2ConnectionString);
+                        });
+                    })
+                    .UseLightweightSessions()
+                    // MainDatabaseConnectionString is the tenant-neutral "master" store for nodes,
+                    // assignments, and dead letters.
+                    .IntegrateWithWolverine(x => x.MainDatabaseConnectionString = Servers.SqlServerConnectionString);
+
+                opts.Discovery.DisableConventionalDiscovery();
+                opts.Services.AddResourceSetupOnStartup();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    private static async Task<string> CreateDatabaseIfNotExists(SqlConnection conn, string databaseName)
+    {
+        var builder = new SqlConnectionStringBuilder(Servers.SqlServerConnectionString);
+
+        await using (var check = conn.CreateCommand())
+        {
+            check.CommandText = "SELECT DB_ID(@name)";
+            check.Parameters.AddWithValue("@name", databaseName);
+            var exists = await check.ExecuteScalarAsync();
+            if (exists is null || exists == DBNull.Value)
+            {
+                await using var create = conn.CreateCommand();
+                create.CommandText = $"CREATE DATABASE [{databaseName}]";
+                await create.ExecuteNonQueryAsync();
+            }
+        }
+
+        builder.InitialCatalog = databaseName;
+        return builder.ConnectionString;
+    }
+
+    [Fact]
+    public void primary_store_is_multi_tenanted()
+    {
+        // Before GH-3445 this was a plain SqlServerMessageStore and MultiTenanted was empty.
+        theHost.GetRuntime().Stores.MultiTenanted
+            .ShouldContain(x => x is MultiTenantedMessageStore);
+    }
+
+    [Fact]
+    public async Task durability_agents_exist_for_every_tenant_database()
+    {
+        var uris = (await theHost.GetRuntime().Stores.AllKnownAgentsAsync())
+            .Select(x => x.ToString()).ToArray();
+
+        uris.ShouldContain(x => x.Contains("polecat_primary_t1"));
+        uris.ShouldContain(x => x.Contains("polecat_primary_t2"));
+    }
+}

src/Persistence/Wolverine.Polecat/WolverineOptionsPolecatExtensions.cs

@@ -81,6 +81,19 @@ public static PolecatConfigurationExpression IntegrateWithWolverine(
                              store.Options.DatabaseSchemaName ??
                              "wolverine";
 
+            // GH-3445: honor Polecat's database-per-tenant on the primary store just like the ancillary
+            // path (and the Marten twin at WolverineOptionsMartenExtensions.cs) — build a
+            // MultiTenantedMessageStore over a PolecatMessageDatabaseSource so each tenant's envelope
+            // storage lands in that tenant's database, with a "main" store for tenant-neutral operations.
+            // Dispatch on cardinality rather than the tenancy's type name (see the ancillary seam's note
+            // on marten#4864). This is where MainDatabaseConnectionString is finally read.
+            if (store.Options.Tenancy != null &&
+                store.Options.Tenancy.Cardinality != JasperFx.Descriptors.DatabaseCardinality.Single)
+            {
+                return BuildMultiTenantedMessageStore(schemaName, store, runtime,
+                    integration.MainDatabaseConnectionString);
+            }
+
             return BuildSqlServerMessageStore(schemaName, store, runtime, logger);
         });
 
@@ -154,6 +167,48 @@ internal static IMessageStore BuildSqlServerMessageStore(
         return new SqlServerMessageStore(settings, runtime.Options.Durability, logger, sagaTypes);
     }
 
+    // GH-3445: primary-store database-per-tenant. Mirrors the ancillary
+    // BuildMultiTenantedMessageDatabase<T> and the Marten twin BuildMultiTenantedMessageDatabase, but
+    // with Role.Main for the master store. The master holds tenant-neutral state (nodes, assignments,
+    // dead letters) and each tenant's envelope tables live in that tenant's own SQL Server database via
+    // PolecatMessageDatabaseSource.
+    internal static IMessageStore BuildMultiTenantedMessageStore(
+        string schemaName,
+        IDocumentStore store,
+        IWolverineRuntime runtime,
+        string? masterDatabaseConnectionString)
+    {
+        var connectionString = masterDatabaseConnectionString ?? store.Options.ConnectionString;
+
+        if (connectionString.IsEmpty())
+        {
+            throw new ArgumentOutOfRangeException(nameof(masterDatabaseConnectionString),
+                $"Wolverine requires a main message store database even if the current Polecat tenancy model does not. Configure it via {nameof(PolecatIntegration)}.{nameof(PolecatIntegration.MainDatabaseConnectionString)} in the IntegrateWithWolverine() configuration.");
+        }
+
+        var masterSettings = new DatabaseSettings
+        {
+            SchemaName = schemaName,
+            AutoCreate = AutoCreate.CreateOrUpdate,
+            Role = MessageStoreRole.Main,
+            CommandQueuesEnabled = true,
+            ConnectionString = connectionString
+        };
+
+        var sagaTypes = runtime.Services.GetServices<SagaTableDefinition>();
+        var main = new SqlServerMessageStore(masterSettings, runtime.Options.Durability,
+            runtime.LoggerFactory.CreateLogger<SqlServerMessageStore>(), sagaTypes)
+        {
+            Name = "Main"
+        };
+
+        var source = new PolecatMessageDatabaseSource(schemaName, AutoCreate.CreateOrUpdate, store, runtime);
+
+        main.Initialize(runtime);
+
+        return new MultiTenantedMessageStore(main, runtime, source);
+    }
+
     /// <summary>
     ///     Register a custom subscription that will process a batch of Polecat events at a time with
     ///     a user defined action