实现事务性发件箱模式

src/HelloShop.OrderingService/Behaviors/TransactionBehavior.cs

@@ -3,13 +3,14 @@
 
 using HelloShop.OrderingService.Extensions;
 using HelloShop.OrderingService.Infrastructure;
+using HelloShop.OrderingService.Services;
 using MediatR;
 using Microsoft.EntityFrameworkCore;
 using System.Data;
 
 namespace HelloShop.OrderingService.Behaviors
 {
-    public class TransactionBehavior<TRequest, TResponse>(OrderingServiceDbContext dbContext, ILoggerFactory loggerFactory) : IPipelineBehavior<TRequest, TResponse> where TRequest : IRequest<TResponse>
+    public class TransactionBehavior<TRequest, TResponse>(OrderingServiceDbContext dbContext, ILoggerFactory loggerFactory, IDistributedEventService eventService) : IPipelineBehavior<TRequest, TResponse> where TRequest : IRequest<TResponse>
     {
         private readonly ILogger<TransactionBehavior<TRequest, TResponse>> _logger = loggerFactory.CreateLogger<TransactionBehavior<TRequest, TResponse>>();
 
@@ -42,6 +43,8 @@ namespace HelloShop.OrderingService.Behaviors
                     }
 
                     await transaction.CommitAsync();
+
+                    await eventService.PublishEventsThroughEventBusAsync(transaction.TransactionId, cancellationToken);
                 });
 
                 return response ?? throw new ApplicationException($"Command {typeName} returned null response");

src/HelloShop.OrderingService/Commands/Orders/CreateOrderCommandHandler.cs

@@ -14,10 +14,14 @@ using Microsoft.EntityFrameworkCore;
 
 namespace HelloShop.OrderingService.Commands.Orders
 {
-    public class CreateOrderCommandHandler(IMediator mediator, OrderingServiceDbContext dbContext, IMapper mapper, IDistributedEventBus distributedEventBus) : IRequestHandler<CreateOrderCommand, bool>
+    public class CreateOrderCommandHandler(IMediator mediator, IDistributedEventService distributedEventService, OrderingServiceDbContext dbContext, IMapper mapper) : IRequestHandler<CreateOrderCommand, bool>
     {
         public async Task<bool> Handle(CreateOrderCommand request, CancellationToken cancellationToken)
         {
+            var orderStartedIntegrationEvent = new OrderStartedDistributedEvent(request.UserId);
+
+            await distributedEventService.AddAndSaveEventAsync(orderStartedIntegrationEvent, cancellationToken);
+
             Address address = mapper.Map<Address>(request);
 
             IEnumerable<OrderItem> orderItems = mapper.Map<IEnumerable<OrderItem>>(request.OrderItems);
@@ -47,8 +51,6 @@ namespace HelloShop.OrderingService.Commands.Orders
 
             await mediator.Publish(new OrderStartedLocalEvent(order), cancellationToken);
 
-            await distributedEventBus.PublishAsync(new OrderStartedDistributedEvent(request.UserId), cancellationToken);
-
             return await Task.FromResult(true);
         }
     }

src/HelloShop.OrderingService/DistributedEvents/EventHandling/OrderPaymentSucceededDistributedEventHandler.cs

@@ -4,12 +4,13 @@
 using HelloShop.OrderingService.DistributedEvents.Events;
 using HelloShop.OrderingService.Entities.Orders;
 using HelloShop.OrderingService.Infrastructure;
+using HelloShop.OrderingService.Services;
 using HelloShop.ServiceDefaults.DistributedEvents.Abstractions;
 using Microsoft.EntityFrameworkCore;
 
 namespace HelloShop.OrderingService.DistributedEvents.EventHandling
 {
-    public class OrderPaymentSucceededDistributedEventHandler(OrderingServiceDbContext dbContext, IDistributedEventBus distributedEventBus) : IDistributedEventHandler<OrderPaymentSucceededDistributedEvent>
+    public class OrderPaymentSucceededDistributedEventHandler(OrderingServiceDbContext dbContext, IDistributedEventService distributedEventService) : IDistributedEventHandler<OrderPaymentSucceededDistributedEvent>
     {
         public async Task HandleAsync(OrderPaymentSucceededDistributedEvent @event)
         {
@@ -29,13 +30,15 @@ namespace HelloShop.OrderingService.DistributedEvents.EventHandling
 
                 await dbContext.SaveChangesAsync();
 
-                await transaction.CommitAsync();
-
                 var orderStockList = order.OrderItems.Select(orderItem => new OrderStockItem(orderItem.ProductId, orderItem.Units));
 
                 var integrationEvent = new OrderPaidDistributedEvent(order.Id, orderStockList);
 
-                await distributedEventBus.PublishAsync(integrationEvent);
+                await distributedEventService.AddAndSaveEventAsync(integrationEvent);
+
+                await distributedEventService.PublishEventsThroughEventBusAsync(transaction.TransactionId);
+
+                await transaction.CommitAsync();
             });
         }
     }

