Azure blob storage projection

Directory.Packages.props

@@ -22,10 +22,15 @@
     <TUnitVersion>0.77.3</TUnitVersion>
   </PropertyGroup>
   <ItemGroup>
+    <PackageVersion Include="Aspire.Hosting.Azure.ServiceBus" Version="13.4.5" />
+    <PackageVersion Include="Aspire.Hosting.Azure.Sql" Version="13.4.5" />
+    <PackageVersion Include="Aspire.Hosting.Azure.Storage" Version="13.4.5" />
+    <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 +42,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 +68,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 +119,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/">
@@ -162,6 +164,12 @@
         <Project Path="samples/postgres/Bookings.Payments/Bookings.Payments.csproj" />
         <Project Path="samples/postgres/Bookings/Bookings.csproj" />
     </Folder>
+    <Folder Name="/Samples/Azure/">
+        <Project Path="samples/azure/Bookings.AppHost/Bookings.AppHost.csproj" />
+        <Project Path="samples/azure/Bookings.Domain/Bookings.Domain.csproj" />
+        <Project Path="samples/azure/Bookings.Payments/Bookings.Payments.csproj" />
+        <Project Path="samples/azure/Bookings/Bookings.csproj" />
+    </Folder>
     <Folder Name="/SignalR/" />
     <Folder Name="/SignalR/src/">
         <Project Path="src/SignalR/src/Eventuous.SignalR.Server/Eventuous.SignalR.Server.csproj" />

samples/azure/Bookings.AppHost/AppHost.cs

@@ -0,0 +1,39 @@
+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<Projects.Bookings>("bookings")
+    .WithExternalHttpEndpoints()
+    .WithReference(bookingsDb)
+    .WithReference(serviceBus)
+    .WithReference(blobs)
+    .WaitFor(bookingsDb)
+    .WaitFor(serviceBus)
+    .WaitFor(blobs);
+
+var payments = builder.AddProject<Projects.Bookings_Payments>("payments")
+    .WithExternalHttpEndpoints()
+    .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();

samples/azure/Bookings.AppHost/Bookings.AppHost.csproj

@@ -0,0 +1,25 @@
+<Project Sdk="Aspire.AppHost.Sdk/13.4.5">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <UserSecretsId>301020ec-674b-4bcd-ba8f-2eddd4c017df</UserSecretsId>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Bookings\Bookings.csproj" />
+    <ProjectReference Include="..\Bookings.Payments\Bookings.Payments.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Aspire.Hosting.Azure.ServiceBus" />
+    <PackageReference Include="Aspire.Hosting.Azure.Sql" />
+    <PackageReference Include="Aspire.Hosting.Azure.Storage" />
+    <PackageReference Include="Scalar.Aspire" />
+  </ItemGroup>
+</Project>

samples/azure/Bookings.AppHost/appsettings.Development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  }
+}

samples/azure/Bookings.AppHost/appsettings.json

@@ -0,0 +1,9 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning",
+      "Aspire.Hosting.Dcp": "Warning"
+    }
+  }
+}

samples/azure/Bookings.Domain/Bookings.Domain.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk">
+    <PropertyGroup>
+      <Configurations>Debug;Release</Configurations>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="NodaTime"/>
+    </ItemGroup>
+    <ItemGroup>
+      <ProjectReference Include="$(CoreRoot)\Eventuous.Domain\Eventuous.Domain.csproj" />
+      <ProjectReference Include="$(CoreRoot)\Eventuous.Shared\Eventuous.Shared.csproj" />
+    </ItemGroup>
+    <ItemGroup>
+        <Compile Include="..\..\kurrentdb\Bookings.Domain\*.cs" />
+        <Compile Include="..\..\kurrentdb\Bookings.Domain\Bookings\*.cs" LinkBase="Bookings" />
+    </ItemGroup>
+</Project>

samples/azure/Bookings.Payments/Bookings.Payments.csproj

