Skip to content
Merged
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
15 changes: 15 additions & 0 deletions docs/articles/inbox-pattern.md
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,21 @@ builder.Services.AddCosmosInbox<AppDbContext>();

The marker is a self-contained document keyed and partitioned by `MessageId` in its own container, so a duplicate insert conflicts and is treated as already-processed. Because Cosmos cannot commit the marker and the business write atomically, the store writes the marker **after** the consumer's own commit and the guarantee is **effectively-once** — keep your consumer's writes idempotent (deterministic ids / upserts) so a redelivery that races ahead of the marker is harmless.

### Custom stores

Implement `IIdempotencyStore` to back the inbox with a different persistence technology. Register your store, then
call `AddInboxCore` to get the same retention sweep, metrics, and `TimeProvider` wiring that
`AddRelationalInbox`/`AddCosmosInbox` give the first-party stores — there is no first-party-only registration path:

```csharp
builder.Services.TryAddScoped<IIdempotencyStore, MyCustomIdempotencyStore>();
builder.Services.AddInboxCore(o => o.Retention.Enabled = true);
```

Implement `IInboxRetentionStore` too if your store should support the retention sweep. `AddInboxCore` registers the
sweep whenever `Retention.Enabled` is set, and the sweep silently skips (after a one-time warning) if the
registered store doesn't implement it.

## Typical Flow

