Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions .agents/rules/eventing.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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`).
2 changes: 1 addition & 1 deletion .agents/rules/modules/webhooks.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,6 @@ Tenant-scoped outbound webhook subscriptions with HMAC-signed delivery and retri
## Gotchas

- **Fan-out is an open-generic handler** — `WebhookFanoutHandler<TEvent>` 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=<hex HMACSHA256>` (`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`).
20 changes: 20 additions & 0 deletions src/BuildingBlocks/Eventing.Abstractions/IEventTenantScope.cs
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,24 @@ public interface IEventTenantScope
/// ambient context unchanged (global, non-tenant-scoped events).
/// </summary>
IDisposable Begin(string? tenantId);

/// <summary>
/// Resolves the dispatch context and runs <paramref name="action"/> within that same
/// asynchronous flow. Construct handlers and Inbox inside the action, not beforehand.
/// The default bridge uses <see cref="Begin"/>; multitenancy implementations can resolve
/// full catalog metadata and define global routing before invoking the action.
/// </summary>
async Task ExecuteAsync(

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Obtain approval before changing BuildingBlocks

When no explicit approval accompanies this change, do not land the new shared interface method: the repository's AGENTS.md marks all of src/BuildingBlocks as protected and prohibits modifications without explicit approval, while this commit changes both the public eventing abstraction and the framework event bus. Record the approval or move the behavior behind a module-owned adapter before merging.

AGENTS.md reference: AGENTS.md:L82-L82

Useful? React with 👍 / 👎.

string? tenantId,
Func<CancellationToken, Task> action,
CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(action);
ct.ThrowIfCancellationRequested();

using (Begin(tenantId))
{
await action(ct).ConfigureAwait(false);
}
}
}
6 changes: 3 additions & 3 deletions src/BuildingBlocks/Eventing/InMemory/InMemoryEventBus.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ public void ConfigureServices(IHostApplicationBuilder builder)
Services.TenantInitialPasswordBuffer>();

builder.Services.AddHeroDbContext<TenantDbContext>();
builder.Services.AddScoped<IEventTenantResolver, EventTenantResolver>();

// 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.
Expand Down
Original file line number Diff line number Diff line change
@@ -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<AppTenantInfo?> ResolveAsync(string tenantId, CancellationToken ct = default)
=> context.Set<AppTenantInfo>().AsNoTracking()
.SingleOrDefaultAsync(tenant => tenant.Id == tenantId, ct);
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -11,15 +12,14 @@ namespace FSH.Modules.Multitenancy.Services;
/// handler DbContexts resolved afterwards capture a real <c>TenantInfo</c> instead of the
/// null one a background scope would otherwise carry.
///
/// Mirrors the create-scope-then-set-tenant pattern already used by
/// <c>WebhookDispatchJob</c> / <c>SqlAuditSink</c>, 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.
/// </summary>
public sealed class FinbuckleEventTenantScope : IEventTenantScope
{
private readonly IMultiTenantContextAccessor<AppTenantInfo> _accessor;
private readonly IMultiTenantContextSetter _setter;
private readonly IServiceScopeFactory? _scopeFactory;

public FinbuckleEventTenantScope(
IMultiTenantContextAccessor<AppTenantInfo> accessor,
Expand All @@ -29,6 +29,16 @@ public FinbuckleEventTenantScope(
_setter = setter;
}

public FinbuckleEventTenantScope(
IMultiTenantContextAccessor<AppTenantInfo> accessor,
IMultiTenantContextSetter setter,
IServiceScopeFactory scopeFactory)
: this(accessor, setter)
{
ArgumentNullException.ThrowIfNull(scopeFactory);
_scopeFactory = scopeFactory;
}

public IDisposable Begin(string? tenantId)
{
if (string.IsNullOrWhiteSpace(tenantId))
Expand All @@ -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<AppTenantInfo>(new AppTenantInfo(tenantId, tenantId));

return new RestoreScope(_setter, previous);
}

public async Task ExecuteAsync(
string? tenantId,
Func<CancellationToken, Task> 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<AppTenantInfo>(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<IEventTenantResolver>()
.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<AppTenantInfo>(tenant);
await action(ct).ConfigureAwait(false);
}
finally
{
_setter.MultiTenantContext = previous;
}
}

private sealed class RestoreScope : IDisposable
{
private readonly IMultiTenantContextSetter _setter;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,8 @@ namespace FSH.Modules.Multitenancy.Services;
/// Installs an <see cref="AppTenantInfo"/> carrying the tenant's connection string, so an
/// <c>EventingDbContext</c> built inside the scope routes to that tenant's database.
///
/// Deliberately distinct from <see cref="FinbuckleEventTenantScope"/>, 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 <i>database</i> is what matters.
/// Distinct from <see cref="FinbuckleEventTenantScope"/>, which resolves the event's consumer
/// tenant. This scope selects the source database for a drain pass, independently of the consumer.
/// </summary>
public sealed class FinbuckleEventingDrainScope : IEventingDrainScope
{
Expand All @@ -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,
};
Expand All @@ -66,13 +59,4 @@ public RestoreScope(IMultiTenantContextSetter setter, IMultiTenantContext<AppTen
public void Dispose() => _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.
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
using FSH.Framework.Shared.Multitenancy;

namespace FSH.Modules.Multitenancy.Services;

public interface IEventTenantResolver
{
Task<AppTenantInfo?> ResolveAsync(string tenantId, CancellationToken ct = default);
}
Loading