From 6463d66bf6dd4d57aee8de6569f6d85076501b81 Mon Sep 17 00:00:00 2001 From: HiroshiYoshino Date: Mon, 5 Oct 2026 17:16:40 +0900 Subject: [PATCH 1/2] test(eventing): reproduce dedicated-tenant dispatch routing regression Framework-owned Outbox rows already follow a tenant's dedicated database, but event dispatch rebuilds the ambient tenant from its ID alone. This loses connection metadata and can write consumer effects and Inbox rows to default while completing the correct source Outbox. Add a same-tenant Finbuckle scope diagnostic and a public-dispatch PostgreSQL regression using default plus one dedicated database. Keep all 13 assertions for persisted business rows, per-handler Inbox placement, source completion, tenant ownership, and unchanged redelivery snapshots. Retain the supplemental same-tenant, cross-tenant, and global routing test. Production remains unchanged; the two primary regressions intentionally fail on the unfixed baseline. Supplemental test host ownership is corrected in the following fix. Red checkpoint: 12 nearby existing tests passed and the two primary regressions failed for the expected metadata/routing mismatch, with no skips. The original execution checkpoint is retained in backup history. --- .../DedicatedTenantOutboxRoutingTests.cs | 227 ++++++++++++++++ .../OutboxDedicatedTenantDispatchTests.cs | 243 ++++++++++++++++++ .../FinbuckleEventTenantScopeTests.cs | 29 +++ 3 files changed, 499 insertions(+) create mode 100644 src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs create mode 100644 src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs diff --git a/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs b/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs new file mode 100644 index 0000000000..caf5761e60 --- /dev/null +++ b/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs @@ -0,0 +1,227 @@ +using Finbuckle.MultiTenant.Abstractions; +using Finbuckle.MultiTenant.EntityFrameworkCore.Stores; +using Finbuckle.MultiTenant.Extensions; +using FSH.Framework.Eventing; +using FSH.Framework.Eventing.Abstractions; +using FSH.Framework.Eventing.Outbox; +using FSH.Framework.Eventing.Persistence; +using FSH.Framework.Persistence; +using FSH.Framework.Shared.Multitenancy; +using FSH.Modules.Catalog.Data; +using FSH.Modules.Catalog.Domain; +using FSH.Modules.Files.Contracts.Events; +using FSH.Modules.Multitenancy.Data; +using FSH.Modules.Multitenancy.Services; +using Integration.Tests.Infrastructure; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; +using Npgsql; + +namespace Integration.Tests.Tests.Eventing; + +[Collection(FshCollectionDefinition.Name)] +public sealed class DedicatedTenantOutboxRoutingTests +{ + private readonly FshWebApplicationFactory _factory; + + public DedicatedTenantOutboxRoutingTests(FshWebApplicationFactory factory) + { + _factory = factory; + } + + [Fact] + public async Task DispatchAsync_Should_PersistConsumerAndInboxOnlyInDedicatedDatabase_When_SourceTenantMatchesEventTenant() + { + string defaultConnection = _factory.Services.GetRequiredService>().Value.ConnectionString; + string databaseName = $"outbox_dispatch_{Guid.CreateVersion7():N}"; + string tenantId = $"dispatch-{Guid.CreateVersion7():N}"; + var connectionBuilder = new NpgsqlConnectionStringBuilder(defaultConnection) { Database = databaseName }; + string dedicatedConnection = connectionBuilder.ConnectionString; + var tenant = new AppTenantInfo(tenantId, tenantId, dedicatedConnection, "dispatch@example.test"); + Guid eventId = Guid.CreateVersion7(); + var integrationEvent = new FileFinalizedIntegrationEvent( + eventId, DateTime.UtcNow, tenantId, $"corr-{eventId:N}", "Files", + Guid.CreateVersion7(), "Test", null, "application/octet-stream", 1, 1); + await using ServiceProvider provider = BuildProvider(defaultConnection); + var accessor = provider.GetRequiredService>(); + var setter = provider.GetRequiredService(); + var previousContext = accessor.MultiTenantContext; + var drainScope = provider.GetRequiredService(); + bool databaseCreated = false; + + try + { + await ExecuteDatabaseCommandAsync(defaultConnection, $"CREATE DATABASE {QuoteIdentifier(databaseName)}"); + databaseCreated = true; + + using (drainScope.Begin(new EventingDrainTarget(tenantId, dedicatedConnection))) + using (var scope = provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().Database.MigrateAsync(); + await scope.ServiceProvider.GetRequiredService().Database.MigrateAsync(); + await scope.ServiceProvider.GetRequiredService().AddAsync(integrationEvent); + } + + using (var scope = provider.CreateScope()) + { + var store = scope.ServiceProvider.GetRequiredService>(); + (await store.AddAsync(tenant)).ShouldBeTrue(); + var targets = await scope.ServiceProvider.GetRequiredService().GetTargetsAsync(); + targets.Any(target => target.TenantId is null && target.ConnectionString is null).ShouldBeTrue(); + EventingDrainTarget target = targets.Single(target => target.TenantId == tenantId); + target.ConnectionString.ShouldBe(dedicatedConnection); + + using (drainScope.Begin(target)) + using (var sourceScope = provider.CreateScope()) + { + await sourceScope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + } + + DatabaseSnapshot dedicated = await ReadSnapshotAsync(dedicatedConnection, integrationEvent); + DatabaseSnapshot shared = await ReadSnapshotAsync(defaultConnection, integrationEvent); + + using (drainScope.Begin(new EventingDrainTarget(tenantId, dedicatedConnection))) + using (var sourceScope = provider.CreateScope()) + { + var context = sourceScope.ServiceProvider.GetRequiredService(); + await context.OutboxMessages.Where(message => message.Id == eventId) + .ExecuteUpdateAsync(update => update.SetProperty(message => message.ProcessedOnUtc, (DateTime?)null)); + await sourceScope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + + DatabaseSnapshot dedicatedAfterRedelivery = await ReadSnapshotAsync(dedicatedConnection, integrationEvent); + DatabaseSnapshot sharedAfterRedelivery = await ReadSnapshotAsync(defaultConnection, integrationEvent); + + Assert.Multiple( + () => dedicated.ConsumerRows.ShouldBe(1, "the tenant-aware consumer must write to the dedicated database"), + () => dedicated.ConsumerTenantRows.ShouldBe(1, "the business row must belong to the event tenant"), + () => shared.ConsumerRows.ShouldBe(0, "the default database must not receive a dedicated tenant's side effect"), + () => dedicated.InboxRows.ShouldBe(1, "the consumer's Inbox must be recorded in the dedicated database"), + () => dedicated.InboxTenantRows.ShouldBe(1, "the Inbox row must belong to the event tenant"), + () => shared.InboxRows.ShouldBe(0, "the default database must not receive the consumer's Inbox"), + () => dedicated.OutboxRows.ShouldBe(1), + () => dedicated.ProcessedRows.ShouldBe(1, "completion must update the original source Outbox"), + () => dedicated.ClaimedRows.ShouldBe(0), + () => dedicated.Retries.ShouldBe(0), + () => shared.OutboxRows.ShouldBe(0), + () => dedicatedAfterRedelivery.ShouldBe(dedicated, "Inbox must suppress duplicate business writes on redelivery"), + () => sharedAfterRedelivery.ShouldBe(shared, "redelivery must not introduce default-database side effects")); + } + finally + { + setter.MultiTenantContext = previousContext; + using (var scope = provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService>().RemoveAsync(tenantId); + } + + await CleanupDefaultRowsAsync(defaultConnection, integrationEvent); + if (databaseCreated) + { + await ExecuteDatabaseCommandAsync(defaultConnection, $"DROP DATABASE {QuoteIdentifier(databaseName)} WITH (FORCE)"); + } + } + } + + private ServiceProvider BuildProvider(string defaultConnection) + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["EventingOptions:Provider"] = "InMemory", + ["EventingOptions:UseHostedServiceDispatcher"] = "false" + }).Build(); + var services = new ServiceCollection(); + services.AddSingleton(configuration); + services.AddSingleton(_factory.Services.GetRequiredService()); + services.AddSingleton(TimeProvider.System); + services.AddLogging(); + services.Configure(options => + { + options.Provider = "POSTGRESQL"; + options.ConnectionString = defaultConnection; + options.MigrationsAssembly = "FSH.Starter.Migrations.PostgreSQL"; + }); + services.AddScoped(); + services.AddScoped(); + services.AddDbContext(options => options.UseNpgsql(defaultConnection)); + services.AddMultiTenant() + .WithStore>(ServiceLifetime.Scoped); + services.AddEventingCore(configuration); + services.AddHeroDbContext(); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Singleton()); + services.Replace(ServiceDescriptor.Scoped()); + services.AddScoped, FileEventMarkerHandler>(); + return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true }); + } + + private static string QuoteIdentifier(string identifier) + { + using var builder = new NpgsqlCommandBuilder(); + return builder.QuoteIdentifier(identifier); + } + + private static async Task ExecuteDatabaseCommandAsync(string connectionString, string sql) + { + await using var connection = new NpgsqlConnection(connectionString); + await connection.OpenAsync(); +#pragma warning disable CA2100 + await using var command = new NpgsqlCommand(sql, connection); +#pragma warning restore CA2100 + await command.ExecuteNonQueryAsync(); + } + + private static async Task ReadSnapshotAsync(string connectionString, FileFinalizedIntegrationEvent integrationEvent) + { + await using var connection = new NpgsqlConnection(connectionString); + await connection.OpenAsync(); + await using var command = new NpgsqlCommand(""" + SELECT + (SELECT COUNT(*) FROM catalog."Brands" WHERE "Name" = @marker), + (SELECT COUNT(*) FROM catalog."Brands" WHERE "Name" = @marker AND "TenantId" = @tenant), + (SELECT COUNT(*) FROM framework."InboxMessages" WHERE "Id" = @event AND "HandlerName" = @handler), + (SELECT COUNT(*) FROM framework."InboxMessages" WHERE "Id" = @event AND "HandlerName" = @handler AND "TenantId" = @tenant), + (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event), + (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event AND "ProcessedOnUtc" IS NOT NULL), + (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event AND "ClaimedUntilUtc" IS NOT NULL), + (SELECT COALESCE(SUM("RetryCount"), 0) FROM framework."OutboxMessages" WHERE "Id" = @event) + """, connection); + command.Parameters.AddWithValue("marker", $"dispatch-{integrationEvent.Id:N}"); + command.Parameters.AddWithValue("tenant", integrationEvent.TenantId!); + command.Parameters.AddWithValue("event", integrationEvent.Id); + command.Parameters.AddWithValue("handler", typeof(FileEventMarkerHandler).FullName!); + await using var reader = await command.ExecuteReaderAsync(); + (await reader.ReadAsync()).ShouldBeTrue(); + return new DatabaseSnapshot(reader.GetInt64(0), reader.GetInt64(1), reader.GetInt64(2), reader.GetInt64(3), + reader.GetInt64(4), reader.GetInt64(5), reader.GetInt64(6), reader.GetInt64(7)); + } + + private static async Task CleanupDefaultRowsAsync(string connectionString, FileFinalizedIntegrationEvent integrationEvent) + { + await using var connection = new NpgsqlConnection(connectionString); + await connection.OpenAsync(); + await using var command = new NpgsqlCommand(""" + DELETE FROM catalog."Brands" WHERE "Name" = @marker; + DELETE FROM framework."InboxMessages" WHERE "Id" = @event; + DELETE FROM framework."OutboxMessages" WHERE "Id" = @event; + """, connection); + command.Parameters.AddWithValue("marker", $"dispatch-{integrationEvent.Id:N}"); + command.Parameters.AddWithValue("event", integrationEvent.Id); + await command.ExecuteNonQueryAsync(); + } + + private sealed record DatabaseSnapshot(long ConsumerRows, long ConsumerTenantRows, long InboxRows, + long InboxTenantRows, long OutboxRows, long ProcessedRows, long ClaimedRows, long Retries); + + public sealed class FileEventMarkerHandler(CatalogDbContext context) : IIntegrationEventHandler + { + public async Task HandleAsync(FileFinalizedIntegrationEvent integrationEvent, CancellationToken ct = default) + { + context.Brands.Add(Brand.Create($"dispatch-{integrationEvent.Id:N}", null, null)); + await context.SaveChangesAsync(ct).ConfigureAwait(false); + } + } +} \ No newline at end of file diff --git a/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs b/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs new file mode 100644 index 0000000000..a9a87f5431 --- /dev/null +++ b/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs @@ -0,0 +1,243 @@ +using System.Collections.Concurrent; +using Finbuckle.MultiTenant; +using Finbuckle.MultiTenant.Abstractions; +using FSH.Framework.Eventing.Abstractions; +using FSH.Framework.Eventing.Outbox; +using FSH.Framework.Eventing.Persistence; +using FSH.Framework.Persistence; +using FSH.Framework.Shared.Multitenancy; +using FSH.Framework.Shared.Persistence; +using FSH.Modules.Catalog.Data; +using Integration.Tests.Infrastructure; +using Microsoft.AspNetCore.Mvc.Testing; +using Microsoft.AspNetCore.TestHost; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Npgsql; +using Shouldly; +using Xunit; + +namespace Integration.Tests.Tests.Eventing; + +[Collection(FshCollectionDefinition.Name)] +public sealed class OutboxDedicatedTenantDispatchTests +{ + private readonly FshWebApplicationFactory _factory; + + public OutboxDedicatedTenantDispatchTests(FshWebApplicationFactory factory) + { + _factory = factory; + } + + [Fact] + public async Task Dispatch_Should_RouteHandlersByEventTenant_AndCompleteOutboxInSourceDatabase() + { + string defaultConnectionString = _factory.Services + .GetRequiredService>() + .Value.ConnectionString; + NpgsqlConnectionStringBuilder defaultBuilder = new(defaultConnectionString); + string defaultDatabaseName = defaultBuilder.Database + ?? throw new InvalidOperationException("The default PostgreSQL database name is missing."); + string suffix = Guid.CreateVersion7().ToString("N"); + string sourceDatabaseName = $"outbox_source_{suffix}"; + string targetDatabaseName = $"outbox_target_{suffix}"; + NpgsqlConnectionStringBuilder adminBuilder = new(defaultBuilder.ConnectionString) + { + Database = "postgres", + }; + NpgsqlConnectionStringBuilder sourceBuilder = new(defaultBuilder.ConnectionString) + { + Database = sourceDatabaseName, + }; + NpgsqlConnectionStringBuilder targetBuilder = new(defaultBuilder.ConnectionString) + { + Database = targetDatabaseName, + }; + + try + { + await CreateDatabaseAsync(adminBuilder.ConnectionString, sourceDatabaseName); + await CreateDatabaseAsync(adminBuilder.ConnectionString, targetDatabaseName); + + AppTenantInfo sourceTenant = NewTenant(sourceDatabaseName, sourceBuilder.ConnectionString); + AppTenantInfo targetTenant = NewTenant(targetDatabaseName, targetBuilder.ConnectionString); + await AddTenantsAsync(_factory.Services, sourceTenant, targetTenant); + await MigrateTenantDatabaseAsync(_factory.Services, sourceTenant); + await MigrateTenantDatabaseAsync(_factory.Services, targetTenant); + + DatabaseRoutingRecorder recorder = new(); + await using WebApplicationFactory app = CreateFactory(recorder); + DatabaseRoutingIntegrationEvent[] events = + [ + NewEvent(sourceTenant.Id), + NewEvent(targetTenant.Id), + NewEvent(null), + ]; + + using (IServiceScope writeScope = app.Services.CreateScope()) + { + writeScope.ServiceProvider.GetRequiredService() + .MultiTenantContext = new MultiTenantContext(sourceTenant); + IOutboxStore outbox = writeScope.ServiceProvider.GetRequiredService(); + + foreach (DatabaseRoutingIntegrationEvent @event in events) + { + await outbox.AddAsync(@event, CancellationToken.None); + } + } + + EventingDrainTarget sourceTarget; + using (IServiceScope targetProviderScope = app.Services.CreateScope()) + { + var targets = await targetProviderScope.ServiceProvider + .GetRequiredService() + .GetTargetsAsync(CancellationToken.None); + sourceTarget = targets.Single(target => + string.Equals(target.ConnectionString, sourceBuilder.ConnectionString, StringComparison.Ordinal)); + } + + IEventingDrainScope drainScope = app.Services.GetRequiredService(); + using (drainScope.Begin(sourceTarget)) + { + using (IServiceScope dispatchScope = app.Services.CreateScope()) + { + EventingDbContext sourceOutbox = dispatchScope.ServiceProvider.GetRequiredService(); + sourceOutbox.Database.GetDbConnection().Database.ShouldBe(sourceDatabaseName); + + await dispatchScope.ServiceProvider + .GetRequiredService() + .DispatchAsync(CancellationToken.None); + } + + using IServiceScope verifyScope = app.Services.CreateScope(); + EventingDbContext verifyOutbox = verifyScope.ServiceProvider.GetRequiredService(); + List messages = await verifyOutbox.OutboxMessages + .AsNoTracking() + .Where(message => events.Select(@event => @event.Id).Contains(message.Id)) + .ToListAsync(); + + messages.Count.ShouldBe(events.Length); + messages.ShouldAllBe(message => message.ProcessedOnUtc != null && !message.IsDead); + } + + recorder.Count.ShouldBe(events.Length); + recorder.DatabaseFor(events[0].Id).ShouldBe(sourceDatabaseName); + recorder.DatabaseFor(events[1].Id).ShouldBe(targetDatabaseName); + recorder.DatabaseFor(events[2].Id).ShouldBe(defaultDatabaseName); + } + finally + { + await DropDatabaseAsync(adminBuilder.ConnectionString, targetDatabaseName); + await DropDatabaseAsync(adminBuilder.ConnectionString, sourceDatabaseName); + } + } + + private WebApplicationFactory CreateFactory(DatabaseRoutingRecorder recorder) => + _factory.WithWebHostBuilder(builder => + { + builder.ConfigureTestServices(services => + { + services.AddSingleton(recorder); + services.AddScoped< + IIntegrationEventHandler, + DatabaseRoutingHandler>(); + }); + }); + + private static async Task AddTenantsAsync(IServiceProvider services, params AppTenantInfo[] tenants) + { + using IServiceScope scope = services.CreateScope(); + IMultiTenantStore store = scope.ServiceProvider.GetRequiredService>(); + + foreach (AppTenantInfo tenant in tenants) + { + await store.AddAsync(tenant); + } + } + + private static async Task MigrateTenantDatabaseAsync(IServiceProvider services, AppTenantInfo tenant) + { + using IServiceScope scope = services.CreateScope(); + IServiceProvider provider = scope.ServiceProvider; + provider.GetRequiredService() + .MultiTenantContext = new MultiTenantContext(tenant); + + foreach (IDbInitializer initializer in provider.GetServices()) + { + await initializer.MigrateAsync(CancellationToken.None); + } + } + + private static AppTenantInfo NewTenant(string id, string connectionString) => + new(id, id, connectionString, "outbox-test@example.test") + { + IsActive = true, + }; + + private static DatabaseRoutingIntegrationEvent NewEvent(string? tenantId) + { + Guid id = Guid.CreateVersion7(); + return new DatabaseRoutingIntegrationEvent( + id, + DateTime.UtcNow, + tenantId, + $"outbox-routing-{id:N}", + "Integration.Tests"); + } + + private static async Task CreateDatabaseAsync(string adminConnectionString, string databaseName) + { + await using NpgsqlConnection connection = new(adminConnectionString); + await connection.OpenAsync(); + await using NpgsqlCommand command = connection.CreateCommand(); + using NpgsqlCommandBuilder commandBuilder = new(); + #pragma warning disable CA2100 // The test database name is generated locally and identifier-quoted. + command.CommandText = $"CREATE DATABASE {commandBuilder.QuoteIdentifier(databaseName)}"; + #pragma warning restore CA2100 + await command.ExecuteNonQueryAsync(); + } + + private static async Task DropDatabaseAsync(string adminConnectionString, string databaseName) + { + await using NpgsqlConnection connection = new(adminConnectionString); + await connection.OpenAsync(); + await using NpgsqlCommand command = connection.CreateCommand(); + using NpgsqlCommandBuilder commandBuilder = new(); + #pragma warning disable CA2100 // The test database name is generated locally and identifier-quoted. + command.CommandText = $"DROP DATABASE IF EXISTS {commandBuilder.QuoteIdentifier(databaseName)} WITH (FORCE)"; + #pragma warning restore CA2100 + await command.ExecuteNonQueryAsync(); + } + + public sealed record DatabaseRoutingIntegrationEvent( + Guid Id, + DateTime OccurredOnUtc, + string? TenantId, + string CorrelationId, + string Source) : IIntegrationEvent; + + public sealed class DatabaseRoutingHandler( + CatalogDbContext catalogDbContext, + DatabaseRoutingRecorder recorder) + : IIntegrationEventHandler + { + public async Task HandleAsync(DatabaseRoutingIntegrationEvent @event, CancellationToken ct = default) + { + (await catalogDbContext.Database.CanConnectAsync(ct).ConfigureAwait(false)).ShouldBeTrue(); + recorder.Record(@event.Id, catalogDbContext.Database.GetDbConnection().Database); + } + } + + public sealed class DatabaseRoutingRecorder + { + private readonly ConcurrentDictionary _databaseNames = new(); + + public int Count => _databaseNames.Count; + + public void Record(Guid eventId, string databaseName) => _databaseNames[eventId] = databaseName; + + public string? DatabaseFor(Guid eventId) => + _databaseNames.TryGetValue(eventId, out string? databaseName) ? databaseName : null; + } +} \ No newline at end of file diff --git a/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs b/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs index a01c9aeca1..5acdf4958a 100644 --- a/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs +++ b/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs @@ -1,7 +1,10 @@ using Finbuckle.MultiTenant; using Finbuckle.MultiTenant.Abstractions; +using Finbuckle.MultiTenant.Extensions; +using FSH.Framework.Eventing.Abstractions; using FSH.Framework.Shared.Multitenancy; using FSH.Modules.Multitenancy.Services; +using Microsoft.Extensions.DependencyInjection; using Shouldly; using Xunit; @@ -11,6 +14,32 @@ public sealed class FinbuckleEventTenantScopeTests { private readonly StubAccessor _accessor = new(); + [Fact] + public void Begin_Should_PreserveDedicatedConnection_When_EventTenantMatchesDrainTenant() + { + const string connectionString = "Host=localhost;Database=tenant_acme;Username=postgres;Password=postgres"; + var services = new ServiceCollection(); + services.AddMultiTenant(); + using ServiceProvider provider = services.BuildServiceProvider(); + var accessor = provider.GetRequiredService>(); + var setter = provider.GetRequiredService(); + var drainScope = new FinbuckleEventingDrainScope(accessor, setter); + var sut = new FinbuckleEventTenantScope(accessor, setter); + + using (drainScope.Begin(new EventingDrainTarget("acme", connectionString))) + { + accessor.MultiTenantContext.TenantInfo.ShouldNotBeNull(); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBe(connectionString); + + using (sut.Begin("acme")) + { + accessor.MultiTenantContext.TenantInfo.ShouldNotBeNull(); + accessor.MultiTenantContext.TenantInfo.Id.ShouldBe("acme"); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBe(connectionString); + } + } + } + [Fact] public void Begin_Should_SetTenantContext_When_TenantIdProvided() { From 549931f37f350f84ff09c91ccae953c2a9f416b8 Mon Sep 17 00:00:00 2001 From: HiroshiYoshino Date: Tue, 6 Oct 2026 01:01:26 +0900 Subject: [PATCH 2/2] fix(eventing): resolve full tenant metadata before constructing consumers An Outbox drain target selects the source database, not the destination for every event it contains. A root renewal can originate in default while its tenant-aware consumers and Inbox belong in a dedicated database. Preserve this separation instead of inheriting the source connection or reconstructing an ID-only tenant for dispatch. Add ExecuteAsync with a default synchronous bridge. The Finbuckle implementation resolves full tenant metadata from the default catalog through a short-lived scoped resolver, disposes the lookup scope, and runs consumer construction inside the same async callback. Restore the previous ambient context on success, error, and cancellation. Keep source claim, completion, retry, and dead-letter updates in the original database. Explicitly reset default drain context. Stop Webhook fanout from overwriting resolved metadata; keep central Billing business data in default while its framework Inbox follows the event tenant. Preserve Begin and the original two-argument constructor. Same-ID Begin retains full ambient metadata; different-ID Begin remains legacy ID-only. Global async dispatch uses minimal root/default context and preserves a null event TenantId in Inbox. Reject unknown and inactive tenants before consumer construction; inactive-only dedicated databases are still not automatically drained. Cover real Finbuckle async propagation, construction order, cancellation, isolation, two-database routing, Webhook subscriptions, root HTTP renewal, and redelivery after source completion failure. Replace the supplemental test's extra host with a lightweight provider to avoid disposing shared Hangfire storage. Production Hangfire and its shared factory are unchanged. Update the internal eventing and Webhook guidance. The original Green tree passed all 2,067 solution tests (no failures or skips), including 797 integration and 55 architecture tests; the full build had no warnings or errors. An isolated ID-only mutation failed the primary regression and passed again after restoration. The earlier history cleanup removed only an unrelated .gitignore entry; validated production, test, and guidance files remain unchanged. No distributed transaction, exactly-once guarantee, schema migration, or automatic repair of previously misrouted data is introduced. External docs publication and deployment/data-recovery review remain separate. --- .agents/rules/eventing.md | 12 +- .agents/rules/modules/webhooks.md | 2 +- .../IEventTenantScope.cs | 20 + .../Eventing/InMemory/InMemoryEventBus.cs | 6 +- .../MultitenancyModule.cs | 1 + .../Services/EventTenantResolver.cs | 12 + .../Services/FinbuckleEventTenantScope.cs | 68 +- .../Services/FinbuckleEventingDrainScope.cs | 24 +- .../Services/IEventTenantResolver.cs | 8 + .../Services/WebhookFanoutHandler.cs | 75 +-- .../InMemoryEventBusTenantScopeTests.cs | 88 ++- .../DedicatedTenantOutboxRoutingTests.cs | 610 +++++++++++++++++- .../OutboxDedicatedTenantDispatchTests.cs | 45 +- .../FinbuckleEventTenantScopeTests.cs | 220 +++++++ .../Services/WebhookFanoutHandlerTests.cs | 33 + 15 files changed, 1123 insertions(+), 101 deletions(-) create mode 100644 src/Modules/Multitenancy/Modules.Multitenancy/Services/EventTenantResolver.cs create mode 100644 src/Modules/Multitenancy/Modules.Multitenancy/Services/IEventTenantResolver.cs diff --git a/.agents/rules/eventing.md b/.agents/rules/eventing.md index aa7cdc27d7..901efbcfd8 100644 --- a/.agents/rules/eventing.md +++ b/.agents/rules/eventing.md @@ -35,6 +35,14 @@ Because rows follow the tenant connection, the dispatcher can't just poll one da Defaults in BuildingBlocks are single-database (`SingleDatabaseDrainTargetProvider`, `NullEventingDrainScope`); the multitenancy module replaces them with `TenantStoreDrainTargetProvider` (default DB + one target per distinct **active** per-tenant connection string; tenants sharing a database collapse to one target) and `FinbuckleEventingDrainScope`. One unreachable tenant database is logged and skipped, not fatal to the cycle. +Source and consumer routing are separate. A default drain explicitly installs root with an empty connection string and restores the previous context. `InMemoryEventBus` calls `IEventTenantScope.ExecuteAsync` before constructing its consumer scope. The Finbuckle implementation resolves full tenant metadata from the default catalog through a short-lived scoped `EventTenantResolver`, then invokes the dispatch callback in the same async flow. Never return an asynchronously installed AsyncLocal context to a caller that constructs consumers afterwards. + +Global events execute in root/default; their Inbox retains the event's null `TenantId`. Unknown/deleted and inactive tenants fail before consumer construction; the original source Outbox retries and eventually dead-letters. Dedicated inactive databases are excluded from automatic drain targets, so their pending rows require operator review/reactivation; dead letters need explicit redrive. Do not silently send rejected events to default. + +The synchronous `Begin` and two-argument constructor remain compatibility bridges, not asynchronous catalog resolution APIs. `Begin(null/whitespace)` leaves context untouched, while `ExecuteAsync(null/whitespace, ...)` selects default. Matching synchronous scopes retain full ambient metadata and restore it on disposal; different IDs retain legacy identity-only behavior. Use `ExecuteAsync` for event dispatch. The interface's default method keeps existing synchronous/no-op implementations source-compatible. + +Central Billing business data remains in default even while its consumer runs for a dedicated tenant; its framework Inbox follows the event tenant. Consumer success/Inbox and source completion are separate commits. Redelivery after source completion failure is suppressed by the destination Inbox, but a failure between a business write and Inbox insertion is not guaranteed exactly-once. No distributed transaction or automatic repair of previously misrouted data is introduced. External docs/changelog and deployment/data-recovery guidance remain tracked by the finish issue. + ## Multi-instance safety `ClaimBatchAsync` leases rows with `FOR UPDATE SKIP LOCKED` in a single `UPDATE … RETURNING`, so several API instances partition a batch instead of all publishing the same message. `ClaimedUntilUtc` is an expiry (`OutboxClaimLeaseSeconds`, default 300), so a dispatcher that dies mid-batch has its rows recovered rather than stranded — raise it if a batch can take longer than the lease, or a second instance re-claims rows still in flight. Completing or failing a message releases the lease. Non-Postgres providers have no portable `SKIP LOCKED`: they fall back to an unclaimed read and log a warning that only one instance is safe. @@ -68,7 +76,7 @@ Bus = `EventingOptions.Provider`: `"RabbitMQ"` → `RabbitMqEventBus` (durable t ## Gotchas - **Renaming/moving an integration event type breaks deserialization** — the outbox stores the assembly-qualified type name; `Type.GetType()` returns null → the message dead-letters. Keep event type names/namespaces stable, or migrate dead rows. -- **Background handlers carry no HTTP/tenant context.** An open-generic or background handler that reads a tenant-filtered DbContext must restore Finbuckle context first via `IMultiTenantContextSetter` (see `WebhookFanoutHandler`, `modules/webhooks.md`). -- In-memory bus runs handlers **synchronously in the publisher's scope** — keep handler work minimal; exceptions surface to the originating request (relevant for Notifications consuming Chat events). Via the outbox that scope is the dispatcher's, not the request's. +- **Background handlers carry no HTTP context.** The in-memory bus resolves the full event tenant before constructing handlers/Inbox. `WebhookFanoutHandler` uses that context without ID-only replacement; standalone jobs must establish full tenant context before constructing their DbContexts (see `modules/webhooks.md`). +- The in-memory bus awaits handlers in a fresh consumer DI scope within the publisher's dispatch flow. Keep work minimal; exceptions surface to the caller. Via Outbox, failures update the original source database rather than the consumer's database. - Set `UseHostedServiceDispatcher=false` to drive the outbox via Hangfire instead of the hosted service. - A background publisher must set the tenant context **before** `AddAsync` — otherwise, with per-tenant databases, the row lands in the wrong one (see `TenantExpiryScanJob`). diff --git a/.agents/rules/modules/webhooks.md b/.agents/rules/modules/webhooks.md index 8bc3f95097..87917c1a9a 100644 --- a/.agents/rules/modules/webhooks.md +++ b/.agents/rules/modules/webhooks.md @@ -8,6 +8,6 @@ Tenant-scoped outbound webhook subscriptions with HMAC-signed delivery and retri ## Gotchas - **Fan-out is an open-generic handler** — `WebhookFanoutHandler` is registered as an open generic, so it handles **every** `IIntegrationEvent` with no per-event wiring. It skips events with null `TenantId` (subscriptions are tenant-only) and matches event-type name against each subscription's `EventsCsv` (`*` wildcard supported). -- **Restore tenant context in the background** — both the fan-out handler and `WebhookDispatchJob` set `IMultiTenantContext` in a fresh scope before reading the tenant-filtered DbContext (background pumps/Hangfire carry no HTTP context). This is the canonical pattern for any background reader of tenant data — see `eventing.md`, `jobs.md`. +- **Fan-out consumes the bus-resolved tenant** - `IEventTenantScope.ExecuteAsync` resolves full catalog metadata before handler/DbContext construction. Fan-out must not replace it with ID-only information; a mismatched ambient ID fails before querying subscriptions. Direct callers must satisfy the same construction precondition. Standalone `WebhookDispatchJob` establishes its own context; job routing and external HTTP delivery are separate from successful fan-out/Inbox recording. See `eventing.md`, `jobs.md`. - **HMAC signing** — `X-Webhook-Signature: sha256=` (`WebhookPayloadSigner.Sign`), plus `X-Webhook-Event` and `X-Webhook-Delivery-Id` headers. - **Delivery** — `WebhookDispatcher.EnqueueAsync` enqueues a Hangfire `WebhookDispatchJob` per subscription; `[AutomaticRetry(Attempts=4, DelaysInSeconds={30,120,600,3600})]`. Transient (5xx/408/429) throws to reschedule; permanent 4xx completes silently. Each attempt persists a `WebhookDelivery` row. The `"Webhooks"` HttpClient uses `AddHeroResilience` (see `resilience.md`). diff --git a/src/BuildingBlocks/Eventing.Abstractions/IEventTenantScope.cs b/src/BuildingBlocks/Eventing.Abstractions/IEventTenantScope.cs index acfe796a1e..9cc25a31fe 100644 --- a/src/BuildingBlocks/Eventing.Abstractions/IEventTenantScope.cs +++ b/src/BuildingBlocks/Eventing.Abstractions/IEventTenantScope.cs @@ -24,4 +24,24 @@ public interface IEventTenantScope /// ambient context unchanged (global, non-tenant-scoped events). /// IDisposable Begin(string? tenantId); + + /// + /// Resolves the dispatch context and runs within that same + /// asynchronous flow. Construct handlers and Inbox inside the action, not beforehand. + /// The default bridge uses ; multitenancy implementations can resolve + /// full catalog metadata and define global routing before invoking the action. + /// + async Task ExecuteAsync( + string? tenantId, + Func action, + CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(action); + ct.ThrowIfCancellationRequested(); + + using (Begin(tenantId)) + { + await action(ct).ConfigureAwait(false); + } + } } diff --git a/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs b/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs index 7066bcda88..9aeb3473fb 100644 --- a/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs +++ b/src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs @@ -61,7 +61,7 @@ private async Task PublishSingleAsync(IIntegrationEvent @event, CancellationToke // Set tenant context BEFORE resolving handlers — MultiTenantDbContext captures TenantInfo at // construction, so a late tenant NREs the query filter. This is what makes background publishers work. - using (_tenantScope.Begin(@event.TenantId)) + await _tenantScope.ExecuteAsync(@event.TenantId, async dispatchToken => { using var scope = _serviceProvider.CreateScope(); var provider = scope.ServiceProvider; @@ -77,9 +77,9 @@ private async Task PublishSingleAsync(IIntegrationEvent @event, CancellationToke foreach (var handler in handlers) { - await InvokeHandlerAsync(handler, dispatch.HandleMethod, eventType, @event, inbox, ct).ConfigureAwait(false); + await InvokeHandlerAsync(handler, dispatch.HandleMethod, eventType, @event, inbox, dispatchToken).ConfigureAwait(false); } - } + }, ct).ConfigureAwait(false); } private static object[] ResolveHandlers(IServiceProvider provider, Type handlerInterfaceType) diff --git a/src/Modules/Multitenancy/Modules.Multitenancy/MultitenancyModule.cs b/src/Modules/Multitenancy/Modules.Multitenancy/MultitenancyModule.cs index fb290123da..f9ce9bc220 100644 --- a/src/Modules/Multitenancy/Modules.Multitenancy/MultitenancyModule.cs +++ b/src/Modules/Multitenancy/Modules.Multitenancy/MultitenancyModule.cs @@ -68,6 +68,7 @@ public void ConfigureServices(IHostApplicationBuilder builder) Services.TenantInitialPasswordBuffer>(); builder.Services.AddHeroDbContext(); + builder.Services.AddScoped(); // Replace (not Add) the no-op event tenant scope with a Finbuckle-backed one so background // event dispatch establishes the tenant before tenant-filtered handler DbContexts are built. diff --git a/src/Modules/Multitenancy/Modules.Multitenancy/Services/EventTenantResolver.cs b/src/Modules/Multitenancy/Modules.Multitenancy/Services/EventTenantResolver.cs new file mode 100644 index 0000000000..fed2753104 --- /dev/null +++ b/src/Modules/Multitenancy/Modules.Multitenancy/Services/EventTenantResolver.cs @@ -0,0 +1,12 @@ +using FSH.Framework.Shared.Multitenancy; +using FSH.Modules.Multitenancy.Data; +using Microsoft.EntityFrameworkCore; + +namespace FSH.Modules.Multitenancy.Services; + +public sealed class EventTenantResolver(TenantDbContext context) : IEventTenantResolver +{ + public Task ResolveAsync(string tenantId, CancellationToken ct = default) + => context.Set().AsNoTracking() + .SingleOrDefaultAsync(tenant => tenant.Id == tenantId, ct); +} \ No newline at end of file diff --git a/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventTenantScope.cs b/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventTenantScope.cs index 5fb7646395..dcd6f814f9 100644 --- a/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventTenantScope.cs +++ b/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventTenantScope.cs @@ -2,6 +2,7 @@ using Finbuckle.MultiTenant.Abstractions; using FSH.Framework.Eventing.Abstractions; using FSH.Framework.Shared.Multitenancy; +using Microsoft.Extensions.DependencyInjection; namespace FSH.Modules.Multitenancy.Services; @@ -11,15 +12,14 @@ namespace FSH.Modules.Multitenancy.Services; /// handler DbContexts resolved afterwards capture a real TenantInfo instead of the /// null one a background scope would otherwise carry. /// -/// Mirrors the create-scope-then-set-tenant pattern already used by -/// WebhookDispatchJob / SqlAuditSink, generalized to the event pipeline. -/// Only tenant identity is set (sufficient for the row-level tenant query filter in the -/// shared-database model); per-tenant connection strings are not resolved here. +/// Async dispatch resolves the full tenant from the default catalog before constructing +/// consumers. The callback keeps that context in the same asynchronous execution flow. /// public sealed class FinbuckleEventTenantScope : IEventTenantScope { private readonly IMultiTenantContextAccessor _accessor; private readonly IMultiTenantContextSetter _setter; + private readonly IServiceScopeFactory? _scopeFactory; public FinbuckleEventTenantScope( IMultiTenantContextAccessor accessor, @@ -29,6 +29,16 @@ public FinbuckleEventTenantScope( _setter = setter; } + public FinbuckleEventTenantScope( + IMultiTenantContextAccessor accessor, + IMultiTenantContextSetter setter, + IServiceScopeFactory scopeFactory) + : this(accessor, setter) + { + ArgumentNullException.ThrowIfNull(scopeFactory); + _scopeFactory = scopeFactory; + } + public IDisposable Begin(string? tenantId) { if (string.IsNullOrWhiteSpace(tenantId)) @@ -38,12 +48,62 @@ public IDisposable Begin(string? tenantId) } var previous = _accessor.MultiTenantContext; + if (string.Equals(previous.TenantInfo?.Id, tenantId, StringComparison.Ordinal)) + { + return new RestoreScope(_setter, previous); + } + _setter.MultiTenantContext = new MultiTenantContext(new AppTenantInfo(tenantId, tenantId)); return new RestoreScope(_setter, previous); } + public async Task ExecuteAsync( + string? tenantId, + Func action, + CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(action); + ct.ThrowIfCancellationRequested(); + var previous = _accessor.MultiTenantContext; + + try + { + var root = new AppTenantInfo(MultitenancyConstants.Root.Id, MultitenancyConstants.Root.Id) + { + IsActive = true + }; + _setter.MultiTenantContext = new MultiTenantContext(root); + + AppTenantInfo tenant = root; + if (!string.IsNullOrWhiteSpace(tenantId)) + { + if (_scopeFactory is null) + { + throw new InvalidOperationException("Async event dispatch requires a tenant catalog scope."); + } + + await using var scope = _scopeFactory.CreateAsyncScope(); + tenant = await scope.ServiceProvider.GetRequiredService() + .ResolveAsync(tenantId, ct).ConfigureAwait(false) + ?? throw new InvalidOperationException($"Event tenant '{tenantId}' is not registered."); + ct.ThrowIfCancellationRequested(); + if (!tenant.IsActive) + { + throw new InvalidOperationException($"Event tenant '{tenantId}' is inactive."); + } + } + + _setter.MultiTenantContext = new MultiTenantContext(tenant); + await action(ct).ConfigureAwait(false); + } + finally + { + _setter.MultiTenantContext = previous; + } + } + private sealed class RestoreScope : IDisposable { private readonly IMultiTenantContextSetter _setter; diff --git a/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventingDrainScope.cs b/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventingDrainScope.cs index 75b8a64207..1044766560 100644 --- a/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventingDrainScope.cs +++ b/src/Modules/Multitenancy/Modules.Multitenancy/Services/FinbuckleEventingDrainScope.cs @@ -9,9 +9,8 @@ namespace FSH.Modules.Multitenancy.Services; /// Installs an carrying the tenant's connection string, so an /// EventingDbContext built inside the scope routes to that tenant's database. /// -/// Deliberately distinct from , which sets tenant identity -/// only: that scope wraps handler dispatch, where the row-level tenant filter is what matters. -/// This one wraps a drain pass, where the target database is what matters. +/// Distinct from , which resolves the event's consumer +/// tenant. This scope selects the source database for a drain pass, independently of the consumer. /// public sealed class FinbuckleEventingDrainScope : IEventingDrainScope { @@ -30,19 +29,13 @@ public IDisposable Begin(EventingDrainTarget target) { ArgumentNullException.ThrowIfNull(target); - if (target.TenantId is null && target.ConnectionString is null) - { - // Default pass: leave the ambient context alone so the context falls through to the - // configured default connection. - return NoopScope.Instance; - } - var previous = _accessor.MultiTenantContext; // Built by hand rather than via the tenant-shaped constructor: only the id and the // connection string matter for routing, and the richer constructor also stamps validity // and activation state we have no business inventing here. - var info = new AppTenantInfo(target.TenantId!, target.TenantId!) + string tenantId = target.TenantId ?? MultitenancyConstants.Root.Id; + var info = new AppTenantInfo(tenantId, tenantId) { ConnectionString = target.ConnectionString ?? string.Empty, }; @@ -66,13 +59,4 @@ public RestoreScope(IMultiTenantContextSetter setter, IMultiTenantContext _setter.MultiTenantContext = _previous; } - private sealed class NoopScope : IDisposable - { - public static readonly NoopScope Instance = new(); - - public void Dispose() - { - // Nothing to restore — the ambient context was never touched. - } - } } diff --git a/src/Modules/Multitenancy/Modules.Multitenancy/Services/IEventTenantResolver.cs b/src/Modules/Multitenancy/Modules.Multitenancy/Services/IEventTenantResolver.cs new file mode 100644 index 0000000000..584f5bc7a3 --- /dev/null +++ b/src/Modules/Multitenancy/Modules.Multitenancy/Services/IEventTenantResolver.cs @@ -0,0 +1,8 @@ +using FSH.Framework.Shared.Multitenancy; + +namespace FSH.Modules.Multitenancy.Services; + +public interface IEventTenantResolver +{ + Task ResolveAsync(string tenantId, CancellationToken ct = default); +} \ No newline at end of file diff --git a/src/Modules/Webhooks/Modules.Webhooks/Services/WebhookFanoutHandler.cs b/src/Modules/Webhooks/Modules.Webhooks/Services/WebhookFanoutHandler.cs index 46806263cf..42b2840465 100644 --- a/src/Modules/Webhooks/Modules.Webhooks/Services/WebhookFanoutHandler.cs +++ b/src/Modules/Webhooks/Modules.Webhooks/Services/WebhookFanoutHandler.cs @@ -54,56 +54,47 @@ public async Task HandleAsync(TEvent @event, CancellationToken ct = default) return; } - // Install the tenant context for the subscription read — the WebhookDbContext Finbuckle filter needs - // it, and the background event pumps (OutboxDispatcher / event bus) carry no HTTP context. - var prev = _tenantContextAccessor.MultiTenantContext; - try + if (!string.Equals(_tenantContextAccessor.MultiTenantContext.TenantInfo?.Id, @event.TenantId, StringComparison.Ordinal)) { - var info = new AppTenantInfo(@event.TenantId, @event.TenantId); - ((IMultiTenantContextSetter)_tenantContextAccessor).MultiTenantContext = - new MultiTenantContext(info); + throw new InvalidOperationException("Webhook fan-out requires the event tenant to be resolved before handler construction."); + } + + var eventType = typeof(TEvent).Name; - var eventType = typeof(TEvent).Name; + // Pull active subscriptions, then match event type in memory: EventsCsv is a CSV blob (no join + // table), and there are typically 0–20 subscriptions per tenant so in-memory matching is fine. + var subscriptions = await _db.Subscriptions + .AsNoTracking() + .Where(s => s.IsActive) + .ToListAsync(ct) + .ConfigureAwait(false); - // Pull active subscriptions, then match event type in memory: EventsCsv is a CSV blob (no join - // table), and there are typically 0–20 subscriptions per tenant so in-memory matching is fine. - var subscriptions = await _db.Subscriptions - .AsNoTracking() - .Where(s => s.IsActive) - .ToListAsync(ct) - .ConfigureAwait(false); + var matching = subscriptions.Where(s => s.MatchesEvent(eventType)).ToList(); + if (matching.Count == 0) + { + return; + } - var matching = subscriptions.Where(s => s.MatchesEvent(eventType)).ToList(); - if (matching.Count == 0) + var payload = _serializer.Serialize(@event); + foreach (var subscription in matching) + { + try { - return; + await _dispatcher + .EnqueueAsync(@event.TenantId, subscription.Id, eventType, payload, ct) + .ConfigureAwait(false); } - - var payload = _serializer.Serialize(@event); - foreach (var subscription in matching) + catch (Exception ex) when (ex is not OperationCanceledException) { - try - { - await _dispatcher - .EnqueueAsync(@event.TenantId, subscription.Id, eventType, payload, ct) - .ConfigureAwait(false); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - // One bad subscription must not abort fan-out to others; this catches synchronous - // enqueue-side failures (Hangfire transient errors etc), not delivery (the job retries). - _logger.LogWarning( - ex, - "Failed to enqueue webhook delivery for subscription {SubscriptionId} (tenant {TenantId}, event {EventType})", - subscription.Id, - @event.TenantId, - eventType); - } + // One bad subscription must not abort fan-out to others; this catches synchronous + // enqueue-side failures (Hangfire transient errors etc), not delivery (the job retries). + _logger.LogWarning( + ex, + "Failed to enqueue webhook delivery for subscription {SubscriptionId} (tenant {TenantId}, event {EventType})", + subscription.Id, + @event.TenantId, + eventType); } } - finally - { - ((IMultiTenantContextSetter)_tenantContextAccessor).MultiTenantContext = prev; - } } } diff --git a/src/Tests/Framework.Tests/Eventing/InMemoryEventBusTenantScopeTests.cs b/src/Tests/Framework.Tests/Eventing/InMemoryEventBusTenantScopeTests.cs index 61222f3b2a..1850f5ba7f 100644 --- a/src/Tests/Framework.Tests/Eventing/InMemoryEventBusTenantScopeTests.cs +++ b/src/Tests/Framework.Tests/Eventing/InMemoryEventBusTenantScopeTests.cs @@ -1,7 +1,9 @@ using FSH.Framework.Eventing.Abstractions; using FSH.Framework.Eventing.InMemory; +using FSH.Framework.Eventing.Inbox; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; namespace Framework.Tests.Eventing; @@ -37,8 +39,89 @@ public async Task PublishAsync_Should_BeginTenantScope_WithEventTenantId_WhileHa scope.IsActive.ShouldBeFalse("the scope must be disposed once dispatch completes"); } + [Fact] + public async Task PublishAsync_Should_ConstructHandlerAndInboxOnlyAfterAsyncLookup() + { + var lookup = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var recording = new RecordingTenantScope(); + var tenantScope = new DelayedTenantScope(recording, lookup.Task); + int handlerConstructions = 0; + int inboxConstructions = 0; + var inbox = Substitute.For(); + TenantProbingHandler? handler = null; + var services = new ServiceCollection(); + services.AddScoped>(_ => + { + recording.IsActive.ShouldBeTrue(); + recording.BegunWith.ShouldHaveSingleItem().ShouldBe("acme"); + handlerConstructions++; + handler = new TenantProbingHandler(recording); + return handler; + }); + services.AddScoped(_ => + { + recording.IsActive.ShouldBeTrue(); + inboxConstructions++; + return inbox; + }); + using var provider = services.BuildServiceProvider(); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, tenantScope); + var integrationEvent = new TenantScopedEvent("acme"); + using var cancellation = new CancellationTokenSource(); + + Task dispatch = bus.PublishAsync(integrationEvent, cancellation.Token); + dispatch.IsCompleted.ShouldBeFalse(); + handlerConstructions.ShouldBe(0); + inboxConstructions.ShouldBe(0); + lookup.SetResult(); + await dispatch; + + handlerConstructions.ShouldBe(1); + inboxConstructions.ShouldBe(1); + handler.ShouldNotBeNull().ScopeWasActiveDuringHandle.ShouldBeTrue(); + recording.IsActive.ShouldBeFalse(); + await inbox.Received(1).MarkProcessedAsync(integrationEvent.Id, typeof(TenantProbingHandler).FullName!, + "acme", typeof(TenantScopedEvent).AssemblyQualifiedName!, cancellation.Token); + } + + [Fact] + public async Task PublishAsync_Should_NotConstructConsumers_When_DefaultBridgeIsCancelled() + { + var recording = new RecordingTenantScope(); + int constructions = 0; + var services = new ServiceCollection(); + services.AddScoped>(_ => + { + constructions++; + return new TenantProbingHandler(recording); + }); + using var provider = services.BuildServiceProvider(); + var bus = new InMemoryEventBus(provider, NullLogger.Instance, recording); + using var cancellation = new CancellationTokenSource(); + await cancellation.CancelAsync(); + + await Should.ThrowAsync(() => bus.PublishAsync(new TenantScopedEvent("acme"), cancellation.Token)); + + constructions.ShouldBe(0); + recording.BegunWith.ShouldBeEmpty(); + } + #region Test doubles + private sealed class DelayedTenantScope(RecordingTenantScope inner, Task lookup) : IEventTenantScope + { + public IDisposable Begin(string? tenantId) => throw new InvalidOperationException("Use async dispatch."); + + public async Task ExecuteAsync(string? tenantId, Func action, CancellationToken ct = default) + { + await lookup.WaitAsync(ct).ConfigureAwait(false); + using (inner.Begin(tenantId)) + { + await action(ct).ConfigureAwait(false); + } + } + } + private sealed record TenantScopedEvent(string? TenantId) : IIntegrationEvent { public Guid Id { get; } = Guid.CreateVersion7(); @@ -70,10 +153,11 @@ private sealed class TenantProbingHandler(RecordingTenantScope scope) { public bool ScopeWasActiveDuringHandle { get; private set; } - public Task HandleAsync(TenantScopedEvent @event, CancellationToken ct = default) + public async Task HandleAsync(TenantScopedEvent @event, CancellationToken ct = default) { + await Task.Yield(); + ct.ThrowIfCancellationRequested(); ScopeWasActiveDuringHandle = scope.IsActive; - return Task.CompletedTask; } } diff --git a/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs b/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs index caf5761e60..9ce9c0331a 100644 --- a/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs +++ b/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs @@ -3,21 +3,31 @@ using Finbuckle.MultiTenant.Extensions; using FSH.Framework.Eventing; using FSH.Framework.Eventing.Abstractions; +using FSH.Framework.Eventing.Inbox; using FSH.Framework.Eventing.Outbox; using FSH.Framework.Eventing.Persistence; using FSH.Framework.Persistence; using FSH.Framework.Shared.Multitenancy; using FSH.Modules.Catalog.Data; using FSH.Modules.Catalog.Domain; +using FSH.Modules.Billing.Data; +using FSH.Modules.Billing.Domain; +using FSH.Modules.Billing.Services; +using FSH.Modules.Billing.IntegrationEventHandlers; using FSH.Modules.Files.Contracts.Events; +using FSH.Modules.Multitenancy.Contracts.Events; using FSH.Modules.Multitenancy.Data; using FSH.Modules.Multitenancy.Services; +using FSH.Modules.Webhooks.Data; +using FSH.Modules.Webhooks.Domain; +using FSH.Modules.Webhooks.Services; using Integration.Tests.Infrastructure; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Options; using Npgsql; +using System.Collections.Concurrent; namespace Integration.Tests.Tests.Eventing; @@ -118,7 +128,7 @@ await context.OutboxMessages.Where(message => message.Id == eventId) await scope.ServiceProvider.GetRequiredService>().RemoveAsync(tenantId); } - await CleanupDefaultRowsAsync(defaultConnection, integrationEvent); + await CleanupDefaultRowsAsync(defaultConnection, integrationEvent.Id); if (databaseCreated) { await ExecuteDatabaseCommandAsync(defaultConnection, $"DROP DATABASE {QuoteIdentifier(databaseName)} WITH (FORCE)"); @@ -126,7 +136,444 @@ await context.OutboxMessages.Where(message => message.Id == eventId) } } - private ServiceProvider BuildProvider(string defaultConnection) + [Fact] + public async Task PublishAsync_Should_ConstructConsumersWithFullTenantOnlyAfterPendingCatalogLookup() + { + var lookup = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + AppTenantInfo? expected = null; + int handlerConstructions = 0; + int inboxConstructions = 0; + void ObserveTenant(IServiceProvider services) + { + AppTenantInfo tenant = services.GetRequiredService>() + .MultiTenantContext.TenantInfo.ShouldNotBeNull(); + tenant.Id.ShouldBe(expected.ShouldNotBeNull().Id); + tenant.ConnectionString.ShouldBe(expected.ConnectionString); + tenant.Name.ShouldBe("pending-catalog-name"); + tenant.Plan.ShouldBe("pro"); + } + + await using RoutingEnvironment environment = await CreateEnvironmentAsync(services => + { + services.Replace(ServiceDescriptor.Scoped(consumer => new PendingTenantResolver( + new EventTenantResolver(consumer.GetRequiredService()), + consumer.GetRequiredService>(), lookup.Task))); + services.Replace(ServiceDescriptor.Scoped>(consumer => + { + ObserveTenant(consumer); + handlerConstructions++; + return new ObservingFileHandler(new FileEventMarkerHandler(consumer.GetRequiredService()), + () => ObserveTenant(consumer)); + })); + services.Replace(ServiceDescriptor.Scoped(consumer => + { + ObserveTenant(consumer); + inboxConstructions++; + return ActivatorUtilities.CreateInstance(consumer); + })); + }); + expected = environment.DedicatedTenant; + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().Set() + .Where(tenant => tenant.Id == expected.Id).ExecuteUpdateAsync(update => update + .SetProperty(tenant => tenant.Name, "pending-catalog-name") + .SetProperty(tenant => tenant.Plan, "pro")); + } + var integrationEvent = NewFileEvent(expected.Id); + environment.Events.Add(integrationEvent); + using (environment.Drain.Begin(environment.DedicatedTarget)) + { + var previous = environment.Accessor.MultiTenantContext; + Task dispatch = environment.Provider.GetRequiredService().PublishAsync(integrationEvent); + dispatch.IsCompleted.ShouldBeFalse(); + handlerConstructions.ShouldBe(0); + inboxConstructions.ShouldBe(0); + environment.Accessor.MultiTenantContext.ShouldBeSameAs(previous); + lookup.SetResult(); + await dispatch; + environment.Accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + handlerConstructions.ShouldBe(1); + inboxConstructions.ShouldBe(1); + string handlerName = typeof(ObservingFileHandler).FullName!; + DatabaseSnapshot dedicated = await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent, handlerName); + dedicated.ConsumerTenantRows.ShouldBe(1); + dedicated.InboxTenantRows.ShouldBe(1); + DatabaseSnapshot shared = await ReadSnapshotAsync(environment.DefaultConnection, integrationEvent, handlerName); + shared.ConsumerRows.ShouldBe(0); + shared.InboxRows.ShouldBe(0); + } + + [Theory] + [InlineData("dedicated", true)] + [InlineData("shared", false)] + [InlineData("default", true)] + [InlineData("global", false)] + public async Task DispatchAsync_Should_SeparateSourceFromConsumerDatabase(string destination, bool defaultSource) + { + await using RoutingEnvironment environment = await CreateEnvironmentAsync(); + string? tenantId = destination switch + { + "dedicated" => environment.DedicatedTenant.Id, + "shared" => environment.SharedTenant.Id, + "default" => environment.DefaultTenant.Id, + _ => null + }; + var integrationEvent = NewFileEvent(tenantId); + environment.Events.Add(integrationEvent); + var source = defaultSource ? environment.DefaultTarget : environment.DedicatedTarget; + using (environment.Drain.Begin(source)) + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().AddAsync(integrationEvent); + var previous = environment.Accessor.MultiTenantContext; + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + environment.Accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + DatabaseSnapshot dedicated = await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent); + DatabaseSnapshot shared = await ReadSnapshotAsync(environment.DefaultConnection, integrationEvent); + bool dedicatedConsumer = destination is "dedicated" or "shared"; + dedicated.ConsumerRows.ShouldBe(dedicatedConsumer ? 1 : 0); + dedicated.ConsumerTenantRows.ShouldBe(dedicated.ConsumerRows); + dedicated.InboxRows.ShouldBe(dedicated.ConsumerRows); + dedicated.InboxTenantRows.ShouldBe(dedicated.ConsumerRows); + shared.ConsumerRows.ShouldBe(dedicatedConsumer ? 0 : 1); + shared.ConsumerTenantRows.ShouldBe(shared.ConsumerRows); + shared.InboxRows.ShouldBe(shared.ConsumerRows); + shared.InboxTenantRows.ShouldBe(shared.ConsumerRows); + dedicated.OutboxRows.ShouldBe(defaultSource ? 0 : 1); + dedicated.ProcessedRows.ShouldBe(dedicated.OutboxRows); + shared.OutboxRows.ShouldBe(defaultSource ? 1 : 0); + shared.ProcessedRows.ShouldBe(shared.OutboxRows); + dedicated.ClaimedRows.ShouldBe(0); + shared.ClaimedRows.ShouldBe(0); + dedicated.Retries.ShouldBe(0); + shared.Retries.ShouldBe(0); + using (environment.Drain.Begin(source)) + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().OutboxMessages + .Where(message => message.Id == integrationEvent.Id) + .ExecuteUpdateAsync(update => update.SetProperty(message => message.ProcessedOnUtc, (DateTime?)null)); + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + (await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent)).ShouldBe(dedicated); + (await ReadSnapshotAsync(environment.DefaultConnection, integrationEvent)).ShouldBe(shared); + } + + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(false, true)] + public async Task DispatchAsync_Should_RetryThenDeadLetterRejectedEvents_WithoutConsumerWrites(bool inactive, bool consumerFailure) + { + await using RoutingEnvironment environment = await CreateEnvironmentAsync(services => + { + if (consumerFailure) + { + services.RemoveAll>(); + services.AddScoped, FailingFileHandler>(); + } + }); + string tenantId = inactive || consumerFailure ? environment.DedicatedTenant.Id : $"unknown-{Guid.CreateVersion7():N}"; + if (inactive) + { + using var scope = environment.Provider.CreateScope(); + await scope.ServiceProvider.GetRequiredService().Set() + .Where(tenant => tenant.Id == tenantId).ExecuteUpdateAsync(update => update.SetProperty(tenant => tenant.IsActive, false)); + } + var integrationEvent = NewFileEvent(tenantId); + environment.Events.Add(integrationEvent); + environment.Provider.GetRequiredService>().Value.OutboxMaxRetries = 2; + + using (environment.Drain.Begin(environment.DedicatedTarget)) + { + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().AddAsync(integrationEvent); + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + + using (var scope = environment.Provider.CreateScope()) + { + var context = scope.ServiceProvider.GetRequiredService(); + OutboxMessage message = await context.OutboxMessages.SingleAsync(message => message.Id == integrationEvent.Id); + message.ProcessedOnUtc.ShouldBeNull(); + message.RetryCount.ShouldBe(1); + message.NextRetryAt.ShouldNotBeNull(); + message.ClaimedUntilUtc.ShouldBeNull(); + message.IsDead.ShouldBeFalse(); + message.NextRetryAt = null; + await context.SaveChangesAsync(); + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + + using var verify = environment.Provider.CreateScope(); + OutboxMessage failed = await verify.ServiceProvider.GetRequiredService() + .OutboxMessages.SingleAsync(message => message.Id == integrationEvent.Id); + failed.ProcessedOnUtc.ShouldBeNull(); + failed.RetryCount.ShouldBe(2); + failed.IsDead.ShouldBeTrue(); + failed.NextRetryAt.ShouldBeNull(); + failed.ClaimedUntilUtc.ShouldBeNull(); + } + + DatabaseSnapshot dedicated = await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent); + DatabaseSnapshot shared = await ReadSnapshotAsync(environment.DefaultConnection, integrationEvent); + dedicated.ConsumerRows.ShouldBe(0); + dedicated.InboxRows.ShouldBe(0); + shared.ConsumerRows.ShouldBe(0); + shared.InboxRows.ShouldBe(0); + shared.OutboxRows.ShouldBe(0); + if (consumerFailure) + { + (await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent, typeof(FailingFileHandler).FullName!)).InboxRows.ShouldBe(0); + (await ReadSnapshotAsync(environment.DefaultConnection, integrationEvent, typeof(FailingFileHandler).FullName!)).InboxRows.ShouldBe(0); + } + } + + [Fact] + public async Task Begin_Should_SelectDefaultDrain_WithoutLeakingDedicatedConnection_AndRestoreSource() + { + await using RoutingEnvironment environment = await CreateEnvironmentAsync(); + var previous = environment.Accessor.MultiTenantContext; + using (environment.Drain.Begin(environment.DedicatedTarget)) + { + var dedicated = environment.Accessor.MultiTenantContext; + using (environment.Drain.Begin(new EventingDrainTarget(null, null))) + using (var scope = environment.Provider.CreateScope()) + { + environment.Accessor.MultiTenantContext.TenantInfo.ShouldNotBeNull().Id.ShouldBe(MultitenancyConstants.Root.Id); + environment.Accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBeEmpty(); + scope.ServiceProvider.GetRequiredService().Database.GetDbConnection().Database + .ShouldBe(new NpgsqlConnectionStringBuilder(environment.DefaultConnection).Database); + } + environment.Accessor.MultiTenantContext.ShouldBeSameAs(dedicated); + } + environment.Accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task DispatchAsync_Should_ReadDedicatedWebhookSubscriptions_AndDeduplicateRedelivery(bool global) + { + var recorder = new WebhookRecorder(); + await using RoutingEnvironment environment = await CreateEnvironmentAsync(services => AddWebhookServices(services, recorder)); + var integrationEvent = NewFileEvent(global ? null : environment.DedicatedTenant.Id); + environment.Events.Add(integrationEvent); + Guid correct = await SeedSubscriptionAsync(environment, environment.DedicatedTarget, nameof(FileFinalizedIntegrationEvent)); + Guid wrong = await SeedSubscriptionAsync(environment, + new EventingDrainTarget(environment.DedicatedTenant.Id, null), nameof(FileFinalizedIntegrationEvent)); + + using (environment.Drain.Begin(environment.DedicatedTarget)) + { + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().AddAsync(integrationEvent); + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + using (var scope = environment.Provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().OutboxMessages + .Where(message => message.Id == integrationEvent.Id) + .ExecuteUpdateAsync(update => update.SetProperty(message => message.ProcessedOnUtc, (DateTime?)null)); + await scope.ServiceProvider.GetRequiredService().DispatchAsync(); + } + } + + recorder.Records.Count.ShouldBe(global ? 0 : 1); + recorder.Records.Any(record => record.Subscription == wrong).ShouldBeFalse(); + if (!global) recorder.Records.Single().ShouldBe((environment.DedicatedTenant.Id, correct)); + string inboxConnection = global ? environment.DefaultConnection : environment.DedicatedConnection; + string wrongConnection = global ? environment.DedicatedConnection : environment.DefaultConnection; + string handler = typeof(WebhookFanoutHandler).FullName!; + (await ReadSnapshotAsync(inboxConnection, integrationEvent, handler)).InboxTenantRows.ShouldBe(1); + (await ReadSnapshotAsync(wrongConnection, integrationEvent, handler)).InboxRows.ShouldBe(0); + (await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent)).ProcessedRows.ShouldBe(1); + } + + [Fact] + public async Task DispatchAsync_Should_KeepRenewBillingInDefault_AndDeduplicateAfterSourceCompletionFailure() + { + var recorder = new WebhookRecorder(); + string defaultConnection = _factory.Services.GetRequiredService>().Value.ConnectionString; + await using RoutingEnvironment environment = await CreateEnvironmentAsync(services => + { + AddWebhookServices(services, recorder); + services.AddHeroDbContext(); + services.AddScoped(); + services.AddScoped(); + services.AddScoped, RenewMarkerHandler>(); + services.AddScoped, TenantRenewedIntegrationEventHandler>(); + }); + var plan = BillingPlan.Create($"routing-{Guid.CreateVersion7():N}", "Routing", "USD", 10m); + using (var scope = environment.Provider.CreateScope()) + { + var billing = scope.ServiceProvider.GetRequiredService(); + billing.Plans.Add(plan); + billing.Subscriptions.Add(Subscription.Create(environment.DedicatedTenant.Id, plan.Id, + DateTime.UtcNow, environment.DedicatedTenant.ValidUpto)); + await billing.SaveChangesAsync(); + await scope.ServiceProvider.GetRequiredService().Set() + .Where(tenant => tenant.Id == environment.DedicatedTenant.Id) + .ExecuteUpdateAsync(update => update.SetProperty(tenant => tenant.Plan, plan.Key)); + } + environment.PlanIds.Add(plan.Id); + Guid subscription = await SeedSubscriptionAsync(environment, environment.DedicatedTarget, nameof(TenantRenewedIntegrationEvent)); + Guid wrongSubscription = await SeedSubscriptionAsync(environment, + new EventingDrainTarget(environment.DedicatedTenant.Id, null), nameof(TenantRenewedIntegrationEvent)); + using HttpClient client = await new AuthHelper(_factory).CreateRootAdminClientAsync(); + using HttpResponseMessage response = await client.PostAsJsonAsync( + $"{TestConstants.TenantsBasePath}/{environment.DedicatedTenant.Id}/renew", + new { tenantId = environment.DedicatedTenant.Id, planKey = plan.Key }); + response.StatusCode.ShouldBe(HttpStatusCode.OK, await response.Content.ReadAsStringAsync()); + + TenantRenewedIntegrationEvent integrationEvent; + using (environment.Drain.Begin(environment.DefaultTarget)) + using (var scope = environment.Provider.CreateScope()) + { + OutboxMessage message = await scope.ServiceProvider.GetRequiredService().OutboxMessages + .SingleAsync(message => message.TenantId == environment.DedicatedTenant.Id + && message.Type.Contains(nameof(TenantRenewedIntegrationEvent))); + integrationEvent = (TenantRenewedIntegrationEvent)scope.ServiceProvider.GetRequiredService() + .Deserialize(message.Payload, message.Type)!; + environment.Events.Add(integrationEvent); + var outbox = scope.ServiceProvider.GetRequiredService(); + var failureStore = new CompletionFailingStore(outbox, + scope.ServiceProvider.GetRequiredService(), integrationEvent.Id, true); + await CreateDispatcher(scope.ServiceProvider, failureStore).DispatchAsync(); + } + + string markerHandler = typeof(RenewMarkerHandler).FullName!; + DatabaseSnapshot dedicated = await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent, markerHandler); + DatabaseSnapshot shared = await ReadSnapshotAsync(defaultConnection, integrationEvent, markerHandler); + dedicated.ConsumerRows.ShouldBe(1); + dedicated.ConsumerTenantRows.ShouldBe(1); + dedicated.InboxTenantRows.ShouldBe(1); + dedicated.OutboxRows.ShouldBe(0); + shared.ConsumerRows.ShouldBe(0); + shared.InboxRows.ShouldBe(0); + shared.OutboxRows.ShouldBe(1); + shared.ProcessedRows.ShouldBe(0); + shared.Retries.ShouldBe(1); + shared.ClaimedRows.ShouldBe(0); + string[] handlers = [markerHandler, typeof(TenantRenewedIntegrationEventHandler).FullName!, + typeof(WebhookFanoutHandler).FullName!]; + foreach (string handler in handlers) + { + (await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent, handler)).InboxTenantRows.ShouldBe(1); + (await ReadSnapshotAsync(defaultConnection, integrationEvent, handler)).InboxRows.ShouldBe(0); + } + recorder.Records.ShouldHaveSingleItem().ShouldBe((environment.DedicatedTenant.Id, subscription)); + recorder.Records.Any(record => record.Subscription == wrongSubscription).ShouldBeFalse(); + using (var scope = environment.Provider.CreateScope()) + { + var billing = scope.ServiceProvider.GetRequiredService(); + billing.Database.GetDbConnection().Database.ShouldBe(new NpgsqlConnectionStringBuilder(defaultConnection).Database); + (await billing.Subscriptions.CountAsync(row => row.TenantId == environment.DedicatedTenant.Id)).ShouldBe(1); + (await billing.Subscriptions.SingleAsync(row => row.TenantId == environment.DedicatedTenant.Id)) + .EndUtc.ShouldBe(integrationEvent.PeriodEndUtc); + (await billing.Invoices.CountAsync(row => row.TenantId == environment.DedicatedTenant.Id)).ShouldBe(1); + } + await using (var connection = new NpgsqlConnection(environment.DedicatedConnection)) + { + await connection.OpenAsync(); + await using var command = new NpgsqlCommand("SELECT to_regclass('billing.\"Invoices\"') IS NULL", connection); + (await command.ExecuteScalarAsync()).ShouldBe(true); + } + + using (environment.Drain.Begin(environment.DefaultTarget)) + using (var scope = environment.Provider.CreateScope()) + { + var context = scope.ServiceProvider.GetRequiredService(); + await context.OutboxMessages.Where(message => message.Id == integrationEvent.Id) + .ExecuteUpdateAsync(update => update.SetProperty(message => message.NextRetryAt, (DateTime?)null)); + var store = new CompletionFailingStore(scope.ServiceProvider.GetRequiredService(), + context, integrationEvent.Id, false); + await CreateDispatcher(scope.ServiceProvider, store).DispatchAsync(); + } + (await ReadSnapshotAsync(environment.DedicatedConnection, integrationEvent, markerHandler)).ShouldBe(dedicated); + DatabaseSnapshot completed = await ReadSnapshotAsync(defaultConnection, integrationEvent, markerHandler); + completed.ProcessedRows.ShouldBe(1); + completed.Retries.ShouldBe(1); + completed.ClaimedRows.ShouldBe(0); + recorder.Records.Count.ShouldBe(1); + using (var scope = environment.Provider.CreateScope()) + { + var billing = scope.ServiceProvider.GetRequiredService(); + (await billing.Subscriptions.CountAsync(row => row.TenantId == environment.DedicatedTenant.Id)).ShouldBe(1); + (await billing.Invoices.CountAsync(row => row.TenantId == environment.DedicatedTenant.Id)).ShouldBe(1); + } + } + + private static void AddWebhookServices(IServiceCollection services, WebhookRecorder recorder) + { + services.AddHeroDbContext(); + services.AddSingleton(recorder); + services.AddScoped(typeof(IIntegrationEventHandler<>), typeof(WebhookFanoutHandler<>)); + } + + private static async Task SeedSubscriptionAsync(RoutingEnvironment environment, EventingDrainTarget target, string eventType) + { + using (environment.Drain.Begin(target)) + using (var scope = environment.Provider.CreateScope()) + { + var context = scope.ServiceProvider.GetRequiredService(); + var subscription = WebhookSubscription.Create("https://routing.example.test/hook", [eventType], "test-hash"); + context.Subscriptions.Add(subscription); + await context.SaveChangesAsync(); + if (target.ConnectionString is null) environment.SubscriptionIds.Add(subscription.Id); + return subscription.Id; + } + } + + private async Task CreateEnvironmentAsync(Action? configure = null) + { + string connection = _factory.Services.GetRequiredService>().Value.ConnectionString; + string suffix = Guid.CreateVersion7().ToString("N"); + string database = $"routing_{suffix}"; + string dedicatedConnection = new NpgsqlConnectionStringBuilder(connection) { Database = database }.ConnectionString; + ServiceProvider provider = BuildProvider(connection, configure); + var environment = new RoutingEnvironment(provider, connection, dedicatedConnection, database, + new AppTenantInfo($"dedicated-{suffix}", $"dedicated-{suffix}", dedicatedConnection, "routing@example.test"), + new AppTenantInfo($"shared-{suffix}", $"shared-{suffix}", dedicatedConnection, "routing@example.test"), + new AppTenantInfo($"default-{suffix}", $"default-{suffix}", null, "routing@example.test")); + try + { + await ExecuteDatabaseCommandAsync(connection, $"CREATE DATABASE {QuoteIdentifier(database)}"); + environment.DatabaseCreated = true; + using (environment.Drain.Begin(environment.DedicatedTarget)) + using (var scope = provider.CreateScope()) + { + await scope.ServiceProvider.GetRequiredService().Database.MigrateAsync(); + await scope.ServiceProvider.GetRequiredService().Database.MigrateAsync(); + if (scope.ServiceProvider.GetService() is { } webhooks) + await webhooks.Database.MigrateAsync(); + } + using (var scope = provider.CreateScope()) + { + var store = scope.ServiceProvider.GetRequiredService>(); + (await store.AddAsync(environment.DedicatedTenant)).ShouldBeTrue(); + (await store.AddAsync(environment.SharedTenant)).ShouldBeTrue(); + (await store.AddAsync(environment.DefaultTenant)).ShouldBeTrue(); + } + return environment; + } + catch + { + await environment.DisposeAsync(); + throw; + } + } + + private ServiceProvider BuildProvider(string defaultConnection, Action? configure = null) + => CreateProvider(_factory.Services, defaultConnection, configure); + + internal static ServiceProvider CreateProvider(IServiceProvider hostServices, string defaultConnection, + Action? configure = null) { var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary { @@ -135,7 +582,7 @@ private ServiceProvider BuildProvider(string defaultConnection) }).Build(); var services = new ServiceCollection(); services.AddSingleton(configuration); - services.AddSingleton(_factory.Services.GetRequiredService()); + services.AddSingleton(hostServices.GetRequiredService()); services.AddSingleton(TimeProvider.System); services.AddLogging(); services.Configure(options => @@ -147,6 +594,7 @@ private ServiceProvider BuildProvider(string defaultConnection) services.AddScoped(); services.AddScoped(); services.AddDbContext(options => options.UseNpgsql(defaultConnection)); + services.AddScoped(); services.AddMultiTenant() .WithStore>(ServiceLifetime.Scoped); services.AddEventingCore(configuration); @@ -155,6 +603,7 @@ private ServiceProvider BuildProvider(string defaultConnection) services.Replace(ServiceDescriptor.Singleton()); services.Replace(ServiceDescriptor.Scoped()); services.AddScoped, FileEventMarkerHandler>(); + configure?.Invoke(services); return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true }); } @@ -174,7 +623,7 @@ private static async Task ExecuteDatabaseCommandAsync(string connectionString, s await command.ExecuteNonQueryAsync(); } - private static async Task ReadSnapshotAsync(string connectionString, FileFinalizedIntegrationEvent integrationEvent) + private static async Task ReadSnapshotAsync(string connectionString, IIntegrationEvent integrationEvent, string? handlerName = null) { await using var connection = new NpgsqlConnection(connectionString); await connection.OpenAsync(); @@ -183,23 +632,24 @@ private static async Task ReadSnapshotAsync(string connectionS (SELECT COUNT(*) FROM catalog."Brands" WHERE "Name" = @marker), (SELECT COUNT(*) FROM catalog."Brands" WHERE "Name" = @marker AND "TenantId" = @tenant), (SELECT COUNT(*) FROM framework."InboxMessages" WHERE "Id" = @event AND "HandlerName" = @handler), - (SELECT COUNT(*) FROM framework."InboxMessages" WHERE "Id" = @event AND "HandlerName" = @handler AND "TenantId" = @tenant), + (SELECT COUNT(*) FROM framework."InboxMessages" WHERE "Id" = @event AND "HandlerName" = @handler AND "TenantId" IS NOT DISTINCT FROM @inboxTenant), (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event), (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event AND "ProcessedOnUtc" IS NOT NULL), (SELECT COUNT(*) FROM framework."OutboxMessages" WHERE "Id" = @event AND "ClaimedUntilUtc" IS NOT NULL), (SELECT COALESCE(SUM("RetryCount"), 0) FROM framework."OutboxMessages" WHERE "Id" = @event) """, connection); command.Parameters.AddWithValue("marker", $"dispatch-{integrationEvent.Id:N}"); - command.Parameters.AddWithValue("tenant", integrationEvent.TenantId!); + command.Parameters.AddWithValue("tenant", integrationEvent.TenantId ?? MultitenancyConstants.Root.Id); + command.Parameters.AddWithValue("inboxTenant", NpgsqlTypes.NpgsqlDbType.Text, (object?)integrationEvent.TenantId ?? DBNull.Value); command.Parameters.AddWithValue("event", integrationEvent.Id); - command.Parameters.AddWithValue("handler", typeof(FileEventMarkerHandler).FullName!); + command.Parameters.AddWithValue("handler", handlerName ?? typeof(FileEventMarkerHandler).FullName!); await using var reader = await command.ExecuteReaderAsync(); (await reader.ReadAsync()).ShouldBeTrue(); return new DatabaseSnapshot(reader.GetInt64(0), reader.GetInt64(1), reader.GetInt64(2), reader.GetInt64(3), reader.GetInt64(4), reader.GetInt64(5), reader.GetInt64(6), reader.GetInt64(7)); } - private static async Task CleanupDefaultRowsAsync(string connectionString, FileFinalizedIntegrationEvent integrationEvent) + private static async Task CleanupDefaultRowsAsync(string connectionString, Guid eventId) { await using var connection = new NpgsqlConnection(connectionString); await connection.OpenAsync(); @@ -208,14 +658,96 @@ private static async Task CleanupDefaultRowsAsync(string connectionString, FileF DELETE FROM framework."InboxMessages" WHERE "Id" = @event; DELETE FROM framework."OutboxMessages" WHERE "Id" = @event; """, connection); - command.Parameters.AddWithValue("marker", $"dispatch-{integrationEvent.Id:N}"); - command.Parameters.AddWithValue("event", integrationEvent.Id); + command.Parameters.AddWithValue("marker", $"dispatch-{eventId:N}"); + command.Parameters.AddWithValue("event", eventId); await command.ExecuteNonQueryAsync(); } private sealed record DatabaseSnapshot(long ConsumerRows, long ConsumerTenantRows, long InboxRows, long InboxTenantRows, long OutboxRows, long ProcessedRows, long ClaimedRows, long Retries); + private static FileFinalizedIntegrationEvent NewFileEvent(string? tenantId) + { + Guid id = Guid.CreateVersion7(); + return new(id, DateTime.UtcNow, tenantId, $"corr-{id:N}", "Files", Guid.CreateVersion7(), "Test", null, + "application/octet-stream", 1, 1); + } + + private sealed class RoutingEnvironment( + ServiceProvider provider, string defaultConnection, string dedicatedConnection, string database, + AppTenantInfo dedicatedTenant, AppTenantInfo sharedTenant, AppTenantInfo defaultTenant) : IAsyncDisposable + { + public ServiceProvider Provider { get; } = provider; + public string DefaultConnection { get; } = defaultConnection; + public string DedicatedConnection { get; } = dedicatedConnection; + public AppTenantInfo DedicatedTenant { get; } = dedicatedTenant; + public AppTenantInfo SharedTenant { get; } = sharedTenant; + public AppTenantInfo DefaultTenant { get; } = defaultTenant; + public bool DatabaseCreated { get; set; } + public List Events { get; } = []; + public List SubscriptionIds { get; } = []; + public List PlanIds { get; } = []; + public IMultiTenantContextAccessor Accessor => Provider.GetRequiredService>(); + public IEventingDrainScope Drain => Provider.GetRequiredService(); + public EventingDrainTarget DefaultTarget { get; } = new(MultitenancyConstants.Root.Id, null); + public EventingDrainTarget DedicatedTarget => new(DedicatedTenant.Id, DedicatedConnection); + + public async ValueTask DisposeAsync() + { + try + { + using var cleanupTenant = Drain.Begin(DefaultTarget); + using var scope = Provider.CreateScope(); + string[] tenants = [DedicatedTenant.Id, SharedTenant.Id, DefaultTenant.Id]; + await scope.ServiceProvider.GetRequiredService().Set() + .Where(tenant => tenants.Contains(tenant.Id)).ExecuteDeleteAsync(); + if (scope.ServiceProvider.GetService() is { } webhooks) + await webhooks.Subscriptions.IgnoreQueryFilters() + .Where(subscription => SubscriptionIds.Contains(subscription.Id)).ExecuteDeleteAsync(); + if (scope.ServiceProvider.GetService() is { } billing) + { + await billing.Invoices.Where(invoice => tenants.Contains(invoice.TenantId)).ExecuteDeleteAsync(); + await billing.Subscriptions.Where(subscription => tenants.Contains(subscription.TenantId)).ExecuteDeleteAsync(); + await billing.Plans.Where(plan => PlanIds.Contains(plan.Id)).ExecuteDeleteAsync(); + } + foreach (IIntegrationEvent integrationEvent in Events) + await CleanupDefaultRowsAsync(DefaultConnection, integrationEvent.Id); + } + finally + { + await Provider.DisposeAsync(); + if (DatabaseCreated) + await ExecuteDatabaseCommandAsync(DefaultConnection, $"DROP DATABASE {QuoteIdentifier(database)} WITH (FORCE)"); + } + } + } + + private sealed class PendingTenantResolver( + IEventTenantResolver inner, IMultiTenantContextAccessor accessor, Task lookup) : IEventTenantResolver + { + public async Task ResolveAsync(string tenantId, CancellationToken ct = default) + { + accessor.MultiTenantContext.TenantInfo!.Id.ShouldBe(MultitenancyConstants.Root.Id); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBeEmpty(); + await lookup.WaitAsync(ct).ConfigureAwait(false); + accessor.MultiTenantContext.TenantInfo.Id.ShouldBe(MultitenancyConstants.Root.Id); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBeEmpty(); + return await inner.ResolveAsync(tenantId, ct).ConfigureAwait(false); + } + } + + private sealed class ObservingFileHandler(FileEventMarkerHandler inner, Action observe) + : IIntegrationEventHandler + { + public async Task HandleAsync(FileFinalizedIntegrationEvent @event, CancellationToken ct = default) + { + observe(); + await Task.Yield(); + observe(); + await inner.HandleAsync(@event, ct).ConfigureAwait(false); + } + } + public sealed class FileEventMarkerHandler(CatalogDbContext context) : IIntegrationEventHandler { public async Task HandleAsync(FileFinalizedIntegrationEvent integrationEvent, CancellationToken ct = default) @@ -224,4 +756,62 @@ public async Task HandleAsync(FileFinalizedIntegrationEvent integrationEvent, Ca await context.SaveChangesAsync(ct).ConfigureAwait(false); } } + + public sealed class RenewMarkerHandler(CatalogDbContext context) : IIntegrationEventHandler + { + public async Task HandleAsync(TenantRenewedIntegrationEvent integrationEvent, CancellationToken ct = default) + { + context.Brands.Add(Brand.Create($"dispatch-{integrationEvent.Id:N}", null, null)); + await context.SaveChangesAsync(ct).ConfigureAwait(false); + } + } + + public sealed class FailingFileHandler : IIntegrationEventHandler + { + public Task HandleAsync(FileFinalizedIntegrationEvent @event, CancellationToken ct = default) + => throw new InvalidOperationException("Injected consumer failure."); + } + + private sealed class WebhookRecorder : IWebhookDispatcher + { + public ConcurrentQueue<(string Tenant, Guid Subscription)> Records { get; } = new(); + + public Task EnqueueAsync(string tenantId, Guid subscriptionId, string eventType, string payloadJson, CancellationToken cancellationToken = default) + { + Records.Enqueue((tenantId, subscriptionId)); + return Task.CompletedTask; + } + } + + private sealed class UnusedUsageReporter : IUsageReporter + { + public Task> CaptureForPeriodAsync(string tenantId, int periodYear, int periodMonth, + CancellationToken cancellationToken = default) => throw new InvalidOperationException("Renew does not capture metered usage."); + } + + private static OutboxDispatcher CreateDispatcher(IServiceProvider provider, IOutboxStore store) + => new(store, provider.GetRequiredService(), provider.GetRequiredService(), + provider.GetRequiredService>(), + provider.GetRequiredService>()); + + private sealed class CompletionFailingStore(IOutboxStore inner, EventingDbContext context, Guid eventId, bool failCompletion) : IOutboxStore + { + public Task AddAsync(IIntegrationEvent @event, CancellationToken ct = default) => inner.AddAsync(@event, ct); + public async Task> ClaimBatchAsync(int batchSize, string claimedBy, TimeSpan lease, CancellationToken ct = default) + { + IReadOnlyList messages = await inner.ClaimBatchAsync(batchSize, claimedBy, lease, ct).ConfigureAwait(false); + Guid[] unrelated = messages.Where(message => message.Id != eventId).Select(message => message.Id).ToArray(); + if (unrelated.Length != 0) + await context.OutboxMessages.Where(message => unrelated.Contains(message.Id) && message.ClaimedBy == claimedBy) + .ExecuteUpdateAsync(update => update.SetProperty(message => message.ClaimedBy, (string?)null) + .SetProperty(message => message.ClaimedUntilUtc, (DateTime?)null), ct).ConfigureAwait(false); + return messages.Where(message => message.Id == eventId).ToArray(); + } + public Task MarkAsProcessedAsync(OutboxMessage message, CancellationToken ct = default) + => failCompletion && message.Id == eventId ? throw new InvalidOperationException("Injected source completion failure.") : inner.MarkAsProcessedAsync(message, ct); + public Task MarkAsFailedAsync(OutboxMessage message, string error, bool isDead, CancellationToken ct = default) + => inner.MarkAsFailedAsync(message, error, isDead, ct); + public Task> GetDeadLetteredAsync(int max, CancellationToken ct = default) => inner.GetDeadLetteredAsync(max, ct); + public Task RedriveDeadLettersAsync(IReadOnlyCollection? ids, CancellationToken ct = default) => inner.RedriveDeadLettersAsync(ids, ct); + } } \ No newline at end of file diff --git a/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs b/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs index a9a87f5431..9fa313e494 100644 --- a/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs +++ b/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs @@ -9,8 +9,6 @@ using FSH.Framework.Shared.Persistence; using FSH.Modules.Catalog.Data; using Integration.Tests.Infrastructure; -using Microsoft.AspNetCore.Mvc.Testing; -using Microsoft.AspNetCore.TestHost; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; @@ -54,6 +52,9 @@ public async Task Dispatch_Should_RouteHandlersByEventTenant_AndCompleteOutboxIn { Database = targetDatabaseName, }; + var accessor = _factory.Services.GetRequiredService>(); + var previous = accessor.MultiTenantContext; + var eventIds = new List(); try { @@ -67,15 +68,16 @@ public async Task Dispatch_Should_RouteHandlersByEventTenant_AndCompleteOutboxIn await MigrateTenantDatabaseAsync(_factory.Services, targetTenant); DatabaseRoutingRecorder recorder = new(); - await using WebApplicationFactory app = CreateFactory(recorder); + await using ServiceProvider app = CreateProvider(defaultConnectionString, recorder); DatabaseRoutingIntegrationEvent[] events = [ NewEvent(sourceTenant.Id), NewEvent(targetTenant.Id), NewEvent(null), ]; + eventIds.AddRange(events.Select(integrationEvent => integrationEvent.Id)); - using (IServiceScope writeScope = app.Services.CreateScope()) + using (IServiceScope writeScope = app.CreateScope()) { writeScope.ServiceProvider.GetRequiredService() .MultiTenantContext = new MultiTenantContext(sourceTenant); @@ -88,7 +90,7 @@ public async Task Dispatch_Should_RouteHandlersByEventTenant_AndCompleteOutboxIn } EventingDrainTarget sourceTarget; - using (IServiceScope targetProviderScope = app.Services.CreateScope()) + using (IServiceScope targetProviderScope = app.CreateScope()) { var targets = await targetProviderScope.ServiceProvider .GetRequiredService() @@ -97,10 +99,10 @@ public async Task Dispatch_Should_RouteHandlersByEventTenant_AndCompleteOutboxIn string.Equals(target.ConnectionString, sourceBuilder.ConnectionString, StringComparison.Ordinal)); } - IEventingDrainScope drainScope = app.Services.GetRequiredService(); + IEventingDrainScope drainScope = app.GetRequiredService(); using (drainScope.Begin(sourceTarget)) { - using (IServiceScope dispatchScope = app.Services.CreateScope()) + using (IServiceScope dispatchScope = app.CreateScope()) { EventingDbContext sourceOutbox = dispatchScope.ServiceProvider.GetRequiredService(); sourceOutbox.Database.GetDbConnection().Database.ShouldBe(sourceDatabaseName); @@ -110,7 +112,7 @@ await dispatchScope.ServiceProvider .DispatchAsync(CancellationToken.None); } - using IServiceScope verifyScope = app.Services.CreateScope(); + using IServiceScope verifyScope = app.CreateScope(); EventingDbContext verifyOutbox = verifyScope.ServiceProvider.GetRequiredService(); List messages = await verifyOutbox.OutboxMessages .AsNoTracking() @@ -128,21 +130,30 @@ await dispatchScope.ServiceProvider } finally { + _factory.Services.GetRequiredService().MultiTenantContext = previous; + using (IServiceScope scope = _factory.Services.CreateScope()) + { + var store = scope.ServiceProvider.GetRequiredService>(); + await store.RemoveAsync(sourceDatabaseName); + await store.RemoveAsync(targetDatabaseName); + } + await using (var connection = new NpgsqlConnection(defaultConnectionString)) + { + await connection.OpenAsync(); + await using var command = new NpgsqlCommand("DELETE FROM framework.\"InboxMessages\" WHERE \"Id\" = ANY(@ids)", connection); + command.Parameters.AddWithValue("ids", eventIds.ToArray()); + await command.ExecuteNonQueryAsync(); + } await DropDatabaseAsync(adminBuilder.ConnectionString, targetDatabaseName); await DropDatabaseAsync(adminBuilder.ConnectionString, sourceDatabaseName); } } - private WebApplicationFactory CreateFactory(DatabaseRoutingRecorder recorder) => - _factory.WithWebHostBuilder(builder => + private ServiceProvider CreateProvider(string defaultConnection, DatabaseRoutingRecorder recorder) => + DedicatedTenantOutboxRoutingTests.CreateProvider(_factory.Services, defaultConnection, services => { - builder.ConfigureTestServices(services => - { - services.AddSingleton(recorder); - services.AddScoped< - IIntegrationEventHandler, - DatabaseRoutingHandler>(); - }); + services.AddSingleton(recorder); + services.AddScoped, DatabaseRoutingHandler>(); }); private static async Task AddTenantsAsync(IServiceProvider services, params AppTenantInfo[] tenants) diff --git a/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs b/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs index 5acdf4958a..2dcbcd583e 100644 --- a/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs +++ b/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs @@ -5,6 +5,7 @@ using FSH.Framework.Shared.Multitenancy; using FSH.Modules.Multitenancy.Services; using Microsoft.Extensions.DependencyInjection; +using NSubstitute; using Shouldly; using Xunit; @@ -88,6 +89,225 @@ public void Begin_Should_LeaveContextUntouched_When_TenantIdNullOrWhitespace() } } + [Fact] + public async Task ExecuteAsync_Should_KeepFullTenantAcrossAwaits_AndRestoreCaller() + { + var resolution = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var resolver = Substitute.For(); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var setter = provider.GetRequiredService(); + var previous = new MultiTenantContext(NewTenant("source", "source-db")); + setter.MultiTenantContext = previous; + var tenant = NewTenant("acme", "dedicated-db"); + var sut = provider.GetRequiredService(); + bool called = false; + resolver.ResolveAsync("acme", Arg.Any()).Returns(async call => + { + accessor.MultiTenantContext.TenantInfo!.Id.ShouldBe(MultitenancyConstants.Root.Id); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBeEmpty(); + AppTenantInfo? resolved = await resolution.Task.WaitAsync(call.Arg()).ConfigureAwait(false); + accessor.MultiTenantContext.TenantInfo.Id.ShouldBe(MultitenancyConstants.Root.Id); + return resolved; + }); + + Task dispatch = sut.ExecuteAsync("acme", async ct => + { + called = true; + accessor.MultiTenantContext.TenantInfo.ShouldBeSameAs(tenant); + await Task.Yield(); + accessor.MultiTenantContext.TenantInfo.ShouldBeSameAs(tenant); + ct.ThrowIfCancellationRequested(); + }); + + dispatch.IsCompleted.ShouldBeFalse(); + called.ShouldBeFalse(); + accessor.MultiTenantContext.ShouldBeSameAs(previous); + resolution.SetResult(tenant); + await dispatch; + called.ShouldBeTrue(); + accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Fact] + public void Begin_Should_RestoreMatchingTenant_When_NestedWorkReplacesContext() + { + var previous = new MultiTenantContext(NewTenant("acme", "dedicated-db")); + ((IMultiTenantContextSetter)_accessor).MultiTenantContext = previous; + var sut = new FinbuckleEventTenantScope(_accessor, _accessor); + + using (sut.Begin("acme")) + { + _accessor.MultiTenantContext.ShouldBeSameAs(previous); + ((IMultiTenantContextSetter)_accessor).MultiTenantContext = + new MultiTenantContext(NewTenant("other", "other-db")); + } + + _accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Theory] + [InlineData("acme", "dedicated-db")] + [InlineData("shared", "dedicated-db")] + [InlineData("default", "")] + public async Task ExecuteAsync_Should_UseCatalogMetadata_NotSourceConnection(string tenantId, string connection) + { + var resolver = Substitute.For(); + var tenant = NewTenant(tenantId, connection); + resolver.ResolveAsync(tenantId, Arg.Any()).Returns(Task.FromResult(tenant)); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var setter = provider.GetRequiredService(); + var previous = new MultiTenantContext(NewTenant("source", "wrong-db")); + setter.MultiTenantContext = previous; + + await provider.GetRequiredService().ExecuteAsync(tenantId, _ => + { + accessor.MultiTenantContext.TenantInfo.ShouldBeSameAs(tenant); + accessor.MultiTenantContext.TenantInfo.ShouldNotBeNull().ConnectionString.ShouldBe(connection); + accessor.MultiTenantContext.TenantInfo.Name.ShouldBe("catalog-name"); + return Task.CompletedTask; + }); + + accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Theory] + [InlineData(null)] + [InlineData("")] + [InlineData(" ")] + public async Task ExecuteAsync_Should_RouteGlobalToDefault_WithoutInheritingSource(string? tenantId) + { + var resolver = Substitute.For(); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var setter = provider.GetRequiredService(); + var previous = new MultiTenantContext(NewTenant("source", "dedicated-db")); + setter.MultiTenantContext = previous; + + await provider.GetRequiredService().ExecuteAsync(tenantId, async _ => + { + await Task.Yield(); + accessor.MultiTenantContext.TenantInfo!.Id.ShouldBe(MultitenancyConstants.Root.Id); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBeEmpty(); + }); + + accessor.MultiTenantContext.ShouldBeSameAs(previous); + await resolver.DidNotReceiveWithAnyArgs().ResolveAsync(default!, default); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExecuteAsync_Should_RejectUnknownOrInactive_BeforeConsumer(bool inactive) + { + var resolver = Substitute.For(); + AppTenantInfo? tenant = inactive ? NewTenant("target", "dedicated-db") : null; + if (tenant is not null) tenant.IsActive = false; + resolver.ResolveAsync("target", Arg.Any()).Returns(Task.FromResult(tenant)); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var previous = accessor.MultiTenantContext; + bool called = false; + + await Should.ThrowAsync(() => + provider.GetRequiredService().ExecuteAsync("target", _ => + { + called = true; + return Task.CompletedTask; + })); + + called.ShouldBeFalse(); + accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Theory] + [InlineData(true, false)] + [InlineData(true, true)] + [InlineData(false, false)] + [InlineData(false, true)] + public async Task ExecuteAsync_Should_RestoreOnLookupOrHandlerFailure(bool duringLookup, bool cancellation) + { + using var cancellationSource = new CancellationTokenSource(); + var resolution = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var handled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var resolver = Substitute.For(); + resolver.ResolveAsync("target", Arg.Any()) + .Returns(call => resolution.Task.WaitAsync(call.Arg())); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var previous = accessor.MultiTenantContext; + Task dispatch = provider.GetRequiredService().ExecuteAsync("target", async ct => + { + entered.SetResult(); + await handled.Task.WaitAsync(ct).ConfigureAwait(false); + }, cancellationSource.Token); + + if (!duringLookup) + { + resolution.SetResult(NewTenant("target", "dedicated-db")); + await entered.Task; + } + + if (cancellation) await cancellationSource.CancelAsync(); + else if (duringLookup) resolution.SetException(new InvalidOperationException("lookup failed")); + else handled.SetException(new InvalidOperationException("handler failed")); + + if (cancellation) await Should.ThrowAsync(() => dispatch); + else await Should.ThrowAsync(() => dispatch); + entered.Task.IsCompleted.ShouldBe(!duringLookup); + accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + [Fact] + public async Task ExecuteAsync_Should_IsolateConcurrentTenants_SharingDedicatedDatabase() + { + var resolver = Substitute.For(); + var resolution = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var firstEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var secondEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + resolver.ResolveAsync(Arg.Any(), Arg.Any()).Returns(async call => + { + await resolution.Task.WaitAsync(call.Arg()).ConfigureAwait(false); + return (AppTenantInfo?)NewTenant(call.Arg(), "shared-dedicated-db"); + }); + using ServiceProvider provider = CreateAsyncProvider(resolver); + var accessor = provider.GetRequiredService>(); + var previous = accessor.MultiTenantContext; + var sut = provider.GetRequiredService(); + + async Task HandleAsync(string tenantId, TaskCompletionSource entered, Task otherEntered) + { + accessor.MultiTenantContext.TenantInfo!.Id.ShouldBe(tenantId); + entered.SetResult(); + await otherEntered.ConfigureAwait(false); + accessor.MultiTenantContext.TenantInfo.Id.ShouldBe(tenantId); + accessor.MultiTenantContext.TenantInfo.ConnectionString.ShouldBe("shared-dedicated-db"); + } + + Task first = sut.ExecuteAsync("first", _ => HandleAsync("first", firstEntered, secondEntered.Task)); + Task second = sut.ExecuteAsync("second", _ => HandleAsync("second", secondEntered, firstEntered.Task)); + resolution.SetResult(); + await Task.WhenAll(first, second); + accessor.MultiTenantContext.ShouldBeSameAs(previous); + } + + private static AppTenantInfo NewTenant(string id, string connection) => new(id, id, connection, "event@example.test") + { + Name = "catalog-name", + Plan = "pro" + }; + + private static ServiceProvider CreateAsyncProvider(IEventTenantResolver resolver) + { + var services = new ServiceCollection(); + services.AddMultiTenant(); + services.AddSingleton(resolver); + services.AddSingleton(); + return services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true }); + } + #pragma warning disable S2376 // Finbuckle's IMultiTenantContextSetter is a set-only contract. private sealed class StubAccessor : IMultiTenantContextAccessor, IMultiTenantContextSetter { diff --git a/src/Tests/Webhooks.Tests/Services/WebhookFanoutHandlerTests.cs b/src/Tests/Webhooks.Tests/Services/WebhookFanoutHandlerTests.cs index 226dbdb849..7039d205e7 100644 --- a/src/Tests/Webhooks.Tests/Services/WebhookFanoutHandlerTests.cs +++ b/src/Tests/Webhooks.Tests/Services/WebhookFanoutHandlerTests.cs @@ -130,6 +130,39 @@ public async Task HandleAsync_Should_Not_Enqueue_When_Matching_Subscription_Is_I #region Resilience + [Fact] + public async Task HandleAsync_Should_PreserveResolvedTenantMetadata_DuringEnqueue() + { + await using var db = CreateContext(); + Guid subscriptionId = await SeedSubscriptionAsync(db, [EventType], isActive: true); + var tenant = _tenantAccessor.MultiTenantContext.TenantInfo!; + tenant.ConnectionString = "dedicated-database"; + tenant.Plan = "pro"; + _dispatcher.EnqueueAsync(TenantId, subscriptionId, EventType, Arg.Any(), Arg.Any()) + .Returns(_ => + { + _tenantAccessor.MultiTenantContext.TenantInfo.ShouldBeSameAs(tenant); + _tenantAccessor.MultiTenantContext.TenantInfo!.ConnectionString.ShouldBe("dedicated-database"); + return Task.CompletedTask; + }); + + await CreateHandler(db).HandleAsync(new FakeIntegrationEvent(TenantId)); + + await _dispatcher.Received(1).EnqueueAsync(TenantId, subscriptionId, EventType, Arg.Any(), Arg.Any()); + _tenantAccessor.MultiTenantContext.TenantInfo.ShouldBeSameAs(tenant); + } + + [Fact] + public async Task HandleAsync_Should_RejectMismatchedTenant_WithoutEnqueue() + { + await using var db = CreateContext(); + await SeedSubscriptionAsync(db, [EventType], isActive: true); + + await Should.ThrowAsync(() => CreateHandler(db).HandleAsync(new FakeIntegrationEvent("other"))); + + await _dispatcher.DidNotReceiveWithAnyArgs().EnqueueAsync(default!, default, default!, default!, default); + } + [Fact] public async Task HandleAsync_Should_Continue_Fanout_When_One_Enqueue_Throws() {