Aspire and azure blob storage

Directory.Packages.props

@@ -22,10 +22,12 @@
     <TUnitVersion>0.77.3</TUnitVersion>
   </PropertyGroup>
   <ItemGroup>
+    <PackageVersion Include="Azure.Storage.Blobs" Version="12.29.0" />
     <PackageVersion Include="BenchmarkDotNet" Version="0.14.0" />
     <PackageVersion Include="FluentValidation" Version="12.0.0" />
     <PackageVersion Include="IsExternalInit" Version="1.0.3" />
     <PackageVersion Include="KurrentDB.Client" Version="1.3.0" />
+    <PackageVersion Include="Microsoft.Extensions.Azure" Version="1.14.0" />
     <PackageVersion Include="Microsoft.Extensions.Configuration" Version="$(MicrosoftExtensionsVer)" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.Json" Version="$(MicrosoftExtensionsVer)" />
     <PackageVersion Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="$(MicrosoftExtensionsVer)" />
@@ -37,6 +39,7 @@
     <PackageVersion Include="Microsoft.Extensions.Logging" Version="$(MicrosoftExtensionsVer)" />
     <PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="$(MicrosoftExtensionsVer)" />
     <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.11.1" />
+    <PackageVersion Include="Scalar.Aspire" Version="0.11.0" />
     <PackageVersion Include="Shouldly" Version="4.2.1" />
     <PackageVersion Include="System.Diagnostics.DiagnosticSource" Version="$(MicrosoftExtensionsVer)" />
     <PackageVersion Include="System.Linq.AsyncEnumerable" Version="10.0.0" />
@@ -62,6 +65,7 @@
   </ItemGroup>
   <ItemGroup Label="Testcontainers">
     <PackageVersion Include="Testcontainers" Version="$(TestcontainersVersion)" />
+    <PackageVersion Include="Testcontainers.Azurite" Version="$(TestcontainersVersion)" />
     <PackageVersion Include="Testcontainers.KurrentDb" Version="$(TestcontainersVersion)" />
     <PackageVersion Include="Testcontainers.Kafka" Version="$(TestcontainersVersion)" />
     <PackageVersion Include="Testcontainers.MongoDb" Version="$(TestcontainersVersion)" />
@@ -112,6 +116,7 @@
     <PackageVersion Include="OpenTelemetry.Exporter.Zipkin" Version="1.15.3" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.15.2" />
     <PackageVersion Include="OpenTelemetry.Instrumentation.GrpcNetClient" Version="1.15.1-beta.1" />
+    <PackageVersion Include="OpenTelemetry.Instrumentation.SqlClient" Version="1.15.2" />
     <PackageVersion Include="Serilog.AspNetCore" Version="9.0.0" />
     <PackageVersion Include="Serilog.Extensions.Hosting" Version="9.0.0" />
     <PackageVersion Include="Serilog.Sinks.Console" Version="6.0.0" />

Eventuous.slnx

@@ -14,9 +14,11 @@
     <Folder Name="/Brokers/Azure/" />
     <Folder Name="/Brokers/Azure/src/">
         <Project Path="src/Azure/src/Eventuous.Azure.ServiceBus/Eventuous.Azure.ServiceBus.csproj" />
+        <Project Path="src/Azure/src/Eventuous.Azure.Storage.Blobs/Eventuous.Azure.Storage.Blobs.csproj" />
     </Folder>
     <Folder Name="/Brokers/Azure/test/">
         <Project Path="src/Azure/test/Eventuous.Tests.Azure.ServiceBus/Eventuous.Tests.Azure.ServiceBus.csproj" />
+        <Project Path="src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Eventuous.Tests.Azure.Storage.Blobs.csproj" />
     </Folder>
     <Folder Name="/Brokers/GooglePubSub/" />
     <Folder Name="/Brokers/GooglePubSub/src/">

src/Azure/src/Eventuous.Azure.ServiceBus/Subscriptions/ServiceBusSubscription.cs

@@ -23,7 +23,7 @@ public class ServiceBusSubscription : EventSubscription<ServiceBusSubscriptionOp
     /// <param name="consumePipe">Consume pipe instance</param>
     /// <param name="loggerFactory">Logger factory (optional)</param>
     /// <param name="eventSerializer">Event serializer (optional)</param>
