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 new file mode 100644 index 0000000000..9ce9c0331a --- /dev/null +++ b/src/Tests/Integration.Tests/Tests/Eventing/DedicatedTenantOutboxRoutingTests.cs @@ -0,0 +1,817 @@ +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.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; + +[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.Id); + if (databaseCreated) + { + await ExecuteDatabaseCommandAsync(defaultConnection, $"DROP DATABASE {QuoteIdentifier(databaseName)} WITH (FORCE)"); + } + } + } + + [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 + { + ["EventingOptions:Provider"] = "InMemory", + ["EventingOptions:UseHostedServiceDispatcher"] = "false" + }).Build(); + var services = new ServiceCollection(); + services.AddSingleton(configuration); + services.AddSingleton(hostServices.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.AddScoped(); + 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>(); + configure?.Invoke(services); + 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, IIntegrationEvent integrationEvent, string? handlerName = null) + { + 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" 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 ?? 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", 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, Guid eventId) + { + 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-{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) + { + context.Brands.Add(Brand.Create($"dispatch-{integrationEvent.Id:N}", null, null)); + 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 new file mode 100644 index 0000000000..9fa313e494 --- /dev/null +++ b/src/Tests/Integration.Tests/Tests/Eventing/OutboxDedicatedTenantDispatchTests.cs @@ -0,0 +1,254 @@ +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.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, + }; + var accessor = _factory.Services.GetRequiredService>(); + var previous = accessor.MultiTenantContext; + var eventIds = new List(); + + 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 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.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.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.GetRequiredService(); + using (drainScope.Begin(sourceTarget)) + { + using (IServiceScope dispatchScope = app.CreateScope()) + { + EventingDbContext sourceOutbox = dispatchScope.ServiceProvider.GetRequiredService(); + sourceOutbox.Database.GetDbConnection().Database.ShouldBe(sourceDatabaseName); + + await dispatchScope.ServiceProvider + .GetRequiredService() + .DispatchAsync(CancellationToken.None); + } + + using IServiceScope verifyScope = app.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 + { + _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 ServiceProvider CreateProvider(string defaultConnection, DatabaseRoutingRecorder recorder) => + DedicatedTenantOutboxRoutingTests.CreateProvider(_factory.Services, defaultConnection, services => + { + services.AddSingleton(recorder); + services.AddScoped, 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..2dcbcd583e 100644 --- a/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs +++ b/src/Tests/Multitenancy.Tests/Services/FinbuckleEventTenantScopeTests.cs @@ -1,7 +1,11 @@ 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 NSubstitute; using Shouldly; using Xunit; @@ -11,6 +15,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() { @@ -59,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() {