```
Expand Down
35 changes: 22 additions & 13 deletions src/Vulthil.Messaging.Inbox/InboxCoreServiceCollectionExtensions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -4,23 +4,32 @@
namespace Vulthil.Messaging.Inbox;

/// <summary>
/// Registration shared by every inbox store package (<c>AddRelationalInbox</c>, <c>AddCosmosInbox</c>). The store
/// extensions differ only in which <see cref="IIdempotencyStore"/> implementation they register; the retention
/// sweep, OpenTelemetry metrics, and <see cref="TimeProvider"/> registration are identical regardless of the store,
/// so they live here once instead of being duplicated per store package. Internal, and visible to the store
/// packages via <c>InternalsVisibleTo</c>.
/// Service-collection extension that registers the inbox infrastructure shared by every idempotency store.
/// </summary>
internal static class InboxCoreServiceCollectionExtensions
public static class InboxCoreServiceCollectionExtensions
{
/// <summary>
/// Registers everything an <c>AddXyzInbox</c> store extension needs beyond its own store type: the shared
/// <see cref="TimeProvider"/>, the retention sweep (gated by <see cref="InboxRetentionOptions.Enabled"/>), and
/// OpenTelemetry metrics (gated by <see cref="InboxOptions.EnableMetrics"/>). <paramref name="configure"/> is
/// invoked exactly once here to evaluate both gates from a single materialized <see cref="InboxOptions"/>
/// instance, plus once more when the options system resolves <see cref="Microsoft.Extensions.Options.IOptions{TOptions}"/>
/// for injected consumers.
/// Registers everything an <see cref="IIdempotencyStore"/> needs beyond the store registration itself: the
/// shared <see cref="TimeProvider"/>, the retention sweep (gated by <see cref="InboxRetentionOptions.Enabled"/>),
/// and OpenTelemetry metrics (gated by <see cref="InboxOptions.EnableMetrics"/>).
/// </summary>
internal static IServiceCollection AddInboxCore(this IServiceCollection services, Action<InboxOptions>? configure)
/// <param name="services">The service collection.</param>
/// <param name="configure">
/// An optional action to configure <see cref="InboxOptions"/>. Invoked once eagerly, against a single
/// materialized <see cref="InboxOptions"/> instance, to evaluate both the retention and metrics gates, and once
/// more when the options system materializes <see cref="Microsoft.Extensions.Options.IOptions{TOptions}"/> for
/// injected consumers.
/// </param>
/// <returns>The same service collection, for chaining.</returns>
/// <remarks>
/// The first-party store extensions (<c>AddRelationalInbox</c>, <c>AddCosmosInbox</c>) register their
/// <see cref="IIdempotencyStore"/> implementation and then call this method; a custom store package follows the
/// same pattern — register its own <see cref="IIdempotencyStore"/> (typically with <c>TryAddScoped</c> so an
/// application can still substitute a different implementation), then call <c>AddInboxCore</c> to get the same
/// retention, metrics, and <see cref="TimeProvider"/> wiring the first-party stores get. There is no privileged
/// first-party path: every store, first- or third-party, shares this one entry point.
/// </remarks>
public static IServiceCollection AddInboxCore(this IServiceCollection services, Action<InboxOptions>? configure = null)
{
ArgumentNullException.ThrowIfNull(services);

Expand Down
2 changes: 2 additions & 0 deletions src/Vulthil.Messaging.Inbox/PublicAPI.Unshipped.txt
Original file line number Diff line number Diff line change
@@ -1 +1,3 @@
#nullable enable
Vulthil.Messaging.Inbox.InboxCoreServiceCollectionExtensions
static Vulthil.Messaging.Inbox.InboxCoreServiceCollectionExtensions.AddInboxCore(this Microsoft.Extensions.DependencyInjection.IServiceCollection! services, System.Action<Vulthil.Messaging.Inbox.InboxOptions!>? configure = null) -> Microsoft.Extensions.DependencyInjection.IServiceCollection!
5 changes: 0 additions & 5 deletions src/Vulthil.Messaging.Inbox/Vulthil.Messaging.Inbox.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,4 @@
<ProjectReference Include="..\Vulthil.Messaging\Vulthil.Messaging.csproj" />
</ItemGroup>

<ItemGroup>
<InternalsVisibleTo Include="Vulthil.Messaging.Inbox.Relational" />
<InternalsVisibleTo Include="Vulthil.Messaging.Inbox.Cosmos" />
</ItemGroup>

</Project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Hosting;
using OpenTelemetry.Metrics;
using Vulthil.Messaging.Abstractions.Consumers;
using Vulthil.xUnit;

namespace Vulthil.Messaging.Inbox.Tests;

public sealed class InboxCoreServiceCollectionExtensionsTests : BaseUnitTestCase
{
[Fact]
public void AThirdPartyStoreRegisteredBeforeAddInboxCoreResolvesUnchanged()
{
// Arrange — a custom store package registers its own IIdempotencyStore first, exactly as
// AddRelationalInbox/AddCosmosInbox register theirs before calling AddInboxCore.
var services = new ServiceCollection();
services.TryAddScoped<IIdempotencyStore, FakeIdempotencyStore>();

// Act
services.AddInboxCore();
using var provider = services.BuildServiceProvider();

// Assert
provider.GetRequiredService<IIdempotencyStore>().ShouldBeOfType<FakeIdempotencyStore>();
}

[Fact]
public void RegistersTheSharedTimeProvider()
{
// Arrange
var services = new ServiceCollection();

// Act
services.AddInboxCore();

// Assert
services.ShouldContain(descriptor => descriptor.ServiceType == typeof(TimeProvider));
}

[Fact]
public void EnableMetricsRegistersTheMeterProviderService()
{
// Arrange
var services = new ServiceCollection();

// Act
services.AddInboxCore(o => o.EnableMetrics = true);

// Assert
services.ShouldContain(descriptor => descriptor.ServiceType == typeof(MeterProvider));
}

[Fact]
public void EnableMetricsDisabledSkipsMeterProviderRegistration()
{
// Arrange
var services = new ServiceCollection();

// Act
services.AddInboxCore(o => o.EnableMetrics = false);

// Assert
services.ShouldNotContain(descriptor => descriptor.ServiceType == typeof(MeterProvider));
}

[Fact]
public void RetentionEnabledRegistersTheRetentionBackgroundService()
{
// Arrange
var services = new ServiceCollection();

// Act
services.AddInboxCore(o => o.Retention.Enabled = true);

// Assert
services.ShouldContain(descriptor => descriptor.ServiceType == typeof(IHostedService) && descriptor.ImplementationType == typeof(InboxRetentionBackgroundService));
}

[Fact]
public void RetentionDisabledDoesNotRegisterTheRetentionBackgroundService()
{
// Arrange
var services = new ServiceCollection();

// Act
services.AddInboxCore(o => o.Retention.Enabled = false);

// Assert
services.ShouldNotContain(descriptor => descriptor.ServiceType == typeof(IHostedService) && descriptor.ImplementationType == typeof(InboxRetentionBackgroundService));
}

[Fact]
public void CanBeCalledWithoutAConfigureDelegate()
{
// Arrange
var services = new ServiceCollection();

// Act
var exception = Record.Exception(() => services.AddInboxCore());

// Assert
exception.ShouldBeNull();
}

[Fact]
public void NullServicesThrows()
{
// Arrange
IServiceCollection services = null!;

// Act & Assert
Should.Throw<ArgumentNullException>(() => services.AddInboxCore());
}

private sealed class FakeIdempotencyStore : IIdempotencyStore, IInboxRetentionStore
{
public Task<bool> ProcessAsync(string idempotencyKey, IMessageContext context, Func<CancellationToken, Task> process, CancellationToken cancellationToken) => Task.FromResult(false);

public Task<int> DeleteProcessedAsync(DateTimeOffset olderThanUtc, int batchSize, CancellationToken cancellationToken) => Task.FromResult(0);
}
}