-    public ServiceBusSubscription(ServiceBusClient client, ServiceBusSubscriptionOptions options, ConsumePipe consumePipe, ILoggerFactory? loggerFactory, IEventSerializer? eventSerializer) :
+    public ServiceBusSubscription(ServiceBusClient client, ServiceBusSubscriptionOptions options, ConsumePipe consumePipe, ILoggerFactory? loggerFactory, IEventSerializer? eventSerializer = null) :
         base(options, consumePipe, loggerFactory, eventSerializer) {
         _defaultErrorHandler = Options.ErrorHandler ?? DefaultErrorHandler;
 

src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjector.cs

@@ -0,0 +1,232 @@
+using Azure;
+using Azure.Storage.Blobs.Models;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Context;
+using System.Text.Json;
+
+using static Eventuous.Subscriptions.Diagnostics.SubscriptionsEventSource;
+
+namespace Eventuous.Azure.Storage.Blobs;
+
+/// <summary>
+/// Projects event store events to Azure Blob Storage as state objects of type T.
+/// </summary>
+/// <remarks>
+/// <para>
+/// 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.
+/// </para>
+/// </remarks>
+public class BlobStorageProjector<T> : BaseEventHandler where T : class, new() {
+    /// <summary>Azure Blob Storage container client.</summary>
+    protected readonly BlobContainerClient ContainerClient;
+
+    readonly JsonSerializerOptions _jsonOptions;
+    readonly Dictionary<Type, Func<IMessageConsumeContext, ValueTask<EventHandlingStatus>>> _handlers = new();
+    readonly ITypeMapper _map;
+
+    private readonly int _raceRetries;
+    private readonly IdempotencyMode _idempotencyMode;
+
+    /// <summary>Delegate for custom blob ID generation from consume context.</summary>
+    /// <typeparam name="TEvent">Event type being consumed.</typeparam>
+    /// <param name="context">Event consume context.</param>
+    /// <returns>Blob ID as string.</returns>
+    public delegate ValueTask<string> GetBlobId<TEvent>(IMessageConsumeContext<TEvent> context) where TEvent : class;
+
+    /// <summary>
+    /// Initializes projector with existing container client.
+    /// </summary>
+    /// <param name="container">Azure Blob Storage container client.</param>
+    /// <param name="projectorOptions">Optional projector configuration.</param>
+    /// <param name="serializerOptions">Optional JSON serializer options.</param>
+    /// <param name="mapper">Optional type mapper for event type resolution.</param>
+    public BlobStorageProjector(
+        BlobContainerClient container,
+        JsonSerializerOptions? serializerOptions = null,
+        BlobStorageProjectorOptions? projectorOptions = null,
+        ITypeMapper? mapper = null
+    ) {
+        ContainerClient = container;
+        _jsonOptions = new(projectorOptions?.JsonOptions ?? serializerOptions ?? JsonSerializerOptions.Web);
+        _map = mapper ?? TypeMap.Instance;
+        _raceRetries = projectorOptions?.RaceRetries ?? 0;
+        _idempotencyMode = projectorOptions?.IdempotencyMode ?? IdempotencyMode.None;
+    }
+
+    /// <summary>
+    /// Initializes projector with service client and container name.
+    /// </summary>
+    /// <param name="serviceClient">Azure Blob Storage service client.</param>
+    /// <param name="containerName">Name of the container to use.</param>
+    /// <param name="serializerOptions">Optional JSON serializer options.</param>
+    /// <param name="mapper">Optional type mapper for event type resolution.</param>
+    /// <param name="projectorOptions">Optional projector configuration.</param>
+    public BlobStorageProjector(
+        BlobServiceClient serviceClient,
+        string containerName,
+        JsonSerializerOptions? serializerOptions = null,
+        BlobStorageProjectorOptions? projectorOptions = null,
+        ITypeMapper? mapper = null
+    ) : this(serviceClient.GetBlobContainerClient(containerName), serializerOptions, projectorOptions, mapper) { }
+
+    /// <summary>Registers event handler with sync state update.</summary>
+    /// <typeparam name="TEvent">Event type to handle.</typeparam>
+    /// <param name="handler">State update function receiving current state and event.</param>
+    /// <param name="getBlobId">Optional custom blob ID generator.</param>
+    protected void On<TEvent>(Func<T, TEvent, T> handler, GetBlobId<TEvent>? getBlobId = null) where TEvent : class
+        => On((ctx, state) => new ValueTask<T>(handler(state, ctx.Message)), getBlobId);
+
+    /// <summary>Registers event handler with context and sync state update.</summary>
+    /// <typeparam name="TEvent">Event type to handle.</typeparam>
+    /// <param name="handler">State update function receiving context, current state, and event.</param>
+    /// <param name="getBlobId">Optional custom blob ID generator.</param>
+    protected void On<TEvent>(Func<IMessageConsumeContext<TEvent>, T, T> handler, GetBlobId<TEvent>? getBlobId = null) where TEvent : class
+        => On((ctx, state) => new ValueTask<T>(handler(ctx, state)), getBlobId);
+
+    /// <summary>Registers event handler with async state update.</summary>
+    /// <typeparam name="TEvent">Event type to handle.</typeparam>
+    /// <param name="handler">Async state update function receiving current state and event.</param>
+    /// <param name="getBlobId">Optional custom blob ID generator.</param>
+    protected void On<TEvent>(Func<T, TEvent, ValueTask<T>> handler, GetBlobId<TEvent>? getBlobId = null) where TEvent : class
+        => On((ctx, state) => handler(state, ctx.Message), getBlobId);
+
+    /// <summary>Registers event handler with context, async state update, and custom blob ID.</summary>
+    /// <typeparam name="TEvent">Event type to handle.</typeparam>
+    /// <param name="handler">Async state update function receiving context, current state, and event.</param>
+    /// <param name="getBlobId">Optional custom blob ID generator.</param>
+    protected void On<TEvent>(Func<IMessageConsumeContext<TEvent>, T, ValueTask<T>> handler, GetBlobId<TEvent>? getBlobId = null) where TEvent : class {
+        if (!_handlers.TryAdd(typeof(TEvent), new Handler<TEvent>(this, handler, getBlobId).Handle)) {
+            throw new ArgumentException($"Type {typeof(TEvent).Name} already has a handler");
+        }
+
+        if (!_map.TryGetTypeName<TEvent>(out _)) {
+            Log.MessageTypeNotRegistered<TEvent>();
+        }
+    }
+
+    private BlobClient GetBlobContainerClient(string blobName) => ContainerClient.GetBlobClient(blobName);
+
+    /// <summary>Registers event handler without custom blob ID.</summary>
+    /// <typeparam name="TEvent">Event type to handle.</typeparam>
+    /// <param name="handler">Async state update function receiving context, current state, and event.</param>
+    protected void On<TEvent>(Func<IMessageConsumeContext<TEvent>, T, ValueTask<T>> handler) where TEvent : class
+        => On(handler, default);
+
+    /// <summary>Handles incoming event by dispatching to registered handler.</summary>
+    /// <param name="context">Event consume context.</param>
+    /// <returns>Event handling status indicating success, failure, or ignore.</returns>
+    public override async ValueTask<EventHandlingStatus> HandleEvent(IMessageConsumeContext context) =>
+        _handlers.TryGetValue(context.Message!.GetType(), out var handler)
+            ? await handler(context).NoContext()
+            : EventHandlingStatus.Ignored;
+
+    private T ToObjectFromJson(BinaryData content) => content.ToObjectFromJson<T>(_jsonOptions) ?? new T();
+    private byte[] SerializeToUtf8Bytes(T updated) => JsonSerializer.SerializeToUtf8Bytes(updated, _jsonOptions);
+
+    /// <summary>Gets blob name from ID and context. Can be overridden for custom naming.</summary>
+    /// <param name="id">Blob identifier.</param>
+    /// <param name="context">Event consume context.</param>
+    /// <returns>Blob name as string.</returns>
+    protected virtual string GetBlobName(string id, IMessageConsumeContext context) => GetBlobName(id);
+
+    /// <summary>Gets blob name from ID. Default format: {id}/{T}.json</summary>
+    /// <param name="id">Blob identifier.</param>
+    /// <returns>Blob name as string.</returns>
+    protected virtual string GetBlobName(string id) => $"{id}/{typeof(T).Name}.json";
+
+    private class Handler<TEvent>
+        where TEvent : class {
+        private readonly BlobStorageProjector<T> projector;
+        private readonly Func<IMessageConsumeContext<TEvent>, T, ValueTask<T>> EventHandler;
+        private readonly GetBlobId<TEvent>? GetBlobId;
+
+        public Handler(BlobStorageProjector<T> blobStorageProjector, Func<IMessageConsumeContext<TEvent>, T, ValueTask<T>> handler, GetBlobId<TEvent>? getBlobId) {
+            projector = blobStorageProjector;
+            EventHandler = handler;
+            GetBlobId = getBlobId;
+        }
+
+        public async ValueTask<EventHandlingStatus> Handle(IMessageConsumeContext context) {
+            var typedContext = context as MessageConsumeContext<TEvent> ?? new MessageConsumeContext<TEvent>(context);
+            var blobId = GetBlobId == null
+                ? context.Stream.GetId()
+                : await GetBlobId(typedContext).NoContext();
+            var blobName = projector.GetBlobName(blobId, typedContext);
+
+            var blobClient = projector.GetBlobContainerClient(blobName);
+
+            return await ModifyBlobWithRetries(projector._raceRetries).NoContext();
+
+            async Task<EventHandlingStatus> ModifyBlobWithRetries(int retries) {
+                try {
+                    return await ModifyBlob().NoContext();
+                } catch (RequestFailedException ex) when (ex.Status == 412 || ex.Status == 409) {
+                    return retries > 0 ? await ModifyBlobWithRetries(retries - 1).NoContext() : EventHandlingStatus.Failure;
+                }
+            }
+
+            async Task<EventHandlingStatus> ModifyBlob() {
+                try {
+                    var blobContent = await blobClient.DownloadContentAsync(typedContext.CancellationToken).NoContext();
+
+                    // Check idempotency if enabled
+                    if (projector._idempotencyMode != IdempotencyMode.None) {
+                        if (IsDuplicate(blobContent.Value.Details.Metadata)) {
+                            return EventHandlingStatus.Ignored;
+                        }
+                    }
+
+                    var content = blobContent.Value.Content;
+                    var current = projector.ToObjectFromJson(content);
+
+                    await UploadUpdated(current, new BlobRequestConditions { IfMatch = blobContent.Value.Details.ETag }).NoContext();
+                    return EventHandlingStatus.Success;
+                } catch (RequestFailedException ex) when (ex.Status == 404) {
+                    // Blob doesn't exist, start with a new instance
+                    await UploadUpdated(new T(), new BlobRequestConditions { IfNoneMatch = ETag.All }).NoContext();
+                    return EventHandlingStatus.Success;
+                }
+            }
+
+            bool IsDuplicate(IDictionary<string, string> metadata) => projector._idempotencyMode switch {
+                IdempotencyMode.ByGlobalPosition =>
+                    metadata.TryGetValue("GlobalPosition", out var storedPosition) &&
+                    storedPosition == typedContext.GlobalPosition.ToString(),
+                IdempotencyMode.ByMessageId =>
+                    metadata.TryGetValue("MessageId", out var storedId) &&
+                    storedId == typedContext.MessageId,
+                _ => false
+            };
+
+            async Task UploadUpdated(T current, BlobRequestConditions conditions) {
+                var task = EventHandler(typedContext, current);
+                var updated = task.IsCompletedSuccessfully
+                    ? task.Result
+                    : await task.NoContext();
+                var json = projector.SerializeToUtf8Bytes(updated);
+
+                var uploadOptions = new BlobUploadOptions {
+                    Conditions = conditions,
+                    HttpHeaders = new BlobHttpHeaders {
+                        ContentType = "application/json"
+                    },
+                    Metadata = new Dictionary<string, string> {
+                        ["Stream"] = typedContext.Stream.ToString(),
+                        ["MessageId"] = typedContext.MessageId,
+                        ["StreamPosition"] = typedContext.StreamPosition.ToString(),
+                        ["GlobalPosition"] = typedContext.GlobalPosition.ToString()
+                    }
+                };
+
+                using var stream = new MemoryStream(json);
+                var response = await blobClient.UploadAsync(stream, uploadOptions, typedContext.CancellationToken).NoContext();
+            }
+        }
+    }
+}

src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjectorOptions.cs

@@ -0,0 +1,53 @@
+using System.Text.Json;
+
+namespace Eventuous.Azure.Storage.Blobs;
+
+/// <summary>
+/// Options for configuring the storage blob projector.
+/// </summary>
+public class BlobStorageProjectorOptions {
+    /// <summary>
+    /// Gets or sets the JSON serializer options to use when serializing or deserializing projection state.
+    /// By default, the default JSON serializer options will be used if this property is not set.
+    /// </summary>
+    public JsonSerializerOptions? JsonOptions { get; set; }
+
+    /// <summary>
+    /// Gets or sets the number of retry attempts for race condition handling when saving projection state.
+    /// Default is 0 (no retries).
+    /// </summary>
+    public int RaceRetries { get; set; } = 0;
+
+    /// <summary>
+    /// 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 <see cref="IdempotencyMode.None"/> (no idempotency checking).
+    /// </summary>
+    public IdempotencyMode IdempotencyMode { get; set; } = IdempotencyMode.None;
+}
+
+/// <summary>
+/// Controls how the projection handles idempotency to prevent duplicate message processing.
+/// </summary>
+public enum IdempotencyMode {
+    /// <summary>
+    /// 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.
+    /// </summary>
+    None,
+
+    /// <summary>
+    /// Skips processing if the existing blob was created from a message at the same global position.
+    /// Uses the <c>GlobalPosition</c> metadata stored with the blob for comparison.
+    /// Effective for append-only event streams where global position uniquely identifies a message.
+    /// </summary>
+    ByGlobalPosition,
+
+    /// <summary>
+    /// Skips processing if the existing blob was created from the same message ID.
+    /// Uses the <c>MessageId</c> 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.
+    /// </summary>
+    ByMessageId
+}