@@ -0,0 +1,44 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+    <PropertyGroup>
+        <Configurations>Debug;Release</Configurations>
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="Azure.Storage.Blobs" />
+        <PackageReference Include="Microsoft.Extensions.Azure" />
+        <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" />
+        <PackageReference Include="OpenTelemetry.Exporter.Prometheus.AspNetCore" />
+        <PackageReference Include="OpenTelemetry.Exporter.Zipkin" />
+        <PackageReference Include="OpenTelemetry.Extensions.Hosting" />
+        <PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" />
+        <PackageReference Include="OpenTelemetry.Instrumentation.SqlClient" />
+        <PackageReference Include="Serilog.Extensions.Hosting" />
+        <PackageReference Include="Serilog.Sinks.Console" />
+        <PackageReference Include="Swashbuckle.AspNetCore" />
+    </ItemGroup>
+    <ItemGroup>
+        <Compile Include="..\Bookings\Infrastructure\Logging.cs">
+            <Link>Infrastructure\Logging.cs</Link>
+        </Compile>
+        <Compile Include="..\Bookings\Infrastructure\Telemetry.cs">
+            <Link>Infrastructure\Telemetry.cs</Link>
+        </Compile>
+        <Compile Include="..\..\kurrentdb\Bookings.Payments\Domain\*.cs">
+            <Link>Domain\%(Filename)%(Extension)</Link>
+        </Compile>
+        <Compile Include="..\..\kurrentdb\Bookings.Payments\Application\*.cs">
+            <Link>Application\%(Filename)%(Extension)</Link>
+        </Compile>
+    </ItemGroup>
+    <ItemGroup>
+        <ProjectReference Include="$(SrcRoot)\Extensions\src\Eventuous.Extensions.AspNetCore\Eventuous.Extensions.AspNetCore.csproj"/>
+        <ProjectReference Include="$(SrcRoot)\Extensions\src\Eventuous.Extensions.DependencyInjection\Eventuous.Extensions.DependencyInjection.csproj"/>
+        <ProjectReference Include="$(SrcRoot)\Diagnostics\src\Eventuous.Diagnostics.OpenTelemetry\Eventuous.Diagnostics.OpenTelemetry.csproj"/>
+        <ProjectReference Include="$(SrcRoot)\Gateway\src\Eventuous.Gateway\Eventuous.Gateway.csproj"/>
+        <ProjectReference Include="$(SrcRoot)\Core\gen\Eventuous.Subscriptions.Generators\Eventuous.Subscriptions.Generators.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
+        <ProjectReference Include="$(SrcRoot)\Core\gen\Eventuous.Shared.Generators\Eventuous.Shared.Generators.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
+        <ProjectReference Include="$(SrcRoot)\Experimental\src\Eventuous.Spyglass\Eventuous.Spyglass.csproj"/>
+        <ProjectReference Include="$(SrcRoot)\Experimental\gen\Eventuous.Spyglass.Generators\Eventuous.Spyglass.Generators.csproj" OutputItemType="Analyzer" ReferenceOutputAssembly="false" />
+        <ProjectReference Include="$(SrcRoot)\SqlServer\src\Eventuous.SqlServer\Eventuous.SqlServer.csproj" />
+        <ProjectReference Include="$(SrcRoot)\Azure\src\Eventuous.Azure.ServiceBus\Eventuous.Azure.ServiceBus.csproj" />
+    </ItemGroup>
+</Project>

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<GatewayMessage<ServiceBusProduceOptions>[]> Transform(IMessageConsumeContext original) {
+        var result = original.Message is PaymentEvents.PaymentRecorded evt
+            ? new GatewayMessage<ServiceBusProduceOptions>(
+                Stream,
+                new BookingPaymentRecorded(original.Stream.GetId(), evt.BookingId, evt.Amount, evt.Currency),
+                new(),
+                ProduceOptions
+            )
+            : null;
+
+        return ValueTask.FromResult<GatewayMessage<ServiceBusProduceOptions>[]>(result != null ? [result] : []);
+    }
+}
+
+public static class IntegrationEvents {
+    [EventType("BookingPaymentRecorded")]
+    public record BookingPaymentRecorded(string PaymentId, string BookingId, float Amount, string Currency);
+}

samples/azure/Bookings.Payments/Program.cs

@@ -0,0 +1,28 @@
+using Bookings.Infrastructure;
+using Bookings.Payments;
+using Bookings.Payments.Domain;
+using Eventuous;
+using Serilog;
+
+TypeMap.RegisterKnownEventTypes();
+Logging.ConfigureLog();
+
+var builder = WebApplication.CreateBuilder(args);
+builder.Host.UseSerilog();
+
+builder.Services.AddEndpointsApiExplorer();
+builder.Services.AddSwaggerGen(c => c.SwaggerDoc("v1", new() { Title = "Bookings Payments API", Version = "v1" }));
+// OpenTelemetry instrumentation must be added before adding Eventuous services
+builder.Services.AddTelemetry();
+builder.Services.AddEventuous(builder.Configuration);
+
+var app = builder.Build();
+
+app.Services.AddEventuousLogs();
+app.UseSwagger(c=>c.RouteTemplate = "openapi/{documentName}.json");
+app.UseOpenTelemetryPrometheusScrapingEndpoint();
+
+// Here we discover commands by their annotations
+app.MapDiscoveredCommands<PaymentState>();
+
+app.Run();

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(async 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<SqlServerStore>();
+        services.AddSqlServerCheckpointStore();
+        services.AddCommandService<CommandService, PaymentState>();
+        services.AddProducer<ServiceBusProducer>();
+        services.AddSingleton(new ServiceBusProducerOptions {
+            QueueOrTopicName = "PaymentsIntegration",
+        });
+
+        services
+            .AddGateway<SqlServerAllStreamSubscription, SqlServerAllStreamSubscriptionOptions, ServiceBusProducer, ServiceBusProduceOptions>(
+                "IntegrationSubscription",
+                PaymentsGateway.Transform
+            );
+    }
+}

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": "*"
+}

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<Booking, BookingState, BookingId> {
+    public BookingsCommandService(IEventStore store, Services.IsRoomAvailable isRoomAvailable) : base(store) {
+        On<BookRoom>()
+            .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<RecordPayment>()
+            .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));
+    }
+}

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<MyBookings?> GetUserBookings(string userId) => await projection.LoadDocument(userId);
+}

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);
+}