diff --git a/Directory.Packages.props b/Directory.Packages.props index 59f4734dd..25aff8a3b 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -17,15 +17,24 @@ 4.10.0 + + 13.4.6 + 10.0.1 0.77.3 + 1.16.0 + + + + + @@ -37,12 +46,13 @@ + - - + + @@ -62,6 +72,7 @@ + @@ -106,16 +117,19 @@ - - - - - - + + + + + + + + + \ No newline at end of file diff --git a/Eventuous.slnx b/Eventuous.slnx index e32a93a74..c1b586d3f 100644 --- a/Eventuous.slnx +++ b/Eventuous.slnx @@ -14,9 +14,11 @@ + + @@ -162,6 +164,12 @@ + + + + + + diff --git a/nuget.config b/nuget.config new file mode 100644 index 000000000..993d13d07 --- /dev/null +++ b/nuget.config @@ -0,0 +1,12 @@ + + + + + + + + + + + + \ No newline at end of file diff --git a/samples/azure/Bookings.AppHost/AppHost.cs b/samples/azure/Bookings.AppHost/AppHost.cs new file mode 100644 index 000000000..2d1e7b140 --- /dev/null +++ b/samples/azure/Bookings.AppHost/AppHost.cs @@ -0,0 +1,41 @@ +using Scalar.Aspire; + +var builder = DistributedApplication.CreateBuilder(args); + +var sql = builder.AddAzureSqlServer("sql").RunAsContainer(); +var bookingsDb = sql.AddDatabase("bookings-db"); +var paymentsDb = sql.AddDatabase("payments-db"); + +var serviceBus = builder.AddAzureServiceBus("sbemulators").RunAsEmulator(); +var queue = serviceBus.AddServiceBusQueue("PaymentsIntegration"); + +var blobs = builder.AddAzureStorage("storage").RunAsEmulator().AddBlobs("blobs"); +var containers = blobs.AddBlobContainer("bookings-container"); + +var bookings = builder.AddProject("bookings") + .WithHttpEndpoint() + .WithHttpHealthCheck("/health") + .WithReference(bookingsDb) + .WithReference(serviceBus) + .WithReference(blobs) + .WaitFor(bookingsDb) + .WaitFor(serviceBus) + .WaitFor(blobs); + +var payments = builder.AddProject("payments") + .WithHttpEndpoint() + .WithHttpHealthCheck("/health") + .WithReference(paymentsDb) + .WithReference(serviceBus) + .WithReference(blobs) + .WaitFor(paymentsDb) + .WaitFor(serviceBus) + .WaitFor(blobs); + +var scalar = builder.AddScalarApiReference() + .WithApiReference(bookings) + .WithApiReference(payments) + .WaitFor(bookings) + .WaitFor(payments); + +builder.Build().Run(); diff --git a/samples/azure/Bookings.AppHost/Bookings.AppHost.csproj b/samples/azure/Bookings.AppHost/Bookings.AppHost.csproj new file mode 100644 index 000000000..81da14bb0 --- /dev/null +++ b/samples/azure/Bookings.AppHost/Bookings.AppHost.csproj @@ -0,0 +1,25 @@ + + + + Exe + net10.0 + enable + enable + 301020ec-674b-4bcd-ba8f-2eddd4c017df + + + + + + + + + + + + + + + + + diff --git a/samples/azure/Bookings.AppHost/appsettings.Development.json b/samples/azure/Bookings.AppHost/appsettings.Development.json new file mode 100644 index 000000000..ff66ba6b2 --- /dev/null +++ b/samples/azure/Bookings.AppHost/appsettings.Development.json @@ -0,0 +1,8 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft.AspNetCore": "Warning" + } + } +} diff --git a/samples/azure/Bookings.AppHost/appsettings.json b/samples/azure/Bookings.AppHost/appsettings.json new file mode 100644 index 000000000..2185f9551 --- /dev/null +++ b/samples/azure/Bookings.AppHost/appsettings.json @@ -0,0 +1,9 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft.AspNetCore": "Warning", + "Aspire.Hosting.Dcp": "Warning" + } + } +} diff --git a/samples/azure/Bookings.Domain/Bookings.Domain.csproj b/samples/azure/Bookings.Domain/Bookings.Domain.csproj new file mode 100644 index 000000000..bcba3d519 --- /dev/null +++ b/samples/azure/Bookings.Domain/Bookings.Domain.csproj @@ -0,0 +1,16 @@ + + + Debug;Release + + + + + + + + + + + + + diff --git a/samples/azure/Bookings.Payments/Bookings.Payments.csproj b/samples/azure/Bookings.Payments/Bookings.Payments.csproj new file mode 100644 index 000000000..b931a56fb --- /dev/null +++ b/samples/azure/Bookings.Payments/Bookings.Payments.csproj @@ -0,0 +1,47 @@ + + + Debug;Release + + + + + + + + + + + + + + + + + + + + Infrastructure\Logging.cs + + + Infrastructure\Telemetry.cs + + + Domain\%(Filename)%(Extension) + + + Application\%(Filename)%(Extension) + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/samples/azure/Bookings.Payments/Integration/Payments.cs b/samples/azure/Bookings.Payments/Integration/Payments.cs new file mode 100644 index 000000000..3c989a1d5 --- /dev/null +++ b/samples/azure/Bookings.Payments/Integration/Payments.cs @@ -0,0 +1,31 @@ +using Bookings.Payments.Domain; +using Eventuous; +using Eventuous.Azure.ServiceBus.Producers; +using Eventuous.Gateway; +using Eventuous.Subscriptions.Context; +using static Bookings.Payments.Integration.IntegrationEvents; + +namespace Bookings.Payments.Integration; + +public static class PaymentsGateway { + static readonly StreamName Stream = new("PaymentsIntegration"); + static readonly ServiceBusProduceOptions ProduceOptions = new(); + + public static ValueTask[]> Transform(IMessageConsumeContext original) { + var result = original.Message is PaymentEvents.PaymentRecorded evt + ? new GatewayMessage( + Stream, + new BookingPaymentRecorded(original.Stream.GetId(), evt.BookingId, evt.Amount, evt.Currency), + new(), + ProduceOptions + ) + : null; + + return ValueTask.FromResult[]>(result != null ? [result] : []); + } +} + +public static class IntegrationEvents { + [EventType("BookingPaymentRecorded")] + public record BookingPaymentRecorded(string PaymentId, string BookingId, float Amount, string Currency); +} diff --git a/samples/azure/Bookings.Payments/Program.cs b/samples/azure/Bookings.Payments/Program.cs new file mode 100644 index 000000000..15db438a6 --- /dev/null +++ b/samples/azure/Bookings.Payments/Program.cs @@ -0,0 +1,43 @@ +using Bookings.Infrastructure; +using Bookings.Payments; +using Bookings.Payments.Domain; +using Eventuous; +using Microsoft.OpenApi.Models; +using NodaTime; +using NodaTime.Serialization.SystemTextJson; +using Serilog; +using static Bookings.Payments.Application.PaymentCommands; + +TypeMap.RegisterKnownEventTypes(); +Logging.ConfigureLog(); + +var builder = WebApplication.CreateBuilder(args); +builder.Host.UseSerilog(); + +builder.Services.AddEndpointsApiExplorer(); +builder.Services + .AddControllers() + .AddJsonOptions(cfg => cfg.JsonSerializerOptions.ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)); +builder.Services.ConfigureHttpJsonOptions(cfg => cfg.SerializerOptions.ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)); +builder.Services.AddSwaggerGen(c => { + c.SwaggerDoc("v1", new() { Title = "Bookings API", Version = "v1" }); + c.AddServer(new OpenApiServer { Url = "/" }); // Relative path +}); +// OpenTelemetry instrumentation must be added before adding Eventuous services +builder.Services.AddTelemetry(); +builder.Services.AddEventuous(builder.Configuration); + +builder.Services.AddHealthChecks(); + +var app = builder.Build(); + +app.Services.AddEventuousLogs(); +app.UseSwagger(c=>c.RouteTemplate = "openapi/{documentName}.json"); +app.UseOpenTelemetryPrometheusScrapingEndpoint(); + +app.MapHealthChecks("/health"); + +// Here we discover commands by their annotations +app.MapCommands().MapCommand(); + +app.Run(); \ No newline at end of file diff --git a/samples/azure/Bookings.Payments/Registrations.cs b/samples/azure/Bookings.Payments/Registrations.cs new file mode 100644 index 000000000..400400a16 --- /dev/null +++ b/samples/azure/Bookings.Payments/Registrations.cs @@ -0,0 +1,37 @@ +using Bookings.Payments.Application; +using Bookings.Payments.Domain; +using Bookings.Payments.Integration; +using Eventuous.Azure.ServiceBus.Producers; +using Eventuous.SqlServer; +using Eventuous.SqlServer.Subscriptions; +using Microsoft.Extensions.Azure; + +namespace Bookings.Payments; + +public static class Registrations { + public static void AddEventuous(this IServiceCollection services, IConfiguration configuration) { + services.AddAzureClients(builder => { + var sbConnectionString = configuration.GetConnectionString("sbemulators") ?? throw new InvalidOperationException("Connection string 'sbemulators' not found."); + builder.AddServiceBusClient(sbConnectionString); + var blobConnectionString = configuration.GetConnectionString("blobs") ?? throw new InvalidOperationException("Connection string 'blobs' not found."); + builder.AddBlobServiceClient(blobConnectionString); + }); + + var connectionString = configuration.GetConnectionString("payments-db") ?? throw new InvalidOperationException("Connection string 'payments-db' not found."); + + services.AddEventuousSqlServer(connectionString, initializeDatabase: true); + services.AddEventStore(); + services.AddSqlServerCheckpointStore(); + services.AddCommandService(); + services.AddProducer(); + services.AddSingleton(new ServiceBusProducerOptions { + QueueOrTopicName = "PaymentsIntegration", + }); + + services + .AddGateway( + "IntegrationSubscription", + PaymentsGateway.Transform + ); + } +} diff --git a/samples/azure/Bookings.Payments/appsettings.json b/samples/azure/Bookings.Payments/appsettings.json new file mode 100644 index 000000000..379540232 --- /dev/null +++ b/samples/azure/Bookings.Payments/appsettings.json @@ -0,0 +1,14 @@ +{ + "ConnectionStrings": { + "database": "from aspire", + "sbemulators": "from aspire", + "blobs": "from aspire" + }, + "Logging": { + "LogLevel": { + "Default": "Debug", + "Microsoft.AspNetCore": "Warning" + } + }, + "AllowedHosts": "*" +} diff --git a/samples/azure/Bookings/Application/BookingsCommandService.cs b/samples/azure/Bookings/Application/BookingsCommandService.cs new file mode 100644 index 000000000..c47e857db --- /dev/null +++ b/samples/azure/Bookings/Application/BookingsCommandService.cs @@ -0,0 +1,32 @@ +using Bookings.Domain; +using Bookings.Domain.Bookings; +using Eventuous; +using NodaTime; +using static Bookings.Application.BookingCommands; +// ReSharper disable ArrangeObjectCreationWhenTypeNotEvident + +namespace Bookings.Application; + +public class BookingsCommandService : CommandService { + public BookingsCommandService(IEventStore store, Services.IsRoomAvailable isRoomAvailable) : base(store) { + On() + .InState(ExpectedState.New) + .GetId(cmd => new BookingId(cmd.BookingId)) + .ActAsync( + (booking, cmd, _) => booking.BookRoom( + cmd.GuestId, + new RoomId(cmd.RoomId), + new StayPeriod(LocalDate.FromDateTime(cmd.CheckInDate), LocalDate.FromDateTime(cmd.CheckOutDate)), + new Money(cmd.BookingPrice, cmd.Currency), + new Money(cmd.PrepaidAmount, cmd.Currency), + DateTimeOffset.Now, + isRoomAvailable + ) + ); + + On() + .InState(ExpectedState.Existing) + .GetId(cmd => new BookingId(cmd.BookingId)) + .Act((booking, cmd) => booking.RecordPayment(new Money(cmd.PaidAmount, cmd.Currency), cmd.PaymentId, cmd.PaidBy, DateTimeOffset.Now)); + } +} diff --git a/samples/azure/Bookings/Application/BookingsQueryService.cs b/samples/azure/Bookings/Application/BookingsQueryService.cs new file mode 100644 index 000000000..af4352259 --- /dev/null +++ b/samples/azure/Bookings/Application/BookingsQueryService.cs @@ -0,0 +1,7 @@ +using Bookings.Application.Queries; + +namespace Bookings.Application; + +public class BookingsQueryService([FromKeyedServices("BookingsProjections")] MyBookingsProjection projection) { + public async Task GetUserBookings(string userId) => await projection.LoadDocument(userId); +} diff --git a/samples/azure/Bookings/Application/Commands.cs b/samples/azure/Bookings/Application/Commands.cs new file mode 100644 index 000000000..c210f09e6 --- /dev/null +++ b/samples/azure/Bookings/Application/Commands.cs @@ -0,0 +1,17 @@ +namespace Bookings.Application; + +public static class BookingCommands { + public record BookRoom( + string BookingId, + string GuestId, + string RoomId, + DateTime CheckInDate, + DateTime CheckOutDate, + float BookingPrice, + float PrepaidAmount, + string Currency, + DateTimeOffset BookingDate + ); + + public record RecordPayment(string BookingId, float PaidAmount, string Currency, string PaymentId, string PaidBy); +} \ No newline at end of file diff --git a/samples/azure/Bookings/Application/Queries/BookingDocument.cs b/samples/azure/Bookings/Application/Queries/BookingDocument.cs new file mode 100644 index 000000000..620ca8ef9 --- /dev/null +++ b/samples/azure/Bookings/Application/Queries/BookingDocument.cs @@ -0,0 +1,16 @@ +using NodaTime; + +// ReSharper disable UnusedAutoPropertyAccessor.Global + +namespace Bookings.Application.Queries; + +public record BookingDocument { + public string? GuestId { get; init; } + public string? RoomId { get; init; } + public LocalDate CheckInDate { get; init; } + public LocalDate CheckOutDate { get; init; } + public float BookingPrice { get; init; } + public float PaidAmount { get; init; } + public float Outstanding { get; init; } + public bool Paid { get; init; } +} diff --git a/samples/azure/Bookings/Application/Queries/BookingStateProjection.cs b/samples/azure/Bookings/Application/Queries/BookingStateProjection.cs new file mode 100644 index 000000000..a424c7185 --- /dev/null +++ b/samples/azure/Bookings/Application/Queries/BookingStateProjection.cs @@ -0,0 +1,32 @@ +using Azure.Storage.Blobs; +using Eventuous.Azure.Storage.Blobs; +using Microsoft.AspNetCore.Http.Json; +using Microsoft.Extensions.Options; +using static Bookings.Domain.Bookings.BookingEvents; + +// ReSharper disable UnusedAutoPropertyAccessor.Global + +namespace Bookings.Application.Queries; + +public class BookingStateProjection : BlobStorageProjector { + public BookingStateProjection( + BlobServiceClient client, + IOptions serializerOptions + ) : base(client, "bookings-container", serializerOptions.Value.SerializerOptions) { + On(HandleRoomBooked); + + On((b, evt) => b with { Outstanding = evt.Outstanding }); + + On((b, evt) => b with { Paid = true }); + } + + static BookingDocument HandleRoomBooked(BookingDocument bookingDocument, V1.RoomBooked evt) => + bookingDocument with { + GuestId = evt.GuestId, + RoomId = evt.RoomId, + CheckInDate = evt.CheckInDate, + CheckOutDate = evt.CheckOutDate, + BookingPrice = evt.BookingPrice, + Outstanding = evt.OutstandingAmount + }; +} diff --git a/samples/azure/Bookings/Application/Queries/MyBookings.cs b/samples/azure/Bookings/Application/Queries/MyBookings.cs new file mode 100644 index 000000000..25109c996 --- /dev/null +++ b/samples/azure/Bookings/Application/Queries/MyBookings.cs @@ -0,0 +1,10 @@ +using System.Collections.Immutable; +using NodaTime; + +namespace Bookings.Application.Queries; + +public record MyBookings { + public ImmutableList Bookings { get; init; } = []; + + public record Booking(string BookingId, LocalDate CheckInDate, LocalDate CheckOutDate, float Price); +} \ No newline at end of file diff --git a/samples/azure/Bookings/Application/Queries/MyBookingsProjection.cs b/samples/azure/Bookings/Application/Queries/MyBookingsProjection.cs new file mode 100644 index 000000000..fa9fc4dc6 --- /dev/null +++ b/samples/azure/Bookings/Application/Queries/MyBookingsProjection.cs @@ -0,0 +1,38 @@ +using Azure.Storage.Blobs; +using Bookings.Domain.Bookings; +using Eventuous; +using Eventuous.Azure.Storage.Blobs; +using Eventuous.Subscriptions.Context; +using Microsoft.AspNetCore.Http.Json; +using Microsoft.Extensions.Options; +using static Bookings.Domain.Bookings.BookingEvents; + +namespace Bookings.Application.Queries; + +public class MyBookingsProjection : BlobStorageProjector { + readonly IEventReader eventReader; + + public MyBookingsProjection( + BlobServiceClient client, + IEventReader eventReader, + IOptions serializerOptions + ) : base(client, "bookings-container", serializerOptions.Value.SerializerOptions) { + this.eventReader = eventReader; + + On(AddBooking, ctx => new ValueTask(ctx.Message.GuestId)); + On(CancelBooking, GetGuestIdFromStateAsync); + } + + private async ValueTask GetGuestIdFromStateAsync(IMessageConsumeContext ctx) { + var folded = await eventReader.LoadState(ctx.Stream, true, ctx.CancellationToken); + return folded.State.GuestId ?? throw new InvalidOperationException("MyBookings not found"); + } + + private static MyBookings AddBooking(IMessageConsumeContext ctx, MyBookings b) => b with { + Bookings = b.Bookings.Add(new(ctx.Stream.GetId(), ctx.Message.CheckInDate, ctx.Message.CheckOutDate, ctx.Message.BookingPrice)) + }; + + private static MyBookings CancelBooking(IMessageConsumeContext ctx, MyBookings b) => b with { + Bookings = b.Bookings.RemoveAll(booking => booking.BookingId == ctx.Stream.GetId()) + }; +} diff --git a/samples/azure/Bookings/Bookings.csproj b/samples/azure/Bookings/Bookings.csproj new file mode 100644 index 000000000..b95928d7d --- /dev/null +++ b/samples/azure/Bookings/Bookings.csproj @@ -0,0 +1,36 @@ + + + Linux + Debug;Release + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/samples/azure/Bookings/HttpApi/Bookings/CommandApi.cs b/samples/azure/Bookings/HttpApi/Bookings/CommandApi.cs new file mode 100644 index 000000000..12aebf573 --- /dev/null +++ b/samples/azure/Bookings/HttpApi/Bookings/CommandApi.cs @@ -0,0 +1,28 @@ +using Bookings.Domain.Bookings; +using Eventuous; +using Eventuous.Extensions.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using static Bookings.Application.BookingCommands; + +namespace Bookings.HttpApi.Bookings; + +[Route("/booking")] +public class CommandApi(ICommandService service) : CommandHttpApiBase(service) { + [HttpPost] + [Route("book")] + public Task.Ok>> BookRoom([FromBody] BookRoom cmd, CancellationToken cancellationToken) + => Handle(cmd, cancellationToken); + + /// + /// This endpoint is for demo purposes only. The normal flow to register booking payments is to submit + /// a command via the Booking.Payments HTTP API. It then gets propagated to the Booking aggregate + /// via the integration messaging flow. + /// + /// Command to register the payment + /// Cancellation token + /// + [HttpPost] + [Route("recordPayment")] + public Task.Ok>> RecordPayment([FromBody] RecordPayment cmd, CancellationToken cancellationToken) + => Handle(cmd, cancellationToken); +} diff --git a/samples/azure/Bookings/HttpApi/Bookings/QueryApi.cs b/samples/azure/Bookings/HttpApi/Bookings/QueryApi.cs new file mode 100644 index 000000000..e2c70d3c8 --- /dev/null +++ b/samples/azure/Bookings/HttpApi/Bookings/QueryApi.cs @@ -0,0 +1,18 @@ +using Bookings.Domain.Bookings; +using Eventuous; +using Microsoft.AspNetCore.Mvc; + +namespace Bookings.HttpApi.Bookings; + +[Route("/bookings")] +public class QueryApi(IEventReader store) : ControllerBase { + readonly StreamNameMap _streamNameMap = new(); + + [HttpGet] + [Route("{id}")] + public async Task GetBooking(string id, CancellationToken cancellationToken) { + var booking = await store.LoadState(_streamNameMap, new(id), cancellationToken: cancellationToken); + + return booking.State; + } +} diff --git a/samples/azure/Bookings/Infrastructure/Logging.cs b/samples/azure/Bookings/Infrastructure/Logging.cs new file mode 100644 index 000000000..52e088bf7 --- /dev/null +++ b/samples/azure/Bookings/Infrastructure/Logging.cs @@ -0,0 +1,23 @@ +using Serilog; +using Serilog.Events; + +namespace Bookings.Infrastructure; + +public static class Logging { + public static void ConfigureLog() + => Log.Logger = new LoggerConfiguration() + .MinimumLevel.Verbose() + .MinimumLevel.Override("Microsoft", LogEventLevel.Information) + .MinimumLevel.Override("Microsoft.AspNetCore.Hosting.Diagnostics", LogEventLevel.Warning) + .MinimumLevel.Override("Microsoft.AspNetCore.Routing", LogEventLevel.Warning) + .MinimumLevel.Override("Grpc", LogEventLevel.Information) + .MinimumLevel.Override("EventStore", LogEventLevel.Information) + .MinimumLevel.Override("Npgsql", LogEventLevel.Warning) + .Enrich.FromLogContext() + .WriteTo.Console( + outputTemplate: + "[{Timestamp:HH:mm:ss} {Level:u3}] {Message:lj} {NewLine}{Exception}" + ) + .WriteTo.OpenTelemetry() + .CreateLogger(); +} diff --git a/samples/azure/Bookings/Infrastructure/Telemetry.cs b/samples/azure/Bookings/Infrastructure/Telemetry.cs new file mode 100644 index 000000000..2ad65fd52 --- /dev/null +++ b/samples/azure/Bookings/Infrastructure/Telemetry.cs @@ -0,0 +1,33 @@ +using Eventuous.Diagnostics.OpenTelemetry; +using OpenTelemetry.Logs; +using OpenTelemetry.Metrics; +using OpenTelemetry.Resources; +using OpenTelemetry.Trace; + +namespace Bookings.Infrastructure; + +public static class Telemetry { + public static void AddTelemetry(this IServiceCollection services) { + if (Environment.GetEnvironmentVariable("OTEL_EXPORTER_OTLP_ENDPOINT") == null) + return; + + services.AddOpenTelemetry() + .ConfigureResource(builder => builder.AddService("bookings")) + .WithMetrics( + builder => builder + .AddAspNetCoreInstrumentation() + .AddSqlClientInstrumentation() + .AddHttpClientInstrumentation() + .AddEventuous() + .AddEventuousSubscriptions() + .AddPrometheusExporter() + .AddOtlpExporter()) + .WithTracing( + builder => builder + .AddAspNetCoreInstrumentation() + .AddSqlClientInstrumentation() + .AddHttpClientInstrumentation() + .AddEventuousTracing() + .AddOtlpExporter()); + } +} \ No newline at end of file diff --git a/samples/azure/Bookings/Integration/Payments.cs b/samples/azure/Bookings/Integration/Payments.cs new file mode 100644 index 000000000..953662b94 --- /dev/null +++ b/samples/azure/Bookings/Integration/Payments.cs @@ -0,0 +1,35 @@ +using Bookings.Domain.Bookings; +using Eventuous; +using static Bookings.Application.BookingCommands; +using static Bookings.Integration.IntegrationEvents; +using EventHandler = Eventuous.Subscriptions.EventHandler; + +namespace Bookings.Integration; + +public class PaymentsIntegrationHandler : EventHandler { + public const string Stream = "PaymentsIntegration"; + + readonly ICommandService _applicationService; + + public PaymentsIntegrationHandler(ICommandService applicationService) { + _applicationService = applicationService; + On(async ctx => await HandlePayment(ctx.Message, ctx.CancellationToken)); + } + + Task HandlePayment(BookingPaymentRecorded evt, CancellationToken cancellationToken) + => _applicationService.Handle( + new RecordPayment( + evt.BookingId, + evt.Amount, + evt.Currency, + evt.PaymentId, + "" + ), + cancellationToken + ); +} + +static class IntegrationEvents { + [EventType("BookingPaymentRecorded")] + public record BookingPaymentRecorded(string PaymentId, string BookingId, float Amount, string Currency); +} \ No newline at end of file diff --git a/samples/azure/Bookings/Program.cs b/samples/azure/Bookings/Program.cs new file mode 100644 index 000000000..917d2b149 --- /dev/null +++ b/samples/azure/Bookings/Program.cs @@ -0,0 +1,55 @@ +using Bookings; +using Bookings.Application; +using Bookings.Domain.Bookings; +using Bookings.Infrastructure; +using Eventuous; +using Eventuous.Spyglass; +using Microsoft.OpenApi.Models; +using NodaTime; +using NodaTime.Serialization.SystemTextJson; +using Serilog; +using static Bookings.Integration.IntegrationEvents; + +TypeMap.RegisterKnownEventTypes(typeof(BookingEvents.V1.RoomBooked).Assembly); +TypeMap.RegisterKnownEventTypes(typeof(BookingPaymentRecorded).Assembly); +Logging.ConfigureLog(); + +var builder = WebApplication.CreateBuilder(args); +builder.Logging.SetMinimumLevel(LogLevel.Trace).AddConsole(); +builder.Host.UseSerilog(); + +builder.Services + .AddControllers() + .AddJsonOptions(cfg => cfg.JsonSerializerOptions.ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)); +builder.Services.ConfigureHttpJsonOptions(cfg => cfg.SerializerOptions.ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)); +builder.Services.AddEndpointsApiExplorer(); +builder.Services.AddSwaggerGen(c => { + c.SwaggerDoc("v1", new() { Title = "Bookings API", Version = "v1" }); + c.AddServer(new OpenApiServer { Url = "/" }); // Relative path +}); +builder.Services.AddTelemetry(); +builder.Services.AddEventuous(builder.Configuration); + +builder.Services.AddHealthChecks(); + +var app = builder.Build(); + +app.UseSerilogRequestLogging(); +app.UseEventuousLogs(); +app.UseSwagger(c=>c.RouteTemplate = "openapi/{documentName}.json"); +app.MapControllers(); +app.UseOpenTelemetryPrometheusScrapingEndpoint(); +app.MapEventuousSpyglass(); + +app.MapHealthChecks("/health"); + +app.MapGet( + "/bookings/my/{userId}", + async (string userId, BookingsQueryService queryService) => { + var userBookings = await queryService.GetUserBookings(userId); + + return userBookings == null ? Results.NotFound() : Results.Ok(userBookings); + } +).WithTags("QueryApi"); + +app.Run(); \ No newline at end of file diff --git a/samples/azure/Bookings/Registrations.cs b/samples/azure/Bookings/Registrations.cs new file mode 100644 index 000000000..58dbdfea8 --- /dev/null +++ b/samples/azure/Bookings/Registrations.cs @@ -0,0 +1,59 @@ +using System.Text.Json; +using Bookings.Application; +using Bookings.Application.Queries; +using Bookings.Domain; +using Bookings.Domain.Bookings; +using Bookings.Integration; +using Eventuous; +using Eventuous.Azure.ServiceBus.Subscriptions; +using Eventuous.SqlServer; +using Eventuous.SqlServer.Subscriptions; +using Microsoft.Extensions.Azure; +using NodaTime; +using NodaTime.Serialization.SystemTextJson; + +namespace Bookings; + +public static class Registrations { + public static void AddEventuous(this IServiceCollection services, IConfiguration configuration) { + DefaultEventSerializer.SetDefaultSerializer( + new DefaultEventSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web).ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)) + ); + + services.AddAzureClients(builder => { + var sbConnectionString = configuration.GetConnectionString("sbemulators") ?? throw new InvalidOperationException("Connection string 'sbemulators' not found."); + builder.AddServiceBusClient(sbConnectionString); + var blobConnectionString = configuration.GetConnectionString("blobs") ?? throw new InvalidOperationException("Connection string 'blobs' not found."); + builder.AddBlobServiceClient(blobConnectionString); + }); + + var connectionString = configuration.GetConnectionString("bookings-db") ?? throw new InvalidOperationException("Connection string 'bookings-db' not found."); + + services.AddEventuousSqlServer(connectionString, initializeDatabase: true); + services.AddEventStore(); + services.AddSqlServerCheckpointStore(); + services.AddCommandService(); + + services.AddSingleton((_, _) => new(true)); + + services.AddSingleton( + (from, currency) => new(from.Amount * 2, currency) + ); + + services.AddSubscription( + "BookingsProjections", + builder => builder + .AddEventHandler() + .AddEventHandler() + ); + + services.AddSubscription( + "PaymentIntegration", + builder => builder + .Configure(x => x.QueueOrTopic = new Queue(PaymentsIntegrationHandler.Stream)) + .AddEventHandler() + ); + + services.AddSingleton(); + } +} diff --git a/samples/azure/Bookings/appsettings.json b/samples/azure/Bookings/appsettings.json new file mode 100644 index 000000000..16fd6acb0 --- /dev/null +++ b/samples/azure/Bookings/appsettings.json @@ -0,0 +1,8 @@ +{ + "ConnectionStrings": { + "database": "from aspire", + "sbemulators": "from aspire", + "blobs": "from aspire" + }, + "AllowedHosts": "*" +} diff --git a/samples/azure/Directory.Build.props b/samples/azure/Directory.Build.props new file mode 100644 index 000000000..d6f2db864 --- /dev/null +++ b/samples/azure/Directory.Build.props @@ -0,0 +1,7 @@ + + + + net10.0 + net10.0 + + diff --git a/samples/azure/README.md b/samples/azure/README.md new file mode 100644 index 000000000..572446845 --- /dev/null +++ b/samples/azure/README.md @@ -0,0 +1,95 @@ +# Bookings sample application - Azure + +This project demonstrates how to use Eventuous in an Azure environment, showcasing: + * Event-sourced domain model using `Aggregate` and `AggregateState` + * Aggregate persistence using Azure SQL Server + * Read models stored in Azure Blob Storage + * Integration between services using Azure Service Bus messaging + * Orchestration and local development using .NET Aspire + +## .NET Aspire + +[.NET Aspire](https://learn.microsoft.com/en-us/dotnet/aspire/) is a cloud-ready stack for building observable, production-ready distributed applications. It provides: + + * **Service discovery and orchestration** - Manages dependencies between services and infrastructure + * **Unified configuration** - Centralized configuration for all services and resources + * **Dashboard** - A visual representation of the application's architecture and health + * **Local development** - Run dependencies like Azure SQL Server and Azure Service Bus emulators as containers + +To use .NET Aspire, follow the instruction [here](https://aspire.dev/get-started/install-cli/) or install the workload using the .NET CLI: + +```bash +dotnet workload install aspire +``` + +For a better development experience, install the official extensions: + * [Visual Studio: .NET Aspire tools](https://marketplace.visualstudio.com/items?itemName=ms-dotnettools.dotnet-aspire) + * [Visual Studio Code: .NET Aspire](https://marketplace.visualstudio.com/items?itemName=ms-dotnettools.vscode-dotnet-aspire) + +This sample uses Aspire to orchestrate the Bookings and Bookings.Payments services along with their Azure dependencies (SQL Server, Service Bus, and Blob Storage). + +## Usage + +Start the application using Aspire from the `Bookings.AppHost` directory: + +```bash +cd samples/azure/ +aspire run +``` + +Aspire will: +1. Start the Azure SQL Server container (with separate databases for Bookings and Payments) +2. Start the Azure Service Bus emulator +3. Start the Azure Storage emulator (for Blob Storage) +4. Launch the Bookings and Bookings.Payments services +5. Provide a dashboard (typically at `http://localhost:18945`) to view the application architecture and health + +Once all services are running, use the Scalar API reference in the Aspire dashboard to explore and test the API endpoints. + +**Note:** Aspire can also run your application in debug mode. + +### Example commands + +#### Bookings -> BookRoom (`/bookings/book`) + +- This command raises an event, which gets stored in Azure SQL Server. +- A real-time subscription triggers a projection, which adds or updates documents in Azure Blob Storage: + - one for the booking + - one for the guest + +#### Bookings.Payments -> RecordPayment (`/recordPayment`) + +- When this command is executed, it raises a `PaymentRecorded` event, which gets persisted to Azure SQL Server. +- A gateway inside the Payments service subscribes to this event and publishes an integration event to Azure Service Bus. +- An integration Service Bus subscription receives the integration event and calls the Bookings service to execute the `RecordPayment` command, so it acts as a Reactor. +- When that command gets executed, it raises a `PaymentRecorded` event, which gets persisted to Azure SQL Server. It might also raise a `BookingFullyPaid` or `BookingOverpaid` events, depending on the amount. +- Those new events are projected to Azure Blob Storage documents in the `bookings-container` using the read-model subscription. + +```mermaid +graph TB + HTTP --> RecordPayment + subgraph Payments + direction LR + RecordPayment -- aggregate --> PaymentRecorded[PaymentRecorded
domain event] + Reactor --> PR[PaymentRecorded
integration event] + end + subgraph SQLServer + PaymentRecorded -- eventstore --> SQL[(Azure SQL Server)] + SQL -- gateway --> Reactor + end + subgraph Broker + PR --> ServiceBus[[Azure Service Bus]] + end + subgraph Bookings + direction RL + ServiceBus -- subscription --> IH[Integration
handler] + IH --> RP[RecordPayment] + RP -- aggregate --> PR1[PaymentRecorded] + PR1 -- eventstore --> SQL + SQL -- subscription --> Projections + end + subgraph BlobStorage + Projections --> BC[(BookingState)] + Projections --> MB[(MyBookings)] + end +``` \ No newline at end of file diff --git a/samples/azure/aspire.config.json b/samples/azure/aspire.config.json new file mode 100644 index 000000000..4b4f2dfbc --- /dev/null +++ b/samples/azure/aspire.config.json @@ -0,0 +1,5 @@ +{ + "appHost": { + "path": "Bookings.AppHost/Bookings.AppHost.csproj" + } +} \ No newline at end of file diff --git a/src/Azure/src/Eventuous.Azure.ServiceBus/Subscriptions/ServiceBusSubscription.cs b/src/Azure/src/Eventuous.Azure.ServiceBus/Subscriptions/ServiceBusSubscription.cs index daa30259a..4b6b4e447 100644 --- a/src/Azure/src/Eventuous.Azure.ServiceBus/Subscriptions/ServiceBusSubscription.cs +++ b/src/Azure/src/Eventuous.Azure.ServiceBus/Subscriptions/ServiceBusSubscription.cs @@ -23,7 +23,7 @@ public class ServiceBusSubscription : EventSubscriptionConsume pipe instance /// Logger factory (optional) /// Event serializer (optional) - 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; 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..548ae6ce6 --- /dev/null +++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjector.cs @@ -0,0 +1,243 @@ +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; + +/// +/// 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. +/// +/// +public class BlobStorageProjector : BaseEventHandler where T : class, new() { + /// Azure Blob Storage container client. + protected readonly BlobContainerClient ContainerClient; + + readonly JsonSerializerOptions _jsonOptions; + readonly Dictionary>> _handlers = []; + readonly ITypeMapper _map; + + private readonly int _raceRetries; + private 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 JSON serializer options. + /// Optional type mapper for event type resolution. + 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; + } + + /// + /// Initializes projector with service client and container name. + /// + /// Azure Blob Storage service client. + /// Name of the container to use. + /// Optional JSON serializer options. + /// Optional type mapper for event type resolution. + /// Optional projector configuration. + public BlobStorageProjector( + BlobServiceClient serviceClient, + string containerName, + JsonSerializerOptions? serializerOptions = null, + BlobStorageProjectorOptions? projectorOptions = null, + ITypeMapper? mapper = null + ) : this(serviceClient.GetBlobContainerClient(containerName), serializerOptions, 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(); + } + } + + private BlobClient GetBlobContainerClient(string blobName) => ContainerClient.GetBlobClient(blobName); + + /// Registers event handler without custom blob ID. + /// Event type to handle. + /// Async state update function receiving context, current state, and event. + protected void On(Func, T, ValueTask> handler) where TEvent : class + => On(handler, default); + + /// 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; + + public async Task LoadDocument(string id) { + try { + var blobName = GetBlobName(id); + var blobClient = ContainerClient.GetBlobClient(blobName); + BlobDownloadResult blobContent = await blobClient.DownloadContentAsync().NoContext(); + return ToObjectFromJson(blobContent.Content); + } catch (RequestFailedException ex) when (ex.Status == 404) { + return null; + } + } + + private T ToObjectFromJson(BinaryData content) => content.ToObjectFromJson(_jsonOptions) ?? new T(); + private 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"; + + private class Handler + where TEvent : class { + private readonly BlobStorageProjector projector; + private readonly Func, T, ValueTask> EventHandler; + private readonly GetBlobId? GetBlobId; + + public Handler(BlobStorageProjector blobStorageProjector, Func, T, ValueTask> handler, GetBlobId? getBlobId) { + projector = blobStorageProjector; + EventHandler = handler; + GetBlobId = getBlobId; + } + + public async ValueTask Handle(IMessageConsumeContext context) { + var typedContext = context as MessageConsumeContext ?? new MessageConsumeContext(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 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 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 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 { + ["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(); + } + } + } +} \ No newline at end of file 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..5f9d13f39 --- /dev/null +++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/BlobStorageProjectorOptions.cs @@ -0,0 +1,53 @@ +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. + /// By default, the default JSON serializer options will be used if this property is not set. + /// + 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; } = 0; + + /// + /// 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 global position. + /// Uses the GlobalPosition metadata stored with the blob for comparison. + /// Effective for append-only event streams where global position uniquely identifies a message. + /// + 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 +} \ No newline at end of file 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..d2cd88dd2 --- /dev/null +++ b/src/Azure/src/Eventuous.Azure.Storage.Blobs/README.md @@ -0,0 +1,112 @@ +# 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 + +By using `IOptions` we can also use the Json serialization options as set in ASP DI. + +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, IOptions serializerOptions) + : base(client, "bookings-container", serializerOptions.Value) { + + // 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. Always processes messages and updates blobs. +- **`ByGlobalPosition`** - Skips processing if existing blob has matching global position metadata. +- **`ByMessageId`** - Skips processing if existing blob has matching message ID metadata. + +### 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}.json") +); +``` + +Use per-event blob ID overrides when you need different events to target different blob paths or naming conventions 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 +- 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..2d9ba3fae --- /dev/null +++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/BlobStorageProjectorTests.cs @@ -0,0 +1,682 @@ +using System.Text.Json; +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); + + // ========== 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 + } + + // ========== 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: async () => { + // Simulate concurrent modification: modify the blob directly with a different value + 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); + }, + 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 = "stream/ConcurrentState.json"; + + var projector = new ConcurrentModificationProjector( + fixture.BlobServiceClient, + containerName, + messWithState: async () => { + // Simulate concurrent modification: modify the blob directly with a different value + 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); + }, + 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 BlobStorageProjectorTests.AssertFailure(result2); + } + + [Test] + public async Task ConcurrentModificationOfExistingBlob_ShouldReturnFailure() { + // Arrange + var containerName = await SetupContainer("concurrent-existing"); + var blobName = "stream/ConcurrentState.json"; + + await SetupExistingBlob(containerName, blobName, new ConcurrentState { Value = 1 }); + + var projector = new ConcurrentModificationProjector( + fixture.BlobServiceClient, + containerName, + messWithState: async() => { + // Simulate concurrent modification: modify the blob directly with a different value + 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); + }, + 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 BlobStorageProjectorTests.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] + public async Task Idempotency_ByGlobalPosition_ShouldIgnoreDuplicatePosition() { + // 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: 100u); + + // 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) => + new MessageConsumeContext( + eventId: messageId ?? Guid.NewGuid().ToString(), + eventType: message.GetType().Name, + contentType: "application/json", + stream: DefaultStream, + 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 { + private int _callCount = 0; + private readonly Func? _messWithState; + private 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 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..4ab5a79fb --- /dev/null +++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Eventuous.Tests.Azure.Storage.Blobs.csproj @@ -0,0 +1,19 @@ + + + + + + + + + + + + + + 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..2ccf549ad --- /dev/null +++ b/src/Azure/test/Eventuous.Tests.Azure.Storage.Blobs/Fixtures/IntegrationFixture.cs @@ -0,0 +1,49 @@ +using Azure.Storage.Blobs; +using Eventuous.TestHelpers; +using Testcontainers.Azurite; +using TUnit.Core.Interfaces; + +namespace Eventuous.Tests.Azure.Storage.Blobs.Fixtures; + +public sealed class IntegrationFixture : IAsyncInitializer, IAsyncDisposable { + public IEventStore EventStore { get; set; } = null!; + public BlobServiceClient BlobServiceClient { get; private set; } = null!; + + static IEventSerializer Serializer { get; } = new DefaultEventSerializer(TestPrimitives.DefaultOptions); + + AzuriteContainer _azuriteContainer = null!; + + public async Task AppendEvent( + StreamName streamName, + object evt, + ExpectedStreamVersion? version = null + ) { + return await EventStore.AppendEvents( + streamName, + version ?? ExpectedStreamVersion.Any, + [new(Guid.NewGuid(), evt, new())], + CancellationToken.None + ); + } + + static IntegrationFixture() { + DefaultEventSerializer.SetDefaultSerializer(Serializer); + } + + 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(); + + var connectionString = _azuriteContainer.GetConnectionString(); + BlobServiceClient = new BlobServiceClient(connectionString); + + } + + 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 }; +}