src/HelloShop.OrderingService/Entities/EventLogs/DistributedEventLog.cs

@@ -0,0 +1,42 @@
+// Copyright (c) HelloShop Corporation. All rights reserved.
+// See the license file in the project root for more information.
+
+using HelloShop.ServiceDefaults.DistributedEvents.Abstractions;
+using System.Diagnostics.CodeAnalysis;
+
+namespace HelloShop.OrderingService.Entities.EventLogs
+{
+    public enum DistributedEventStatus { NotPublished, InProgress, Published, PublishedFailed }
+
+    public class DistributedEventLog
+    {
+        public Guid EventId { get; set; }
+
+        public required string EventTypeName { get; set; }
+
+        public required DistributedEvent DistributedEvent { get; set; }
+
+        public DistributedEventStatus Status { get; set; } = DistributedEventStatus.NotPublished;
+
+        public int TimesSent { get; set; }
+
+        public DateTimeOffset CreationTime { get; set; } = DateTimeOffset.UtcNow;
+
+        public required Guid TransactionId { get; set; }
+
+        /// <summary>
+        /// EF Core cannot set navigation properties using a constructor.
+        /// The constructor can be public, private, or have any other accessibility. 
+        /// </summary>
+        private DistributedEventLog() { }
+
+        [SetsRequiredMembers]
+        public DistributedEventLog(DistributedEvent @event, Guid transactionId)
+        {
+            EventId = @event.Id;
+            EventTypeName = @event.GetType().Name;
+            DistributedEvent = @event;
+            TransactionId = transactionId;
+        }
+    }
+}

src/HelloShop.OrderingService/Extensions/Extensions.cs

@@ -48,6 +48,8 @@ namespace HelloShop.OrderingService.Extensions
 
             builder.Services.AddHostedService<GracePeriodWorker>();
             builder.Services.AddHostedService<PaymentWorker>();
+
+            builder.Services.AddTransient<IDistributedEventService, DistributedEventService<OrderingServiceDbContext>>();
         }
 
         public static WebApplication MapApplicationEndpoints(this WebApplication app)

src/HelloShop.OrderingService/HelloShop.OrderingService.csproj

@@ -8,12 +8,12 @@
     <ProjectReference Include="..\HelloShop.ServiceDefaults\HelloShop.ServiceDefaults.csproj" />
   </ItemGroup>
   <ItemGroup>
-    <PackageReference Include="MediatR" Version="12.4.0" />
-    <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="8.0.8" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="8.0.8">
+    <PackageReference Include="MediatR" Version="12.4.1" />
+    <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="8.0.10" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="8.0.10">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="8.0.4" />
+    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="8.0.8" />
   </ItemGroup>
 </Project>

src/HelloShop.OrderingService/Infrastructure/EntityConfigurations/EventLogs/DistributedEventLogEntityTypeConfiguration.cs

@@ -0,0 +1,27 @@
+// Copyright (c) HelloShop Corporation. All rights reserved.
+// See the license file in the project root for more information.
+
+using HelloShop.OrderingService.Entities.EventLogs;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+using Microsoft.EntityFrameworkCore;
+using System.Text.Json;
+using HelloShop.ServiceDefaults.DistributedEvents.Abstractions;
+
+namespace HelloShop.OrderingService.Infrastructure.EntityConfigurations.EventLogs
+{
+    public class DistributedEventLogEntityTypeConfiguration : IEntityTypeConfiguration<DistributedEventLog>
+    {
+        private static readonly JsonSerializerOptions s_jsonSerializerOptions = new(JsonSerializerDefaults.Web);
+
+        public void Configure(EntityTypeBuilder<DistributedEventLog> builder)
+        {
+            builder.ToTable("DistributedEventLogs");
+
+            builder.HasKey(x => x.EventId);
+            builder.Property(x => x.EventTypeName).HasMaxLength(32);
+            builder.Property(x => x.Status).HasConversion<string>();
+
+            builder.Property(x => x.DistributedEvent).HasConversion(v => JsonSerializer.Serialize(v, v.GetType(), s_jsonSerializerOptions), v => JsonSerializer.Deserialize<DistributedEvent>(v, s_jsonSerializerOptions)!);
+        }
+    }
+}

