diff --git a/Directory.Packages.props b/Directory.Packages.props
index a5a221986..e6e4e6d06 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -25,7 +25,7 @@
-
+
@@ -56,6 +56,7 @@
+
@@ -63,6 +64,7 @@
+
diff --git a/Eventuous.slnx b/Eventuous.slnx
index 1af57601b..546b2b5e8 100644
--- a/Eventuous.slnx
+++ b/Eventuous.slnx
@@ -14,9 +14,11 @@
+
+
diff --git a/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjector.cs b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjector.cs
new file mode 100644
index 000000000..0420aa4bf
--- /dev/null
+++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjector.cs
@@ -0,0 +1,229 @@
+// Copyright (C) Eventuous HQ OÜ. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+using Azure;
+using Azure.Storage.Blobs.Models;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Subscriptions.Logging;
+using System.Text.Json;
+
+using static Eventuous.Subscriptions.Diagnostics.SubscriptionsEventSource;
+
+namespace Eventuous.Azure.Storage.Blobs;
+
+///
+/// Projects event store events to Azure Blob Storage as state objects of type T.
+///
+///
+///
+/// This projector works by maintaining a state object of type T in Azure Blob Storage for each event stream.
+/// When an event is received, it retrieves the current state blob (or creates a new state instance if the blob doesn't exist),
+/// applies the event to the state using the registered event handler, and uploads the updated state back to Blob Storage.
+/// The projector uses optimistic concurrency control via ETags to handle concurrent updates, and provides virtual methods
+/// for customizing blob naming conventions. Multiple event types can be handled by registering handlers using the On(TEvent) methods.
+/// The optional getBlobId parameter in event registration allows custom blob ID generation, which is useful when the default
+/// stream ID from context.Stream.GetId() needs to be overridden, such as using event metadata or custom business logic.
+/// The blob container must exist; the projector doesn't create it.
+///
+///
+public class BlobStorageProjector : BaseEventHandler where T : class, new() {
+ /// Azure Blob Storage container client.
+ protected readonly BlobContainerClient ContainerClient;
+
+ readonly JsonSerializerOptions _jsonOptions;
+ readonly Dictionary>> _handlers = new();
+ readonly ITypeMapper _map;
+ readonly int _raceRetries;
+ readonly IdempotencyMode _idempotencyMode;
+
+ /// Delegate for custom blob ID generation from consume context.
+ /// Event type being consumed.
+ /// Event consume context.
+ /// Blob ID as string.
+ public delegate ValueTask GetBlobId(IMessageConsumeContext context) where TEvent : class;
+
+ ///
+ /// Initializes projector with existing container client.
+ ///
+ /// Azure Blob Storage container client.
+ /// Optional projector configuration.
+ /// Optional type mapper for event type resolution.
+ public BlobStorageProjector(BlobContainerClient container, BlobStorageProjectorOptions? projectorOptions = null, ITypeMapper? mapper = null) {
+ ContainerClient = container;
+ _jsonOptions = new(projectorOptions?.JsonOptions ?? JsonSerializerOptions.Web);
+ _map = mapper ?? TypeMap.Instance;
+ _raceRetries = projectorOptions?.RaceRetries ?? 0;
+ _idempotencyMode = projectorOptions?.IdempotencyMode ?? IdempotencyMode.None;
+ }
+
+ ///
+ /// Initializes projector with service client and container name.
+ ///
+ /// Azure Blob Storage service client.
+ /// Name of the container to use.
+ /// Optional projector configuration.
+ /// Optional type mapper for event type resolution.
+ public BlobStorageProjector(
+ BlobServiceClient serviceClient,
+ string containerName,
+ BlobStorageProjectorOptions? projectorOptions = null,
+ ITypeMapper? mapper = null
+ ) : this(serviceClient.GetBlobContainerClient(containerName), projectorOptions, mapper) { }
+
+ /// Registers event handler with sync state update.
+ /// Event type to handle.
+ /// State update function receiving current state and event.
+ /// Optional custom blob ID generator.
+ protected void On(Func handler, GetBlobId? getBlobId = null) where TEvent : class
+ => On((ctx, state) => new ValueTask(handler(state, ctx.Message)), getBlobId);
+
+ /// Registers event handler with context and sync state update.
+ /// Event type to handle.
+ /// State update function receiving context, current state, and event.
+ /// Optional custom blob ID generator.
+ protected void On(Func, T, T> handler, GetBlobId? getBlobId = null) where TEvent : class
+ => On((ctx, state) => new ValueTask(handler(ctx, state)), getBlobId);
+
+ /// Registers event handler with async state update.
+ /// Event type to handle.
+ /// Async state update function receiving current state and event.
+ /// Optional custom blob ID generator.
+ protected void On(Func> handler, GetBlobId? getBlobId = null) where TEvent : class
+ => On((ctx, state) => handler(state, ctx.Message), getBlobId);
+
+ /// Registers event handler with context, async state update, and custom blob ID.
+ /// Event type to handle.
+ /// Async state update function receiving context, current state, and event.
+ /// Optional custom blob ID generator.
+ protected void On(Func, T, ValueTask> handler, GetBlobId? getBlobId = null) where TEvent : class {
+ if (!_handlers.TryAdd(typeof(TEvent), new Handler(this, handler, getBlobId).Handle)) {
+ throw new ArgumentException($"Type {typeof(TEvent).Name} already has a handler");
+ }
+
+ if (!_map.TryGetTypeName(out _)) {
+ Log.MessageTypeNotRegistered();
+ }
+ }
+
+ /// Handles incoming event by dispatching to registered handler.
+ /// Event consume context.
+ /// Event handling status indicating success, failure, or ignore.
+ public override async ValueTask HandleEvent(IMessageConsumeContext context) =>
+ _handlers.TryGetValue(context.Message!.GetType(), out var handler)
+ ? await handler(context).NoContext()
+ : EventHandlingStatus.Ignored;
+
+ T ToObjectFromJson(BinaryData content) => content.ToObjectFromJson(_jsonOptions) ?? new T();
+
+ byte[] SerializeToUtf8Bytes(T updated) => JsonSerializer.SerializeToUtf8Bytes(updated, _jsonOptions);
+
+ /// Gets blob name from ID and context. Can be overridden for custom naming.
+ /// Blob identifier.
+ /// Event consume context.
+ /// Blob name as string.
+ protected virtual string GetBlobName(string id, IMessageConsumeContext context) => GetBlobName(id);
+
+ /// Gets blob name from ID. Default format: {id}/{T}.json
+ /// Blob identifier.
+ /// Blob name as string.
+ protected virtual string GetBlobName(string id) => $"{id}/{typeof(T).Name}.json";
+
+ class Handler(BlobStorageProjector projector, Func, T, ValueTask> eventHandler, GetBlobId? getBlobId)
+ where TEvent : class {
+ bool _warnedZeroGlobalPosition;
+
+ public async ValueTask Handle(IMessageConsumeContext context) {
+ var typedContext = context as MessageConsumeContext ?? new MessageConsumeContext(context);
+
+ if (projector._idempotencyMode == IdempotencyMode.ByGlobalPosition && context.GlobalPosition == 0 && !_warnedZeroGlobalPosition) {
+ _warnedZeroGlobalPosition = true;
+
+ Logger.Current?.WarnLog?.Log(
+ "ByGlobalPosition idempotency requires events with real global positions, but an event arrived with global position 0. Subsequent events may be treated as duplicates and ignored. Use ByMessageId for message broker subscriptions."
+ );
+ }
+
+ var blobId = getBlobId == null
+ ? context.Stream.GetId()
+ : await getBlobId(typedContext).NoContext();
+ var blobName = projector.GetBlobName(blobId, typedContext);
+
+ var blobClient = projector.ContainerClient.GetBlobClient(blobName);
+
+ var retries = projector._raceRetries;
+
+ while (true) {
+ Response? blobContent;
+
+ try {
+ blobContent = await blobClient.DownloadContentAsync(typedContext.CancellationToken).NoContext();
+ } catch (RequestFailedException ex) when (ex.Status == 404 && ex.ErrorCode == BlobErrorCode.BlobNotFound.ToString()) {
+ // Blob doesn't exist, start with a new instance
+ blobContent = null;
+ }
+
+ T current;
+ BlobRequestConditions conditions;
+
+ if (blobContent == null) {
+ current = new T();
+ conditions = new BlobRequestConditions { IfNoneMatch = ETag.All };
+ } else {
+ // Check idempotency if enabled
+ if (projector._idempotencyMode != IdempotencyMode.None && IsDuplicate(blobContent.Value.Details.Metadata)) {
+ return EventHandlingStatus.Ignored;
+ }
+
+ current = projector.ToObjectFromJson(blobContent.Value.Content);
+ conditions = new BlobRequestConditions { IfMatch = blobContent.Value.Details.ETag };
+ }
+
+ // The user-supplied handler and the user-configurable JSON serialization run outside
+ // the catch blocks, so their own Azure exceptions are never mistaken for blob races
+ var updated = await eventHandler(typedContext, current).NoContext();
+ var json = projector.SerializeToUtf8Bytes(updated);
+
+ var uploadOptions = new BlobUploadOptions {
+ Conditions = conditions,
+ HttpHeaders = new BlobHttpHeaders {
+ ContentType = "application/json"
+ },
+ // Azure requires metadata values to be ASCII, while stream names and message ids
+ // can be arbitrary strings, so they are stored percent-encoded
+ Metadata = new Dictionary {
+ ["Stream"] = Uri.EscapeDataString(typedContext.Stream.ToString()),
+ ["MessageId"] = Uri.EscapeDataString(typedContext.MessageId),
+ ["StreamPosition"] = typedContext.StreamPosition.ToString(),
+ ["GlobalPosition"] = typedContext.GlobalPosition.ToString()
+ }
+ };
+
+ try {
+ using var stream = new MemoryStream(json);
+ await blobClient.UploadAsync(stream, uploadOptions, typedContext.CancellationToken).NoContext();
+
+ return EventHandlingStatus.Success;
+ } catch (RequestFailedException ex) when (IsConcurrencyConflict(ex)) {
+ // Lost the optimistic concurrency race: re-read the state and try again
+ if (retries-- <= 0) return EventHandlingStatus.Failure;
+ }
+ }
+
+ bool IsDuplicate(IDictionary metadata) => projector._idempotencyMode switch {
+ IdempotencyMode.ByGlobalPosition =>
+ metadata.TryGetValue("GlobalPosition", out var storedPosition) &&
+ ulong.TryParse(storedPosition, out var currentGlobalPosition) &&
+ typedContext.GlobalPosition <= currentGlobalPosition,
+ IdempotencyMode.ByMessageId =>
+ metadata.TryGetValue("MessageId", out var storedId) &&
+ storedId == Uri.EscapeDataString(typedContext.MessageId),
+ _ => false
+ };
+ }
+
+ static bool IsConcurrencyConflict(RequestFailedException ex)
+ => ex.Status is 412 or 409 &&
+ (ex.ErrorCode == BlobErrorCode.ConditionNotMet.ToString() || ex.ErrorCode == BlobErrorCode.BlobAlreadyExists.ToString());
+ }
+}
diff --git a/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjectorOptions.cs b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjectorOptions.cs
new file mode 100644
index 000000000..73db04d48
--- /dev/null
+++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjectorOptions.cs
@@ -0,0 +1,58 @@
+// Copyright (C) Eventuous HQ OÜ. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
+using System.Text.Json;
+
+namespace Eventuous.Azure.Storage.Blobs;
+
+///
+/// Options for configuring the storage blob projector.
+///
+public class BlobStorageProjectorOptions {
+ ///
+ /// Gets or sets the JSON serializer options to use when serializing or deserializing projection state.
+ /// When not set, is used.
+ ///
+ public JsonSerializerOptions? JsonOptions { get; set; }
+
+ ///
+ /// Gets or sets the number of retry attempts for race condition handling when saving projection state.
+ /// Default is 0 (no retries).
+ ///
+ public int RaceRetries { get; set; }
+
+ ///
+ /// Gets or sets the idempotency mode for the projector. When enabled, the projector will skip processing
+ /// if the blob already exists with a matching identifier (message ID or global position), preventing duplicate processing.
+ /// Default is (no idempotency checking).
+ ///
+ public IdempotencyMode IdempotencyMode { get; set; } = IdempotencyMode.None;
+}
+
+///
+/// Controls how the projection handles idempotency to prevent duplicate message processing.
+///
+public enum IdempotencyMode {
+ ///
+ /// No idempotency checks. The projector will always process messages and update blobs.
+ /// Use when duplicate processing is acceptable or when external mechanisms ensure message uniqueness.
+ ///
+ None,
+
+ ///
+ /// Skips processing if the existing blob was created from a message at the same or later global position.
+ /// Uses the GlobalPosition metadata stored with the blob for comparison.
+ /// Requires a subscription that provides real global positions, such as an all-stream subscription.
+ /// Do not use with message broker subscriptions where the global position is always 0 — every event
+ /// after the first would be treated as a duplicate and ignored. Use instead.
+ ///
+ ByGlobalPosition,
+
+ ///
+ /// Skips processing if the existing blob was created from the same message ID.
+ /// Uses the MessageId metadata stored with the blob for comparison.
+ /// More precise than position-based checks, works even if messages are processed out of order.
+ /// Especially from external message queues where global position may not be available or reliable.
+ ///
+ ByMessageId
+}
diff --git a/src/Azure/src/Eventuous.Azure.Storage.Blobs/Eventuous.Azure.Storage.Blobs.csproj b/src/Azure/src/Eventuous.Azure.Storage.Blobs/Eventuous.Azure.Storage.Blobs.csproj
new file mode 100644
index 000000000..edf8dc2c3
--- /dev/null
+++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/Eventuous.Azure.Storage.Blobs.csproj
@@ -0,0 +1,39 @@
+
+
+
+ README.md
+ true
+ true
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Tools\TaskExtensions.cs
+
+
+ Tools\Ensure.cs
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/src/Azure/src/Eventuous.Azure.Storage.Blobs/README.md b/src/Azure/src/Eventuous.Azure.Storage.Blobs/README.md
new file mode 100644
index 000000000..00a77abd1
--- /dev/null
+++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/README.md
@@ -0,0 +1,125 @@
+# Eventuous Azure Blob Storage Projections
+
+This package adds Azure Blob Storage projections to applications built with Eventuous. It allows you to project event store events to Azure Blob Storage as state objects, maintaining a separate state document for each event stream.
+
+## Using projections
+
+Create your own projection class that inherits from `BlobStorageProjector` where `T` is your state type. The state type must be a class with a parameterless constructor.
+
+Register event handlers using the `On` methods. When an event is received, the projector retrieves the current state blob (or creates a new state instance if the blob doesn't exist), applies the event to the state using the registered event handler, and uploads the updated state back to Blob Storage.
+
+The class provides two constructors:
+
+* `BlobStorageProjector(BlobContainerClient container, ...` where the container client is passed directly
+* `BlobStorageProjector(BlobServiceClient serviceClient, string containerName, ...` where the service client is set up by Azure DI and the container name is set by the projection
+
+The blob container must exist before the projector handles events; the projector doesn't create it.
+
+JSON serialization is configured via `BlobStorageProjectorOptions.JsonOptions`. When you keep serializer options in ASP.NET Core DI, pass them on: `new BlobStorageProjectorOptions { JsonOptions = options.Value }`.
+
+By default, the blob ID is extracted from the stream using `context.Stream.GetId()`. You can override this by providing a custom `getBlobId` function in the event registration:
+
+```csharp
+public class BookingProjection : BlobStorageProjector {
+ public BookingProjection(BlobServiceClient client)
+ : base(client, "bookings-container") {
+
+ // Uses default blob ID from stream
+ On((state, evt) => {
+ state.RoomId = evt.RoomId;
+ state.CheckInDate = evt.CheckIn;
+ return state;
+ });
+
+ // Custom blob ID using event data
+ On(
+ (state, evt) => {
+ state.PaidAmount += evt.AmountPaid;
+ return state;
+ },
+ context => new ValueTask($"custom-{context.Message.BookingId}")
+ );
+ }
+}
+```
+
+## Projector options
+
+The `BlobStorageProjectorOptions` class provides several configuration options for fine-tuning the projector behavior.
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `JsonOptions` | `JsonSerializerOptions?` | `null` (uses `JsonSerializerOptions.Web`) | JSON serializer options for state serialization/deserialization. Controls formatting, naming policies, etc. |
+| `RaceRetries` | `int` | `0` | Number of retry attempts for optimistic concurrency conflicts. Increase when concurrent updates are likely. |
+| `IdempotencyMode` | `IdempotencyMode` | `IdempotencyMode.None` | Controls duplicate message detection behavior. |
+
+### Idempotency modes
+
+The `IdempotencyMode` enum controls how the projector handles duplicate messages:
+
+- **`None`** - No idempotency checks. Will process messages and updates blob, without
+checking for duplicates.
+- **`ByGlobalPosition`** - Skips processing if the existing blob has a global position set in
+its metadata that indicates it has already been processed. The event global position must be greater than that stored in the blob.
+This mode requires a subscription that provides real global positions, such as an all-stream subscription.
+Do not use it with message broker subscriptions where the global position is always zero — the first event would store
+position `0` and every subsequent event would be treated as a duplicate and silently ignored. Use `ByMessageId` instead.
+- **`ByMessageId`** - Use this when building projections directly from integration events. Skips
+processing if the message ID in the blob metadata matches that in the event.
+Note, this means the idempotency is weaker as only the last message ID is checked. Older messages that are replayed will be processed as normal.
+
+### Custom blob naming
+
+By default, blob names are generated using `GetBlobName(string id)` which creates names in the format `{id}/{T}.json`, where `id` defaults to the stream ID from `context.Stream.GetId()`.
+
+You can customize blob naming in two ways:
+
+**1. Override the virtual methods globally for all events:**
+
+```csharp
+protected override string GetBlobName(string id, IMessageConsumeContext context) {
+ // Use stream name and type in the path
+ var streamName = context.Stream.ToString();
+ return $"projections/{streamName}/{id}.json";
+}
+
+protected override string GetBlobName(string id) {
+ return $"{id}/{typeof(T).Name}.json";
+}
+```
+
+**2. Override blob ID per event handler using `getBlobId`:**
+
+```csharp
+On(
+ (state, evt) => {
+ state.PaidAmount += evt.AmountPaid;
+ return state;
+ },
+ // Custom blob ID for this specific event only
+ context => new ValueTask($"payments-{context.Message.BookingId}")
+);
+```
+
+Note that `getBlobId` returns a blob _ID_, not a full blob name: the result is still passed to `GetBlobName`, so with the default naming the example above produces `payments-{id}/BookingState.json`. To change the full blob path, override `GetBlobName` as well.
+
+Use per-event blob ID overrides when you need different events to target different blobs within the same projector, such as when the business identifier differs from the stream identifier.
+
+## Features
+
+- **Automatic state management** - Creates new state instances when blobs don't exist
+- **Optimistic concurrency control** - Uses ETags for safe concurrent updates
+- **Idempotency** - Prevents duplicate processing with configurable modes
+- **Retry handling** - Automatic retries for race conditions
+- **Flexible blob naming** - Customizable blob ID and naming conventions
+- **Metadata storage** - Automatically stores stream info, positions, and message IDs
+
+## Background
+
+The projector stores each state as a separate blob in Azure Blob Storage. Each blob contains:
+- The serialized state object (JSON by default)
+- Metadata including stream name, message ID, stream position, and global position;
+ because Azure requires metadata values to be ASCII, the stream name and message ID are stored percent-encoded
+- Content type set to `application/json`
+
+This approach provides natural partitioning by stream and enables efficient state retrieval for individual streams.
\ No newline at end of file
diff --git a/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/BlobStorageProjectorTests.cs b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/BlobStorageProjectorTests.cs
new file mode 100644
index 000000000..0dad57381
--- /dev/null
+++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/BlobStorageProjectorTests.cs
@@ -0,0 +1,729 @@
+using System.Text.Json;
+using Azure;
+using Azure.Storage.Blobs;
+using Eventuous.Azure.Storage.Blobs;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Tests.Azure.Storage.Blobs.Fixtures;
+
+namespace Eventuous.Tests.Azure.Storage.Blobs;
+
+[ClassDataSource]
+public class BlobStorageProjectorTests(IntegrationFixture fixture) {
+ const string DefaultStream = "stream";
+
+ // ========== HELPER METHODS (surface intent through naming) ==========
+
+ ///
+ /// Creates a test container for the given scenario, surfacing the handler type and test case.
+ /// Returns the container name for use with the new constructor.
+ ///
+ async Task SetupContainer(string scenarioName) {
+ var containerName = $"test-{scenarioName}";
+ var client = fixture.BlobServiceClient.GetBlobContainerClient(containerName);
+ await client.CreateAsync();
+ return containerName;
+ }
+
+ ///
+ /// Gets a BlobContainerClient for the given container name
+ ///
+ BlobContainerClient GetContainer(string containerName) =>
+ fixture.BlobServiceClient.GetBlobContainerClient(containerName);
+
+ ///
+ /// Sets up initial blob state for update scenarios
+ ///
+ async Task SetupExistingBlob(string containerName, string blobName, TState initialState) {
+ var blobClient = GetContainer(containerName).GetBlobClient(blobName);
+ var json = JsonSerializer.SerializeToUtf8Bytes(initialState);
+ await blobClient.UploadAsync(new MemoryStream(json), overwrite: true);
+ }
+
+ ///
+ /// Gets the state from blob, surfacing the expected state type
+ ///
+ async Task GetBlobState(string containerName, string blobName) {
+ var blobClient = GetContainer(containerName).GetBlobClient(blobName);
+ var blob = await blobClient.DownloadContentAsync();
+ return blob.Value.Content.ToObjectFromJson(JsonSerializerOptions.Web)!;
+ }
+
+ ///
+ /// Asserts that the projector result is Success
+ ///
+ static async Task AssertSuccess(EventHandlingStatus result) => await Assert.That(result).IsEqualTo(EventHandlingStatus.Success);
+
+ ///
+ /// Asserts that the projector result is Ignored
+ ///
+ static async Task AssertIgnored(EventHandlingStatus result) => await Assert.That(result).IsEqualTo(EventHandlingStatus.Ignored);
+
+ ///
+ /// Asserts that the projector result is Failure
+ ///
+ static async Task AssertFailure(EventHandlingStatus result) => await Assert.That(result).IsEqualTo(EventHandlingStatus.Failure);
+
+ ///
+ /// Creates a concurrent-modification action that overwrites the blob directly with a different value
+ ///
+ Func OverwriteBlob(string containerName, string blobName) => async () => {
+ var modifiedState = new ConcurrentState { Value = 999 };
+ var modifiedJson = JsonSerializer.SerializeToUtf8Bytes(modifiedState);
+ var blobClient = GetContainer(containerName).GetBlobClient(blobName);
+ await blobClient.UploadAsync(new MemoryStream(modifiedJson), overwrite: true);
+ };
+
+ // ========== SYNC STATE HANDLER TESTS ==========
+
+ [Test]
+ public async Task SyncStateHandler_NewBlob_ShouldCreateAndStoreState() {
+ // Arrange
+ var containerName = await SetupContainer("sync-state-new");
+ var projector = new SyncStateProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, $"{DefaultStream}/SyncState.json");
+ await Assert.That(state.Value).IsEqualTo(10);
+ }
+
+ [Test]
+ public async Task SyncStateHandler_ExistingBlob_ShouldUpdateState() {
+ // Arrange
+ var containerName = await SetupContainer("sync-state-existing");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ await SetupExistingBlob(containerName, blobName, new SyncState { Value = 5 });
+
+ var projector = new SyncStateProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(15); // 5 + 10
+ await Assert.That(state.Counter).IsEqualTo(1);
+ }
+
+ // ========== SYNC CONTEXT-AWARE HANDLER TESTS ==========
+
+ [Test]
+ public async Task SyncContextAwareHandler_NewBlob_ShouldUseContextAndStoreState() {
+ // Arrange
+ var containerName = await SetupContainer("sync-context-new");
+ var projector = new SyncContextAwareProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 20 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, $"{DefaultStream}/SyncContextState.json");
+ await Assert.That(state.Value).IsEqualTo(20);
+ await Assert.That(state.StreamId).IsEqualTo(DefaultStream);
+ }
+
+ // ========== ASYNC STATE HANDLER TESTS ==========
+
+ [Test]
+ public async Task AsyncStateHandler_NewBlob_ShouldCreateAndStoreState() {
+ // Arrange
+ var containerName = await SetupContainer("async-state-new");
+ var projector = new AsyncStateProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 30 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, $"{DefaultStream}/AsyncState.json");
+ await Assert.That(state.Value).IsEqualTo(30);
+ }
+
+ [Test]
+ public async Task AsyncStateHandler_ExistingBlob_ShouldUpdateState() {
+ // Arrange
+ var containerName = await SetupContainer("async-state-existing");
+ var blobName = $"{DefaultStream}/AsyncState.json";
+
+ await SetupExistingBlob(containerName, blobName, new AsyncState { Value = 5 });
+
+ var projector = new AsyncStateProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 35 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(40); // 5 + 35
+ }
+
+ // ========== ASYNC CONTEXT-AWARE HANDLER TESTS ==========
+
+ [Test]
+ public async Task AsyncContextAwareHandler_NewBlob_ShouldUseContextAndStoreState() {
+ // Arrange
+ var containerName = await SetupContainer("async-context-new");
+ var projector = new AsyncContextAwareProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 40, Name = "AsyncContext" });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, $"{DefaultStream}/AsyncContextState.json");
+ await Assert.That(state.Value).IsEqualTo(40);
+ await Assert.That(state.EventName).IsEqualTo("AsyncContext");
+ }
+
+ [Test]
+ public async Task AsyncContextAwareHandler_ExistingBlob_ShouldUpdateStateAndContext() {
+ // Arrange
+ var containerName = await SetupContainer("async-context-existing");
+ var blobName = $"{DefaultStream}/AsyncContextState.json";
+
+ await SetupExistingBlob(containerName, blobName, new AsyncContextState { Value = 10, EventName = "Initial" });
+
+ var projector = new AsyncContextAwareProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 50, Name = "Update" });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(60); // 10 + 50
+ await Assert.That(state.EventName).IsEqualTo("Update");
+ }
+
+ // ========== EDGE CASE TESTS ==========
+
+ [Test]
+ public async Task NoHandler_ShouldReturnIgnored() {
+ // Arrange
+ var containerName = await SetupContainer("no-handler");
+ var projector = new NoHandlerProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 100 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertIgnored(result);
+ }
+
+ // ========== CUSTOM BLOB ID TESTS ==========
+
+ [Test]
+ public async Task CustomBlobId_NewBlob_ShouldUseEventIdForBlobName() {
+ // Arrange
+ var containerName = await SetupContainer("custom-blobid-new");
+ var eventId = Guid.NewGuid().ToString();
+ var projector = new CustomBlobIdProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Id = eventId, Value = 100 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var blobName = $"{eventId}/CustomBlobIdState.json";
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(100);
+ }
+
+ [Test]
+ public async Task CustomBlobId_ExistingBlob_ShouldUpdateWithEventId() {
+ // Arrange
+ var containerName = await SetupContainer("custom-blobid-existing");
+ var eventId = Guid.NewGuid().ToString();
+ var blobName = $"{eventId}/CustomBlobIdState.json";
+
+ await SetupExistingBlob(containerName, blobName, new CustomBlobIdState { Value = 5 });
+
+ var projector = new CustomBlobIdProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Id = eventId, Value = 100 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(105); // 5 + 100
+ }
+
+ [Test]
+ public async Task UnicodeStreamName_ShouldStoreStateWithEncodedMetadata() {
+ // Arrange
+ var containerName = await SetupContainer("unicode-stream");
+ const string streamName = "Booking-Ålesund";
+ var projector = new SyncStateProjector(fixture.BlobServiceClient, containerName);
+ var context = CreateContext(new TestEvent { Value = 10 }, stream: streamName);
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert
+ await AssertSuccess(result);
+
+ var blobName = "Ålesund/SyncState.json";
+ var state = await GetBlobState(containerName, blobName);
+ await Assert.That(state.Value).IsEqualTo(10);
+
+ // Metadata values must be ASCII, so the stream name is stored percent-encoded
+ var properties = await GetContainer(containerName).GetBlobClient(blobName).GetPropertiesAsync();
+ await Assert.That(properties.Value.Metadata["Stream"]).IsEqualTo(Uri.EscapeDataString(streamName));
+ }
+
+ [Test]
+ public async Task HandlerThrowingRequestFailed_ShouldPropagateWithoutRaceRetries() {
+ // Arrange
+ var containerName = await SetupContainer("handler-exception");
+ var projector = new ThrowingHandlerProjector(fixture.BlobServiceClient, containerName, raceRetries: 2);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // Act & Assert - the handler's own Azure exception propagates instead of being
+ // classified as an optimistic concurrency race and retried
+ await Assert.ThrowsAsync(() => projector.HandleEvent(context).AsTask());
+ await Assert.That(projector.HandlerCalls).IsEqualTo(1);
+ }
+
+ // ========== RACE RETRY TESTS ==========
+
+ [Test]
+ public async Task RaceRetries_WithOneRetry_ShouldSucceedAfterRaceCondition() {
+ // Arrange
+ var containerName = await SetupContainer("race-retry");
+ var blobName = $"{DefaultStream}/ConcurrentState.json";
+
+ await SetupExistingBlob(containerName, blobName, new ConcurrentState { Value = 1 });
+
+ var projector = new ConcurrentModificationProjector(
+ fixture.BlobServiceClient,
+ containerName,
+ messWithState: OverwriteBlob(containerName, blobName),
+ raceRetries: 1);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // Act
+ var result = await projector.HandleEvent(context);
+
+ // Assert - with retry, this should succeed
+ await AssertSuccess(result);
+
+ var state = await GetBlobState(containerName, blobName);
+ // First attempt: concurrent modification sets value to 999, causing 412
+ // Retry: reads 999, adds 10, succeeds
+ await Assert.That(state.Value).IsEqualTo(1009); // 999 + 10 (retry succeeded)
+ }
+
+ [Test]
+ public async Task ConcurrentAdditionOfNewBlob_ShouldReturnFailure() {
+ // Arrange
+ var containerName = await SetupContainer("concurrent-new");
+ var blobName = $"{DefaultStream}/ConcurrentState.json";
+
+ var projector = new ConcurrentModificationProjector(
+ fixture.BlobServiceClient,
+ containerName,
+ messWithState: OverwriteBlob(containerName, blobName),
+ onCall: 1);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // This should now fail with 412 because the ETag won't match
+ var result2 = await projector.HandleEvent(context);
+ await AssertFailure(result2);
+ }
+
+ [Test]
+ public async Task ConcurrentModificationOfExistingBlob_ShouldReturnFailure() {
+ // Arrange
+ var containerName = await SetupContainer("concurrent-existing");
+ var blobName = $"{DefaultStream}/ConcurrentState.json";
+
+ await SetupExistingBlob(containerName, blobName, new ConcurrentState { Value = 1 });
+
+ var projector = new ConcurrentModificationProjector(
+ fixture.BlobServiceClient,
+ containerName,
+ messWithState: OverwriteBlob(containerName, blobName),
+ onCall: 2);
+ var context = CreateContext(new TestEvent { Value = 10 });
+
+ // First update should succeed
+ var result1 = await projector.HandleEvent(context);
+ await AssertSuccess(result1);
+
+ // This should now fail with 412 because the ETag won't match
+ var result2 = await projector.HandleEvent(context);
+ await AssertFailure(result2);
+ }
+
+ // ========== IDEMPOTENCY TESTS ==========
+
+ [Test]
+ public async Task Idempotency_ByMessageId_ShouldIgnoreDuplicateMessage() {
+ // Arrange
+ var containerName = await SetupContainer("idempotency-messageid");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ var projector = new IdempotencyProjector(fixture.BlobServiceClient, containerName, IdempotencyMode.ByMessageId);
+ var messageId = Guid.NewGuid().ToString();
+
+ // First context with specific message ID
+ var context1 = CreateContext(new TestEvent { Value = 10 }, messageId: messageId);
+
+ // Act - first processing should succeed
+ var result1 = await projector.HandleEvent(context1);
+ await AssertSuccess(result1);
+
+ var state1 = await GetBlobState(containerName, blobName);
+ await Assert.That(state1.Value).IsEqualTo(10);
+
+ // Second context with SAME message ID (duplicate)
+ var context2 = CreateContext(new TestEvent { Value = 20 }, messageId: messageId);
+
+ // Act - second processing should be ignored
+ var result2 = await projector.HandleEvent(context2);
+ await AssertIgnored(result2);
+
+ // State should NOT have been updated (still 10, not 30)
+ var state2 = await GetBlobState(containerName, blobName);
+ await Assert.That(state2.Value).IsEqualTo(10);
+ }
+
+ [Test]
+ public async Task Idempotency_ByMessageId_ShouldProcessDifferentMessageId() {
+ // Arrange
+ var containerName = await SetupContainer("idempotency-messageid-different");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ var projector = new IdempotencyProjector(fixture.BlobServiceClient, containerName, IdempotencyMode.ByMessageId);
+
+ var messageId1 = Guid.NewGuid().ToString();
+ var context1 = CreateContext(new TestEvent { Value = 10 }, messageId: messageId1);
+
+ // Act - first message
+ var result1 = await projector.HandleEvent(context1);
+ await AssertSuccess(result1);
+
+ var state1 = await GetBlobState(containerName, blobName);
+ await Assert.That(state1.Value).IsEqualTo(10);
+
+ // Different message ID
+ var messageId2 = Guid.NewGuid().ToString();
+ var context2 = CreateContext(new TestEvent { Value = 20 }, messageId: messageId2);
+
+ // Act - different message should be processed
+ var result2 = await projector.HandleEvent(context2);
+ await AssertSuccess(result2);
+
+ // State should have been updated (10 + 20 = 30)
+ var state2 = await GetBlobState(containerName, blobName);
+ await Assert.That(state2.Value).IsEqualTo(30);
+ }
+
+ [Test]
+ [Arguments(100u)]
+ [Arguments(99u)]
+ public async Task Idempotency_ByGlobalPosition_ShouldIgnoreDuplicatePosition(ulong duplicatePosition) {
+ // Arrange
+ var containerName = await SetupContainer("idempotency-globalposition");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ var projector = new IdempotencyProjector(fixture.BlobServiceClient, containerName, IdempotencyMode.ByGlobalPosition);
+
+ // First context with specific global position
+ var context1 = CreateContext(new TestEvent { Value = 10 }, globalPosition: 100u);
+
+ // Act - first processing should succeed
+ var result1 = await projector.HandleEvent(context1);
+ await AssertSuccess(result1);
+
+ var state1 = await GetBlobState(containerName, blobName);
+ await Assert.That(state1.Value).IsEqualTo(10);
+
+ // Second context with SAME global position (duplicate)
+ var context2 = CreateContext(new TestEvent { Value = 20 }, globalPosition: duplicatePosition);
+
+ // Act - second processing should be ignored
+ var result2 = await projector.HandleEvent(context2);
+ await AssertIgnored(result2);
+
+ // State should NOT have been updated (still 10, not 30)
+ var state2 = await GetBlobState(containerName, blobName);
+ await Assert.That(state2.Value).IsEqualTo(10);
+ }
+
+ [Test]
+ public async Task Idempotency_ByGlobalPosition_ShouldProcessDifferentPosition() {
+ // Arrange
+ var containerName = await SetupContainer("idempotency-globalposition-different");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ var projector = new IdempotencyProjector(fixture.BlobServiceClient, containerName, IdempotencyMode.ByGlobalPosition);
+
+ // First context with specific global position
+ var context1 = CreateContext(new TestEvent { Value = 10 }, globalPosition: 100);
+
+ // Act - first processing should succeed
+ var result1 = await projector.HandleEvent(context1);
+ await AssertSuccess(result1);
+
+ var state1 = await GetBlobState(containerName, blobName);
+ await Assert.That(state1.Value).IsEqualTo(10);
+
+ // Different global position
+ var context2 = CreateContext(new TestEvent { Value = 20 }, globalPosition: 101u);
+
+ // Act - different position should be processed
+ var result2 = await projector.HandleEvent(context2);
+ await AssertSuccess(result2);
+
+ // State should have been updated (10 + 20 = 30)
+ var state2 = await GetBlobState(containerName, blobName);
+ await Assert.That(state2.Value).IsEqualTo(30);
+ }
+
+ [Test]
+ public async Task Idempotency_None_ShouldAlwaysProcess() {
+ // Arrange - explicitly set to None (which is also the default)
+ var containerName = await SetupContainer("idempotency-none");
+ var blobName = $"{DefaultStream}/SyncState.json";
+
+ var projector = new IdempotencyProjector(fixture.BlobServiceClient, containerName, IdempotencyMode.None);
+ var messageId = Guid.NewGuid().ToString();
+
+ // First context
+ var context1 = CreateContext(new TestEvent { Value = 10 }, messageId: messageId);
+
+ // Act
+ var result1 = await projector.HandleEvent(context1);
+ await AssertSuccess(result1);
+
+ var state1 = await GetBlobState(containerName, blobName);
+ await Assert.That(state1.Value).IsEqualTo(10);
+
+ // Second context with SAME message ID - should still process
+ var context2 = CreateContext(new TestEvent { Value = 20 }, messageId: messageId);
+
+ // Act - should process even with same message ID
+ var result2 = await projector.HandleEvent(context2);
+ await AssertSuccess(result2);
+
+ // State should have been updated (10 + 20 = 30) - no idempotency
+ var state2 = await GetBlobState(containerName, blobName);
+ await Assert.That(state2.Value).IsEqualTo(30);
+ }
+
+ // ========== TEST CONTEXT FACTORY ==========
+
+ static IMessageConsumeContext CreateContext(object message, string? messageId = null, ulong globalPosition = 0, string stream = DefaultStream) =>
+ new MessageConsumeContext(
+ eventId: messageId ?? Guid.NewGuid().ToString(),
+ eventType: message.GetType().Name,
+ contentType: "application/json",
+ stream: stream,
+ eventNumber: 0,
+ streamPosition: 0,
+ globalPosition: globalPosition,
+ sequence: 0,
+ created: DateTime.UtcNow,
+ message: message,
+ metadata: new Metadata(),
+ subscriptionId: "test-subscription",
+ cancellationToken: CancellationToken.None
+ );
+
+ // ========== TEST STATE CLASSES ==========
+
+ class SyncState {
+ public int Value { get; set; }
+ public int Counter { get; set; }
+ }
+
+ class SyncContextState {
+ public int Value { get; set; }
+ public string StreamId { get; set; } = "";
+ }
+
+ class AsyncState {
+ public int Value { get; set; }
+ }
+
+ class AsyncContextState {
+ public int Value { get; set; }
+ public string EventName { get; set; } = "";
+ }
+
+ class ConcurrentState {
+ public int Value { get; set; }
+ }
+
+ class NoHandlerState { }
+
+ class CustomBlobIdState {
+ public int Value { get; set; }
+ }
+
+ // ========== TEST PROJECTOR CLASSES
+ // Intent: Each class name explicitly surfaces the handler pattern being tested ==========
+
+ ///
+ /// Tests sync handler: On(Func)
+ ///
+ class SyncStateProjector : BlobStorageProjector {
+ public SyncStateProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) {
+ On((ctx, state) => {
+ state.Value += ctx.Message.Value;
+ state.Counter++;
+ return state;
+ });
+ }
+ }
+
+ ///
+ /// Tests sync context-aware handler: On(Func) with context access
+ ///
+ class SyncContextAwareProjector : BlobStorageProjector {
+ public SyncContextAwareProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) {
+ On((ctx, state) => {
+ state.Value += ctx.Message.Value;
+ state.StreamId = ctx.Stream.GetId();
+ return state;
+ });
+ }
+ }
+
+ ///
+ /// Tests async handler: On(Func>)
+ ///
+ class AsyncStateProjector : BlobStorageProjector {
+ public AsyncStateProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) {
+ On(async (ctx, state) => {
+ await Task.Delay(1);
+ state.Value += ctx.Message.Value;
+ return state;
+ });
+ }
+ }
+
+ ///
+ /// Tests async context-aware handler: On(Func>) with context access
+ ///
+ class AsyncContextAwareProjector : BlobStorageProjector {
+ public AsyncContextAwareProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) {
+ On(async (ctx, state) => {
+ await Task.Delay(1);
+ state.Value += ctx.Message.Value;
+ state.EventName = ctx.Message.Name;
+ return state;
+ });
+ }
+ }
+
+ ///
+ /// Tests scenario with no handlers registered
+ ///
+ class NoHandlerProjector : BlobStorageProjector {
+ public NoHandlerProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) { }
+ }
+
+ ///
+ /// Tests concurrent modification scenario with configurable race retries and onCall
+ ///
+ class ConcurrentModificationProjector : BlobStorageProjector {
+ int _callCount;
+ readonly Func? _messWithState;
+ readonly int _onCall;
+
+ public ConcurrentModificationProjector(
+ BlobServiceClient serviceClient,
+ string containerName,
+ Func? messWithState = null,
+ int onCall = 1,
+ int raceRetries = 0
+ ) : base(serviceClient, containerName, projectorOptions: new BlobStorageProjectorOptions { RaceRetries = raceRetries }) {
+ _messWithState = messWithState;
+ _onCall = onCall;
+
+ On(async (ctx, state) => {
+ if (_messWithState != null && ++_callCount == _onCall)
+ await _messWithState();
+ state.Value += ctx.Message.Value;
+ return state;
+ });
+ }
+ }
+
+ ///
+ /// Tests custom blob ID using getBlobId parameter
+ ///
+ class CustomBlobIdProjector : BlobStorageProjector {
+ public CustomBlobIdProjector(BlobServiceClient serviceClient, string containerName)
+ : base(serviceClient, containerName) {
+ On(async (ctx, state) => {
+ state.Value += ctx.Message.Value;
+ return state;
+ }, getBlobId: ctx => new ValueTask(ctx.Message.Id));
+ }
+ }
+
+ ///
+ /// Tests that a handler-thrown RequestFailedException is not mistaken for a blob race
+ ///
+ class ThrowingHandlerProjector : BlobStorageProjector {
+ public int HandlerCalls { get; private set; }
+
+ public ThrowingHandlerProjector(BlobServiceClient serviceClient, string containerName, int raceRetries)
+ : base(serviceClient, containerName, new BlobStorageProjectorOptions { RaceRetries = raceRetries }) {
+ Func, SyncState, SyncState> handler = (_, _) => {
+ HandlerCalls++;
+ throw new RequestFailedException(409, "Handler-side conflict talking to another service");
+ };
+ On(handler);
+ }
+ }
+
+ ///
+ /// Tests idempotency with configurable mode
+ ///
+ class IdempotencyProjector : BlobStorageProjector {
+ public IdempotencyProjector(BlobServiceClient serviceClient, string containerName, IdempotencyMode mode)
+ : base(serviceClient, containerName, projectorOptions: new BlobStorageProjectorOptions { IdempotencyMode = mode }) {
+ On((ctx, state) => {
+ state.Value += ctx.Message.Value;
+ return state;
+ });
+ }
+ }
+}
diff --git a/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Eventuous.Tests.Azure.Storage.Blobs.csproj b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Eventuous.Tests.Azure.Storage.Blobs.csproj
new file mode 100644
index 000000000..49438c888
--- /dev/null
+++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Eventuous.Tests.Azure.Storage.Blobs.csproj
@@ -0,0 +1,17 @@
+
+
+
+
+
+
+
+
+
+
+
+ all
+ runtime; build; native; contentfiles; analyzers; buildtransitive
+
+
+
+
diff --git a/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Fixtures/IntegrationFixture.cs b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Fixtures/IntegrationFixture.cs
new file mode 100644
index 000000000..79bf5c244
--- /dev/null
+++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Fixtures/IntegrationFixture.cs
@@ -0,0 +1,24 @@
+using Azure.Storage.Blobs;
+using Testcontainers.Azurite;
+using TUnit.Core.Interfaces;
+
+namespace Eventuous.Tests.Azure.Storage.Blobs.Fixtures;
+
+public sealed class IntegrationFixture : IAsyncInitializer, IAsyncDisposable {
+ public BlobServiceClient BlobServiceClient { get; private set; } = null!;
+
+ AzuriteContainer _azuriteContainer = null!;
+
+ public async Task InitializeAsync() {
+ // Start Azurite container for blob storage
+ _azuriteContainer = new AzuriteBuilder()
+ .WithImage("mcr.microsoft.com/azure-storage/azurite:latest")
+ .WithCommand("--skipApiVersionCheck")
+ .Build();
+ await _azuriteContainer.StartAsync();
+
+ BlobServiceClient = new BlobServiceClient(_azuriteContainer.GetConnectionString());
+ }
+
+ public async ValueTask DisposeAsync() => await _azuriteContainer.DisposeAsync();
+}
diff --git a/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/TestEvent.cs b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/TestEvent.cs
new file mode 100644
index 000000000..c41a76eb5
--- /dev/null
+++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/TestEvent.cs
@@ -0,0 +1,14 @@
+namespace Eventuous.Tests.Azure.Storage.Blobs;
+
+[EventType("V1.TestEvent")]
+public record TestEvent {
+ static TestEvent() => TypeMap.RegisterKnownEventTypes(typeof(TestEvent).Assembly);
+
+ public string Id { get; set; } = Guid.NewGuid().ToString();
+ public string Name { get; set; } = "Test Event";
+ public int Value { get; set; } = 42;
+
+ public static TestEvent Create() => new() { Id = "test-event", Name = "Test", Value = 1 };
+
+ public static TestEvent Create(int value) => new() { Id = "test-event", Name = "Test", Value = value };
+}
diff --git a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs
index 8268287b9..6e69d92c0 100644
--- a/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs
+++ b/src/Core/src/Eventuous.Persistence/EventStore/IEventReader.cs
@@ -7,6 +7,12 @@ public interface IEventReader {
///
/// Read a fixed number of events from an existing stream as an async enumerable.
/// Throws if the stream does not exist.
+ /// Implementations either stream events as they arrive from the store, or buffer events in an amount
+ /// proportional to before yielding, so memory usage can grow with
+ /// . To read a whole stream, use ,
+ /// which reads in pages, instead of passing as the count.
+ /// Implementations must yield exactly events unless the end of the stream is reached,
+ /// and must return an empty sequence, not throw, when reading past the end of an existing stream.
///
/// Stream name
/// Where to start reading events
@@ -18,6 +24,9 @@ public interface IEventReader {
///
/// Read a number of events from a given stream, backwards (from the stream end).
/// Throws if the stream does not exist.
+ /// Implementations either stream events as they arrive from the store, or buffer events in an amount
+ /// proportional to before yielding, so memory usage can grow with
+ /// .
///
/// Stream name
/// Where to start reading events
diff --git a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs
index ca2ee79f2..18a5f430f 100644
--- a/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs
+++ b/src/Core/src/Eventuous.Persistence/EventStore/StoreFunctions.cs
@@ -1,6 +1,8 @@
// Copyright (C) Eventuous HQ OÜ. All rights reserved
// Licensed under the Apache License, Version 2.0.
+using System.Runtime.CompilerServices;
+
namespace Eventuous;
public static class StoreFunctions {
@@ -148,6 +150,33 @@ CancellationToken cancellationToken
}
}
+ ///
+ /// Reads a stream from the given position to the end, as an async enumerable.
+ /// Events are read in pages of and yielded as they arrive, so the whole stream
+ /// is never buffered in memory. Use this instead of calling
+ /// with as the count.
+ ///
+ /// Name of the stream to read from
+ /// Stream position to start reading from
+ /// Number of events to read per page. It bounds the memory a buffering
+ /// implementation of uses: such implementations hold at most a small
+ /// multiple of a page in memory at a time (e.g. a tiered reader combining two stores).
+ /// Set to false to complete without yielding anything when the stream isn't found,
+ /// instead of throwing . Default is true.
+ /// Cancellation token
+ /// An async enumerable of events retrieved from the stream
+ public IAsyncEnumerable ReadStreamToEnd(
+ StreamName streamName,
+ StreamReadPosition start,
+ int pageSize = 500,
+ bool failIfNotFound = true,
+ CancellationToken cancellationToken = default
+ ) {
+ ArgumentOutOfRangeException.ThrowIfNegativeOrZero(pageSize);
+
+ return ReadToEnd(eventReader, streamName, start, pageSize, failIfNotFound, cancellationToken);
+ }
+
///
/// Reads a stream from the event store to a collection of
///
@@ -163,26 +192,58 @@ public async Task ReadStream(
bool failIfNotFound = true,
CancellationToken cancellationToken = default
) {
- const int pageSize = 500;
-
var streamEvents = new List();
- var position = start;
-
- try {
- while (true) {
- var events = await eventReader.ReadEvents(streamName, position, pageSize, failIfNotFound, cancellationToken).NoContext();
- streamEvents.AddRange(events);
+ await foreach (var evt in eventReader.ReadStreamToEnd(streamName, start, failIfNotFound: failIfNotFound, cancellationToken: cancellationToken).NoContext(cancellationToken)) {
+ streamEvents.Add(evt);
+ }
- if (events.Length < pageSize) break;
+ return [.. streamEvents];
+ }
+ }
- position = new(position.Value + events.Length);
+ // Relies on readers yielding exactly `count` events unless the stream end is reached:
+ // a page shorter than pageSize means there is nothing left to read
+ static async IAsyncEnumerable ReadToEnd(
+ IEventReader eventReader,
+ StreamName streamName,
+ StreamReadPosition start,
+ int pageSize,
+ bool failIfNotFound,
+ [EnumeratorCancellation] CancellationToken cancellationToken
+ ) {
+ var position = start;
+
+ while (true) {
+ var yielded = 0;
+ long lastRevision = 0;
+
+ await using var enumerator = eventReader.ReadEvents(streamName, position, pageSize, cancellationToken).GetAsyncEnumerator(cancellationToken);
+
+ while (true) {
+ bool moved;
+
+ try {
+ moved = await enumerator.MoveNextAsync().NoContext();
+ } catch (StreamNotFound) when (!failIfNotFound) {
+ yield break;
}
- } catch (StreamNotFound) when (!failIfNotFound) {
- return [];
+
+ if (!moved) break;
+
+ var evt = enumerator.Current;
+ yielded++;
+ lastRevision = evt.Revision;
+
+ yield return evt;
}
- return [.. streamEvents];
+ if (yielded < pageSize) yield break;
+
+ // The maximum revision is the end of the representable position space
+ if (lastRevision == long.MaxValue) yield break;
+
+ position = new(lastRevision + 1);
}
}
}
diff --git a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs
index 3095489b8..08d6ffe3b 100644
--- a/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs
+++ b/src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs
@@ -13,16 +13,29 @@ namespace Eventuous;
/// Event reader pointing to archive store
public class TieredEventReader(IEventReader hotReader, IEventReader archiveReader) : IEventReader {
public async IAsyncEnumerable ReadEvents(StreamName streamName, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) {
- var hotEvents = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken).NoContext();
+ var (hotEvents, hotNotFound) = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken).NoContext();
- var archivedEvents = hotEvents.Length switch {
- > 0 when hotEvents[0].Revision > start.Value
- => (await LoadStreamEvents(archiveReader, streamName, start, (int)hotEvents[0].Revision, cancellationToken).NoContext()).Select(x => x with { FromArchive = true }),
- 0 => (await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken).NoContext()).Select(x => x with { FromArchive = true }),
- _ => []
- };
+ IEnumerable archivedEvents;
+ var archiveNotFound = false;
+
+ switch (hotEvents.Length) {
+ case > 0 when hotEvents[0].Revision > start.Value: {
+ // Fill the gap before the first hot event from the archive, bounded by the requested count
+ var gapCount = (int)Math.Min(count, hotEvents[0].Revision - start.Value);
+
+ (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, gapCount, cancellationToken).NoContext();
+ archivedEvents = events.Select(x => x with { FromArchive = true });
+
+ break;
+ }
+ case 0:
+ (var archived, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken).NoContext();
+ archivedEvents = archived.Select(x => x with { FromArchive = true }); break;
+ default:
+ archivedEvents = []; break;
+ }
- var combined = archivedEvents.Concat(hotEvents).Distinct(Comparer);
+ var combined = archivedEvents.Concat(hotEvents).Distinct(Comparer).Take(count);
var any = false;
foreach (var evt in combined) {
@@ -31,28 +44,32 @@ public async IAsyncEnumerable ReadEvents(StreamName streamName, Str
yield return evt;
}
- if (!any) throw new StreamNotFound(streamName);
+ // No events with both tiers reporting a missing stream means the stream doesn't exist;
+ // otherwise an empty result can mean the read window is past the stream end
+ if (!any && hotNotFound && archiveNotFound) throw new StreamNotFound(streamName);
}
public async IAsyncEnumerable ReadEventsBackwards(StreamName streamName, StreamReadPosition start, int count, [EnumeratorCancellation] CancellationToken cancellationToken) {
- var hotEvents = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken, backwards: true).NoContext();
+ var (hotEvents, hotNotFound) = await LoadStreamEvents(hotReader, streamName, start, count, cancellationToken, backwards: true).NoContext();
IEnumerable archivedEvents;
+ var archiveNotFound = false;
switch (hotEvents.Length) {
- case > 0 when hotEvents.Length < count: {
+ // When the hot store read reached revision 0, no events can precede it
+ case > 0 when hotEvents.Length < count && hotEvents[^1].Revision > 0: {
// Hot store returned fewer events than requested, fill the gap from archive
var lastHotRevision = hotEvents[^1].Revision;
- archivedEvents = (await LoadStreamEvents(archiveReader, streamName, new(lastHotRevision - 1), count - hotEvents.Length, cancellationToken, backwards: true).NoContext())
- .Select(x => x with { FromArchive = true });
+ (var events, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, new(lastHotRevision - 1), count - hotEvents.Length, cancellationToken, backwards: true).NoContext();
+ archivedEvents = events.Select(x => x with { FromArchive = true });
break;
}
case 0:
// Hot store has no events, try archive for the full range
- archivedEvents = (await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken, backwards: true).NoContext())
- .Select(x => x with { FromArchive = true }); break;
+ (var archived, archiveNotFound) = await LoadStreamEvents(archiveReader, streamName, start, count, cancellationToken, backwards: true).NoContext();
+ archivedEvents = archived.Select(x => x with { FromArchive = true }); break;
default:
archivedEvents = []; break;
}
@@ -66,10 +83,12 @@ public async IAsyncEnumerable ReadEventsBackwards(StreamName stream
yield return evt;
}
- if (!any) throw new StreamNotFound(streamName);
+ // No events with both tiers reporting a missing stream means the stream doesn't exist;
+ // otherwise an empty result can mean the read window is past the stream end
+ if (!any && hotNotFound && archiveNotFound) throw new StreamNotFound(streamName);
}
- static async Task LoadStreamEvents(
+ static async Task<(StreamEvent[] Events, bool NotFound)> LoadStreamEvents(
IEventReader reader,
StreamName streamName,
StreamReadPosition startPosition,
@@ -78,11 +97,13 @@ static async Task LoadStreamEvents(
bool backwards = false
) {
try {
- return backwards
+ var events = backwards
? await reader.ReadEventsBackwards(streamName, startPosition, localCount, true, cancellationToken).NoContext()
: await reader.ReadEvents(streamName, startPosition, localCount, true, cancellationToken).NoContext();
+
+ return (events, false);
} catch (StreamNotFound) {
- return [];
+ return ([], true);
}
}
diff --git a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs
index 25adbbfba..fbec3133c 100644
--- a/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs
+++ b/src/Core/test/Eventuous.Tests.Persistence.Base/Store/Read.cs
@@ -146,6 +146,117 @@ public async Task ShouldReturnWhenReadingBackwards(CancellationToken cancellatio
await Assert.That(result.Length).IsEqualTo(5);
}
+ [Test]
+ [Category("Store")]
+ public async Task ShouldThrowWhenReadingMissingStream(CancellationToken cancellationToken) {
+ var streamName = Helpers.GetStreamName();
+
+ await Assert.ThrowsAsync(() => _fixture.EventStore.ReadEvents(streamName, StreamReadPosition.Start, 10, true, cancellationToken));
+ }
+
+ [Test]
+ [Category("Store")]
+ public async Task ShouldThrowWhenReadingMissingStreamBackwards(CancellationToken cancellationToken) {
+ var streamName = Helpers.GetStreamName();
+
+ await Assert.ThrowsAsync(() => _fixture.EventStore.ReadEventsBackwards(streamName, StreamReadPosition.End, 10, true, cancellationToken));
+ }
+
+ [Test]
+ [Category("Store")]
+ public async Task ShouldReadStreamToEnd(CancellationToken cancellationToken) {
+ object[] events = [.. _fixture.CreateEvents(25)];
+ var streamName = Helpers.GetStreamName();
+ await _fixture.AppendEvents(streamName, events, ExpectedStreamVersion.NoStream);
+
+ var result = new List();
+
+ await foreach (var evt in _fixture.EventStore.ReadStreamToEnd(streamName, StreamReadPosition.Start, pageSize: 10, cancellationToken: cancellationToken)) {
+ result.Add(evt);
+ }
+
+ IEnumerable