src/HelloShop.OrderingService/Services/DistributedEventService.cs

@@ -0,0 +1,89 @@
+// Copyright (c) HelloShop Corporation. All rights reserved.
+// See the license file in the project root for more information.
+
+using HelloShop.OrderingService.Entities.EventLogs;
+using HelloShop.ServiceDefaults.DistributedEvents.Abstractions;
+using Microsoft.EntityFrameworkCore;
+
+namespace HelloShop.OrderingService.Services
+{
+    public class DistributedEventService<TContext>(TContext dbContext, IDistributedEventBus distributedEventBus, ILogger<DistributedEventService<TContext>> logger) : IDistributedEventService, IDisposable where TContext : DbContext
+    {
+        private volatile bool _disposedValue;
+
+        public async Task UpdateEventStatusAsync(Guid eventId, DistributedEventStatus status, CancellationToken cancellationToken = default)
+        {
+            var eventLogEntry = dbContext.Set<DistributedEventLog>().Single(ie => ie.EventId == eventId);
+
+            eventLogEntry.Status = status;
+
+            if (status == DistributedEventStatus.InProgress)
+            {
+                eventLogEntry.TimesSent++;
+            }
+
+            await dbContext.SaveChangesAsync(cancellationToken);
+        }
+
+        public async Task<IEnumerable<DistributedEventLog>> RetrieveEventLogsPendingToPublishAsync(Guid transactionId, CancellationToken cancellationToken = default)
+        {
+            var result = await dbContext.Set<DistributedEventLog>().Where(e => e.TransactionId == transactionId && e.Status == DistributedEventStatus.NotPublished).ToListAsync(cancellationToken: cancellationToken);
+
+            return result.Count != 0 ? result.OrderBy(o => o.CreationTime) : [];
+        }
+
+        public async Task AddAndSaveEventAsync(DistributedEvent @event, CancellationToken cancellationToken = default)
+        {
+            var transaction = dbContext.Database.CurrentTransaction ?? throw new InvalidOperationException("This method must be called within a transaction scope.");
+
+            var eventLog = new DistributedEventLog(@event, transaction.TransactionId);
+
+            await dbContext.AddAsync(eventLog, cancellationToken);
+
+            await dbContext.SaveChangesAsync(cancellationToken);
+        }
+
+        public async Task PublishEventsThroughEventBusAsync(Guid transactionId, CancellationToken cancellationToken = default)
+        {
+            var pendingEventLogs = await RetrieveEventLogsPendingToPublishAsync(transactionId, cancellationToken);
+
+            foreach (var eventLog in pendingEventLogs)
+            {
+                logger.LogInformation("Publishing integration event {EventId} {DistributedEvent}", eventLog.EventId, eventLog.DistributedEvent);
+
+                try
+                {
+                    await UpdateEventStatusAsync(eventLog.EventId, DistributedEventStatus.InProgress, cancellationToken);
+                    await distributedEventBus.PublishAsync(eventLog.DistributedEvent, cancellationToken);
+                    await UpdateEventStatusAsync(eventLog.EventId, DistributedEventStatus.Published, cancellationToken);
+                }
+                catch (Exception ex)
+                {
+                    logger.LogError(ex, "Error publishing distributed event {EventId}", eventLog.EventId);
+
+                    await UpdateEventStatusAsync(eventLog.EventId, DistributedEventStatus.PublishedFailed, cancellationToken);
+                }
+            }
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!_disposedValue)
+            {
+                if (disposing)
+                {
+                    dbContext.Dispose();
+                }
+
+                _disposedValue = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
+    }
+
+}

src/HelloShop.OrderingService/Services/IDistributedEventService.cs

@@ -0,0 +1,19 @@
+// Copyright (c) HelloShop Corporation. All rights reserved.
+// See the license file in the project root for more information.
+
+using HelloShop.OrderingService.Entities.EventLogs;
+using HelloShop.ServiceDefaults.DistributedEvents.Abstractions;
+
+namespace HelloShop.OrderingService.Services
+{
+    public interface IDistributedEventService
+    {
+        Task<IEnumerable<DistributedEventLog>> RetrieveEventLogsPendingToPublishAsync(Guid transactionId, CancellationToken cancellationToken = default);
+
+        Task AddAndSaveEventAsync(DistributedEvent @event, CancellationToken cancellationToken = default);
+
+        Task UpdateEventStatusAsync(Guid eventId, DistributedEventStatus status, CancellationToken cancellationToken = default);
+
+        Task PublishEventsThroughEventBusAsync(Guid transactionId, CancellationToken cancellationToken = default);
+    }
+}