From 586a6b6834217bbc91829321e268268a8589f772 Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Sat, 10 Oct 2026 16:41:28 +0800 Subject: [PATCH] change --- Directory.Build.props | 2 +- .../Messaging/OmsIntegrationEventPublisher.cs | 68 +++++++++ .../Messaging/OmsOutboxUnitOfWork.cs | 22 +++ .../Messaging/OrderBatchJobEventHandler.cs | 3 +- .../Messaging/OrderBatchJobPublisher.cs | 9 +- .../OrderSideEffectEventPublisher.cs | 65 +-------- .../OrderParcelCreateEventPublisher.cs | 79 ++--------- src/Oms.Application/Oms.Application.csproj | 1 + .../Rbac/UserProfileRepository.cs | 2 + .../Operations/InboundRecordService.cs | 7 +- .../Operations/OutboundRecordService.cs | 7 +- .../Issues/AutoTriggerRuntimeService.cs | 15 +- .../Orders/OrderService.ParcelBilling.cs | 2 +- .../Services/Orders/OrderService.cs | 6 + .../RecordingPhysicalDistributedEventBus.cs | 23 +++ .../Messaging/OmsOutboxTestModule.cs | 51 +++++++ .../OrderBatchJobOutboxPublisherTests.cs | 90 ++++++++++++ .../OrderParcelCreateOutboxPublisherTests.cs | 131 ++++++++++++++++++ .../OrderSideEffectOutboxPublisherTests.cs | 62 +++++++++ src/Oms.Tests/Oms.Tests.csproj | 5 + src/SimpleAdmin.Api/SimpleAdmin.Api.csproj | 3 +- src/SimpleAdmin.Api/SimpleAdminApiModule.cs | 22 ++- .../Storage/IntegrationEventLogSchema.cs | 48 +++++++ .../appsettings.Development.json | 8 ++ 24 files changed, 584 insertions(+), 147 deletions(-) create mode 100644 src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs create mode 100644 src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs create mode 100644 src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs create mode 100644 src/Oms.Tests/Messaging/OmsOutboxTestModule.cs create mode 100644 src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs create mode 100644 src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs create mode 100644 src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs create mode 100644 src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs diff --git a/Directory.Build.props b/Directory.Build.props index 9c939ac..7f7b084 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,6 +1,6 @@ - 8.0.5.18 + 8.0.5.19 diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs new file mode 100644 index 0000000..0f12f49 --- /dev/null +++ b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs @@ -0,0 +1,68 @@ +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.UnitOfWork; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace Oms.Application.Messaging; + +/// +/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。 +/// +internal static class OmsIntegrationEventPublisher +{ + public static async Task PublishAsync( + IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger, + IntegrationEvent @event, + CancellationToken cancellationToken = default) + { + if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) + { + await eventBus.PublishAsync(@event, cancellationToken); + logger.LogDebug( + "Outbox enqueued {EventName} Id={EventId} (ambient TX)", + @event.GetType().Name, + @event.Id); + return; + } + + // 无环境事务时开启短事务只写 Outbox，并告警：业务落库与事件非原子。 + logger.LogWarning( + "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX", + @event.GetType().Name, + @event.Id); + + using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional()); + // 先打开 Outbox DbContext 以产生可征用的 DbTransaction + var dbProvider = uow.ServiceProvider.GetRequiredService>(); + await dbProvider.GetDbContextAsync(); + await eventBus.PublishAsync(@event, cancellationToken); + await uow.CompleteAsync(cancellationToken); + } + + public static async Task TryPublishAsync( + IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger, + IntegrationEvent @event, + CancellationToken cancellationToken = default) + { + try + { + await PublishAsync(eventBus, unitOfWorkManager, logger, @event, cancellationToken); + return true; + } + catch (Exception ex) + { + logger.LogError( + ex, + "Outbox publish failed {EventName} Id={EventId}", + @event.GetType().Name, + @event.Id); + return false; + } + } +} diff --git a/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs b/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs new file mode 100644 index 0000000..dd1c83b --- /dev/null +++ b/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs @@ -0,0 +1,22 @@ +using System.Data; +using Fake.UnitOfWork; + +namespace Oms.Application.Messaging; + +/// +/// 集成事件 Outbox 所需的事务性工作单元约定。 +/// +public static class OmsOutboxUnitOfWork +{ + /// + /// 创单/后置事件发布使用的事务 UoW 属性（MySQL ReadCommitted）。 + /// + public static UnitOfWorkAttribute Transactional() => new() + { + IsTransactional = true, + IsolationLevel = IsolationLevel.ReadCommitted + }; + + public static bool IsAmbientTransactional(IUnitOfWorkManager unitOfWorkManager) => + unitOfWorkManager.Current?.Context.IsTransactional == true; +} diff --git a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs index 580e23c..95b3ad4 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs @@ -29,7 +29,8 @@ public sealed class OrderBatchJobEventHandler( return; } - using var unitOfWork = unitOfWorkManager.Begin(); + // ForecastBilling 成功后继续写轨迹/面单/自动触发 Outbox，须事务性 UoW + using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional()); await processor.ProcessAsync(@event.ToMessage(), cancellationToken); await unitOfWork.CompleteAsync(cancellationToken); } diff --git a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs index c066098..a007ac4 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs @@ -1,14 +1,16 @@ using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Oms.Application.Messaging; /// -/// 通过 Fake 分布式事件总线发布订单批量任务。 +/// 通过 Fake 分布式事件总线（Outbox）发布订单批量任务。 /// public sealed class OrderBatchJobPublisher( IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, IOptions options, ILogger logger) : IOrderBatchJobPublisher { @@ -22,10 +24,11 @@ public sealed class OrderBatchJobPublisher( } var @event = OrderBatchJobIntegrationEvent.FromMessage(message); - await eventBus.PublishAsync(@event, cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); logger.LogDebug( - "Published order batch event {EventId} messageId={MessageId} action={Action}", + "Outbox enqueued order batch event {EventId} messageId={MessageId} action={Action}", @event.Id, message.MessageId, message.Action); diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs index 078be98..9322fc4 100644 --- a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs @@ -1,16 +1,16 @@ using Fake.EventBus.Distributed; -using Microsoft.Extensions.DependencyInjection; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging; using Oms.Application.Services.Orders.Issues; namespace Oms.Application.Messaging.OrderSideEffects; /// -/// 优先 Fake.EventBus.RabbitMQ；发布失败则同进程同步执行 Handler。 +/// 订单旁路副作用事件：事务内写 Outbox，Commit 后投递 RabbitMQ。 /// public sealed class OrderSideEffectEventPublisher( IDistributedEventBus eventBus, - IServiceScopeFactory scopeFactory, + IUnitOfWorkManager unitOfWorkManager, ILogger logger) : IOrderSideEffectEventPublisher { public async Task PublishLifecycleTrackAppendAsync( @@ -33,14 +33,8 @@ public sealed class OrderSideEffectEventPublisher( Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim() }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishAutoTriggerFireAsync( @@ -61,52 +55,7 @@ public sealed class OrderSideEffectEventPublisher( OperatorUserId = operatorUserId }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); - } - - private async Task TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken) - { - try - { - await eventBus.PublishAsync(@event, cancellationToken); - logger.LogDebug( - "Published side-effect event {EventName} Id={EventId}", - @event.GetType().Name, - @event.Id); - return true; - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "Publish side-effect event {EventName} failed, fallback to sync. EventId={EventId}", - @event.GetType().Name, - @event.Id); - return false; - } - } - - private async Task ExecuteInScopeAsync( - Func action, - CancellationToken cancellationToken) - where THandler : notnull - { - try - { - using var scope = scopeFactory.CreateScope(); - var handler = scope.ServiceProvider.GetRequiredService(); - await action(handler); - } - catch (Exception ex) - { - logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name); - } + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } } diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs index e265284..cf35e80 100644 --- a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs @@ -1,4 +1,5 @@ using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -11,10 +12,12 @@ using Oms.Domain.OrdersAggregate.Entities; namespace Oms.Application.Messaging.ParcelCreate; /// -/// 创单后置事件发布器：优先走 Fake.EventBus.RabbitMQ，失败则同进程同步执行 Handler 逻辑。 +/// 创单后置事件发布器：事务内写 Outbox（Fake.EventBus.IntegrationEventLog），Commit 后投递 RabbitMQ。 +/// 揽收 Publish 失败时由调用方同步 ExecutePickupAttachAsync 回落。 /// public sealed class OrderParcelCreateEventPublisher( IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, IServiceScopeFactory scopeFactory, ILogger logger) : IOrderParcelCreateEventPublisher { @@ -33,14 +36,8 @@ public sealed class OrderParcelCreateEventPublisher( Authorization = request.Authorization }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishLabelGenerateAsync( @@ -54,14 +51,8 @@ public sealed class OrderParcelCreateEventPublisher( OperatorUserId = operatorUserId }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishCustoms9610UpsertAsync( @@ -75,14 +66,8 @@ public sealed class OrderParcelCreateEventPublisher( PlatformCode = platformCode }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishPickupAttachAsync( @@ -96,8 +81,8 @@ public sealed class OrderParcelCreateEventPublisher( return false; } - // 揽收回落由调用方 ExecutePickupAttachAsync 同步执行，避免重复建单且便于回写单号。 - return await TryPublishAsync(@event, cancellationToken); + return await OmsIntegrationEventPublisher.TryPublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task ExecutePickupAttachAsync( @@ -155,44 +140,4 @@ public sealed class OrderParcelCreateEventPublisher( ExpectPickupTime = input.ExpectPickupTime.Value }; } - - private async Task TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken) - { - try - { - await eventBus.PublishAsync(@event, cancellationToken); - logger.LogDebug( - "Published parcel-create event {EventName} Id={EventId}", - @event.GetType().Name, - @event.Id); - return true; - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "Publish parcel-create event {EventName} failed, fallback to sync. EventId={EventId}", - @event.GetType().Name, - @event.Id); - return false; - } - } - - private async Task ExecuteInScopeAsync( - Func action, - CancellationToken cancellationToken) - where THandler : notnull - { - try - { - using var scope = scopeFactory.CreateScope(); - var handler = scope.ServiceProvider.GetRequiredService(); - await action(handler); - } - catch (Exception ex) - { - // 同步回落失败也不抛回创单主流程 - logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name); - } - } } diff --git a/src/Oms.Application/Oms.Application.csproj b/src/Oms.Application/Oms.Application.csproj index 73214d1..72efa2e 100644 --- a/src/Oms.Application/Oms.Application.csproj +++ b/src/Oms.Application/Oms.Application.csproj @@ -18,6 +18,7 @@ + diff --git a/src/Oms.Application/Rbac/UserProfileRepository.cs b/src/Oms.Application/Rbac/UserProfileRepository.cs index 73f75a3..7115b11 100644 --- a/src/Oms.Application/Rbac/UserProfileRepository.cs +++ b/src/Oms.Application/Rbac/UserProfileRepository.cs @@ -187,6 +187,8 @@ public class UserProfileRepository(FakeRbacDbContext dbContext) : IScopedDepende if (int.TryParse(match.Groups[1].Value, out var number)) { + + max = Math.Max(max, number); } } diff --git a/src/Oms.Application/Services/Operations/InboundRecordService.cs b/src/Oms.Application/Services/Operations/InboundRecordService.cs index 5d6a56d..0770e99 100644 --- a/src/Oms.Application/Services/Operations/InboundRecordService.cs +++ b/src/Oms.Application/Services/Operations/InboundRecordService.cs @@ -1,4 +1,6 @@ +using System.Data; using Fake.Application; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Mvc; using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Shared; @@ -85,6 +87,7 @@ public class InboundRecordService : ApplicationService /// 入参：入库记录创建参数。 /// 入库记录结果。 [HttpPost("/operation/inbound/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateAsync(CreateInboundRecordInput input) { if (string.IsNullOrWhiteSpace(input.WaybillNo)) @@ -141,8 +144,8 @@ public class InboundRecordService : ApplicationService var savedEntity = await InboundRecordRepository.InsertAsync(entity); - // 自动触发改异步，避免拉长扫码 RT - AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Inbound, operatorId); + // 自动触发走 Outbox，避免拉长扫码 RT + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Inbound, operatorId); var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim()); var output = ObjectMapper.Map(savedEntity); diff --git a/src/Oms.Application/Services/Operations/OutboundRecordService.cs b/src/Oms.Application/Services/Operations/OutboundRecordService.cs index b64d4b7..4f856e3 100644 --- a/src/Oms.Application/Services/Operations/OutboundRecordService.cs +++ b/src/Oms.Application/Services/Operations/OutboundRecordService.cs @@ -1,4 +1,6 @@ +using System.Data; using Fake.Application; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Mvc; using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Shared; @@ -99,6 +101,7 @@ public class OutboundRecordService : ApplicationService /// 入参：出库记录创建参数。 /// 出库记录结果。 [HttpPost("/operation/outbound/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateAsync(CreateOutboundRecordInput input) { if (string.IsNullOrWhiteSpace(input.WaybillNo)) @@ -161,8 +164,8 @@ public class OutboundRecordService : ApplicationService var savedEntity = await OutboundRecordRepository.InsertAsync(entity); - // 自动触发改走 Fake.EventBus，避免拉长出库扫码 RT - AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Outbound, operatorId); + // 自动触发走 Outbox，避免拉长出库扫码 RT + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Outbound, operatorId); return ObjectMapper.Map(savedEntity); } diff --git a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs index 9b70856..34d536e 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -22,17 +22,12 @@ public class AutoTriggerRuntimeService( private const string InterceptOperation = "拦截"; /// - /// 后台异步触发（Fake.EventBus.RabbitMQ），避免拉长入库/出库/创单请求。 + /// 投递自动触发事件（须在事务 UoW 内调用，写入 Outbox）。 /// - public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId) - { - if (orderId <= 0) - { - return; - } - - _ = sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId); - } + public Task EnqueueFireAsync(int orderId, AutoTriggerTiming timing, int operatorUserId) => + orderId <= 0 + ? Task.CompletedTask + : sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId); public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId) { diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index fbd528a..198a429 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -2789,7 +2789,7 @@ public partial class OrderService appendInitialTrack: true, operatorUserId); - AutoTriggerRuntimeService.EnqueueFire( + await AutoTriggerRuntimeService.EnqueueFireAsync( order.Id, AutoTriggerTiming.OrderCreate, operatorUserId); diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 5e90860..6bcbac0 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -1,7 +1,9 @@ +using System.Data; using System.Globalization; using Fake.Application; using Fake.Domain.Exceptions; using Fake.ObjectStorage; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Mvc; using MiniExcelLibs; @@ -265,6 +267,7 @@ public partial class OrderService : ApplicationService /// 入参：订单操作参数。 /// 无返回体，仅完成订单预报处理。 [HttpPost("/order/pre-declare")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task PreDeclareAsync(OrderActionInput input) { var order = await GetOrderOrThrowAsync(input.Id); @@ -605,6 +608,7 @@ public partial class OrderService : ApplicationService /// 入参：订单包裹创建参数。 /// 订单包裹创建结果。 [HttpPost("/order/parcel/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateParcelAsync(OrderParcelCreateInput input) { var inbound = string.IsNullOrWhiteSpace(input.InboundMethod) @@ -636,6 +640,7 @@ public partial class OrderService : ApplicationService /// /// 客户 OpenAPI 创单：同一客户相同 clientNo 视为幂等，返回已存在订单，不新建。 /// + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateParcelForCustomerOpenApiAsync(OrderParcelCreateInput input) { NormalizeParcelCreateInput(input); @@ -734,6 +739,7 @@ public partial class OrderService : ApplicationService /// 入参：批量创建小包请求集合，每条记录沿用单票创建结构。 /// 批量处理结果，包含批次号、成功失败数量与逐条执行明细。 [HttpPost("/order/parcel/batch-create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task BatchCreateParcelAsync(OrderParcelBatchCreateInput input) { if (input == null || input.Count == 0) diff --git a/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs b/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs new file mode 100644 index 0000000..b4a9225 --- /dev/null +++ b/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs @@ -0,0 +1,23 @@ +using System.Collections.Concurrent; +using Fake.EventBus; +using Fake.EventBus.Distributed; + +namespace Oms.Tests.Messaging.Fakes; + +public sealed class RecordingPhysicalDistributedEventBus : IPhysicalDistributedEventBus +{ + private readonly ConcurrentQueue _published = new(); + + public IReadOnlyList Published => _published.ToArray(); + + public void Clear() => _published.Clear(); + + public Task PublishAsync(IntegrationEvent @event, CancellationToken cancellationToken = default) + { + _published.Enqueue(@event); + return Task.CompletedTask; + } + + public Task PublishAsync(Event @event, CancellationToken cancellationToken = default) + => PublishAsync((IntegrationEvent)@event, cancellationToken); +} diff --git a/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs b/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs new file mode 100644 index 0000000..8b22af6 --- /dev/null +++ b/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs @@ -0,0 +1,51 @@ +using Fake.Autofac; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Modularity; +using Fake.SyncEx; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +[DependsOn(typeof(FakeAutofacModule))] +[DependsOn(typeof(FakeEntityFrameworkCoreModule))] +[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))] +public class OmsOutboxTestModule : FakeModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.AddSingleton(sp => + sp.GetRequiredService()); + + context.Services.Configure(o => + { + o.StartupDelay = TimeSpan.FromHours(1); + o.ScanInterval = TimeSpan.FromHours(1); + }); + context.Services.Configure(o => + { + o.StartupDelay = TimeSpan.FromHours(1); + o.CleanupInterval = TimeSpan.FromHours(1); + }); + + var connection = new SqliteConnection("Filename=:memory:"); + connection.Open(); + + context.Services.AddDbContext(builder => + { + builder.UseSqlite(connection); + }, ServiceLifetime.Scoped, ServiceLifetime.Singleton); + } + + public override void PreConfigureApplication(ApplicationConfigureContext context) + { + using var scope = context.ServiceProvider.CreateScope(); + var ctx = scope.ServiceProvider.GetRequiredService(); + SyncContext.Run(async () => await ctx.Database.EnsureCreatedAsync()); + } +} diff --git a/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs new file mode 100644 index 0000000..da907e3 --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs @@ -0,0 +1,90 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Oms.Application.Messaging; +using Oms.Domain.OrdersAggregate.Consts; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderBatchJobPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderBatchJobOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderBatchJobPublisher( + ServiceProvider.GetRequiredService(), + _uowManager, + Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }), + NullLogger.Instance); + } + + [Fact] + public async Task ForecastBilling任务消息_事务Commit后投递() + { + _physicalBus.Clear(); + var message = new OrderBatchJobMessage + { + JobId = 1, + JobItemId = 2, + Action = OrderBatchActionCodes.ForecastBilling, + OrderId = 55, + MessageId = "job:1:55:forecast-billing", + OperatorUserId = 9, + PayloadJson = "{\"CouponInstanceId\":1}" + }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishAsync(message); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + var evt = _physicalBus.Published.Should().ContainSingle() + .Which.Should().BeOfType().Subject; + evt.Action.Should().Be(OrderBatchActionCodes.ForecastBilling); + evt.OrderId.Should().Be(55); + evt.MessageId.Should().Be(message.MessageId); + } + + [Fact] + public async Task 队列未启用时_Publish应抛错() + { + var disabled = new OrderBatchJobPublisher( + ServiceProvider.GetRequiredService(), + _uowManager, + Options.Create(new OrderBatchRabbitMqOptions { Enabled = false }), + NullLogger.Instance); + + var act = () => disabled.PublishAsync(new OrderBatchJobMessage + { + Action = OrderBatchActionCodes.ForecastBilling, + OrderId = 1, + MessageId = "x" + }); + + await act.Should().ThrowAsync() + .WithMessage("*disabled*"); + } +} diff --git a/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs new file mode 100644 index 0000000..f4d036e --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs @@ -0,0 +1,131 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Oms.Application.Integrations.SmallBagTrack; +using Oms.Application.Messaging; +using Oms.Application.Messaging.ParcelCreate; +using Oms.Domain.OrdersAggregate.Entities; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly IDistributedEventBus _distributedEventBus; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IOutboxEventLogService _outbox; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderParcelCreateEventPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderParcelCreateOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _distributedEventBus = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _outbox = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderParcelCreateEventPublisher( + _distributedEventBus, + _uowManager, + ServiceProvider.GetRequiredService(), + NullLogger.Instance); + } + + [Fact] + public void IDistributedEventBus_应为Outbox装饰器() + { + _distributedEventBus.Should().BeOfType(); + } + + [Fact] + public async Task 事务Uow内发布9610_应在Complete后投递物理总线() + { + _physicalBus.Clear(); + var order = new Order { Id = 101, ServiceNo = "SN101" }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishCustoms9610UpsertAsync(order, "lazada"); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().ContainSingle(); + var evt = _physicalBus.Published[0].Should().BeOfType().Subject; + evt.OrderId.Should().Be(101); + evt.PlatformCode.Should().Be("lazada"); + + var pending = (await _outbox.RetrieveEventLogsPendingToPublishAsync(50)).ToList(); + pending.Should().NotContain(e => e.EventId == evt.Id); + } + + [Fact] + public async Task 事务Uow内发布轨迹与面单_Commit后应各投递一次() + { + _physicalBus.Clear(); + var order = new Order { Id = 202, ServiceNo = "SN202" }; + var request = new SmallBagTrackUpsertRequest + { + OrderNo = "SN202", + AppendInitialTrackOnSuccess = true + }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishTrackUpsertAsync(order, request, customerId: 9); + await _publisher.PublishLabelGenerateAsync(order, operatorUserId: 7); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().HaveCount(2); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 202 && e.CustomerId == 9 && e.AppendInitialTrackOnSuccess); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 202 && e.OperatorUserId == 7); + } + + [Fact] + public async Task 回滚后不应投递() + { + _physicalBus.Clear(); + var order = new Order { Id = 303, ServiceNo = "SN303" }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishCustoms9610UpsertAsync(order); + await uow.RollbackAsync(); + } + + _physicalBus.Published.Should().BeEmpty(); + } + + [Fact] + public async Task 无环境事务时_短事务仍能写入并投递() + { + _physicalBus.Clear(); + _uowManager.Current.Should().BeNull(); + + var order = new Order { Id = 404, ServiceNo = "SN404" }; + await _publisher.PublishCustoms9610UpsertAsync(order, "shopee"); + + _physicalBus.Published.Should().ContainSingle() + .Which.Should().BeOfType() + .Which.OrderId.Should().Be(404); + } +} diff --git a/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs new file mode 100644 index 0000000..8f3abf8 --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs @@ -0,0 +1,62 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Oms.Application.Messaging; +using Oms.Application.Messaging.OrderSideEffects; +using Oms.Application.Services.Orders.Issues; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly IDistributedEventBus _distributedEventBus; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderSideEffectEventPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderSideEffectOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _distributedEventBus = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderSideEffectEventPublisher( + _distributedEventBus, + _uowManager, + NullLogger.Instance); + } + + [Fact] + public async Task 事务内发布自动触发与生命周期轨迹_Commit后投递() + { + _physicalBus.Clear(); + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishAutoTriggerFireAsync(11, AutoTriggerTiming.OrderCreate, 3); + await _publisher.PublishLifecycleTrackAppendAsync(11, "SN11", "A0301"); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().HaveCount(2); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 11 && e.Timing == AutoTriggerTiming.OrderCreate.ToString() && e.OperatorUserId == 3); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 11 && e.OrderNo == "SN11"); + } +} diff --git a/src/Oms.Tests/Oms.Tests.csproj b/src/Oms.Tests/Oms.Tests.csproj index 19f3343..040ca5b 100644 --- a/src/Oms.Tests/Oms.Tests.csproj +++ b/src/Oms.Tests/Oms.Tests.csproj @@ -13,9 +13,14 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + all diff --git a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj index 0143f14..568c1a4 100644 --- a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj +++ b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj @@ -15,12 +15,13 @@ + - + diff --git a/src/SimpleAdmin.Api/SimpleAdminApiModule.cs b/src/SimpleAdmin.Api/SimpleAdminApiModule.cs index b24c7ff..19f0d1f 100644 --- a/src/SimpleAdmin.Api/SimpleAdminApiModule.cs +++ b/src/SimpleAdmin.Api/SimpleAdminApiModule.cs @@ -7,6 +7,7 @@ using Fake.AspNetCore.Auditing; using Fake.AspNetCore.ExceptionHandling; using Fake.AspNetCore.Newtonsoft; using Fake.AspNetCore.Swagger; +using Fake.EventBus.IntegrationEventLog; using Fake.FeiShu; using Fake.FileManagement.Application; using Fake.FileManagement.Infrastructure; @@ -37,6 +38,8 @@ namespace SimpleAdmin.Api; typeof(FakeFileManagementApplicationModule), typeof(FakeFileManagementInfrastructureModule), typeof(FakeObjectStorageTencentCosModule), + // 在 RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox + typeof(FakeEventBusIntegrationEventLogModule), typeof(FakeAspNetCoreModule), typeof(FakeSerilogModule) )] @@ -80,6 +83,20 @@ public class SimpleAdminApiModule : FakeModule services.AddOmsWeChat(configuration); services.AddSingleton(); + // Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务 + services.AddDbContext(options => + { + var connectionString = configuration.GetConnectionString("IntegrationEventLog") + ?? configuration.GetConnectionString("Default") + ?? configuration["ConnectionStrings:Default"] + ?? string.Empty; + var serverVersion = ServerVersion.AutoDetect(connectionString); + options.UseMySql(connectionString, serverVersion); +#if DEBUG + options.EnableSensitiveDataLogging(); +#endif + }); + //Json格式化全局处理 context.Services.AddFakeNewtonsoft(); @@ -188,11 +205,14 @@ public class SimpleAdminApiModule : FakeModule // 端点映射 app.MapControllers(); - // FileManagement 与 RBAC 同库并行（EF）；应用 Pending migrations 建 stored_file + // FileManagement / Outbox 与业务同库并行（EF） using (var scope = app.Services.CreateScope()) { var fileDb = scope.ServiceProvider.GetRequiredService(); fileDb.Database.Migrate(); + + var outboxDb = scope.ServiceProvider.GetRequiredService(); + IntegrationEventLogSchema.EnsureTablesAsync(outboxDb).GetAwaiter().GetResult(); } app.Services.GetRequiredService().Send("Application started"); diff --git a/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs b/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs new file mode 100644 index 0000000..f44e97c --- /dev/null +++ b/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs @@ -0,0 +1,48 @@ +using Fake.EventBus.IntegrationEventLog; +using Microsoft.EntityFrameworkCore; + +namespace SimpleAdmin.Api.Storage; + +/// +/// Outbox/Inbox 与业务同库时 EnsureCreated 无效（库已存在），用 IF NOT EXISTS 建表。 +/// +internal static class IntegrationEventLogSchema +{ + public static async Task EnsureTablesAsync( + IntegrationEventLogContext db, + CancellationToken cancellationToken = default) + { + await db.Database.ExecuteSqlRawAsync( + """ + CREATE TABLE IF NOT EXISTS `OutboxEventLog` ( + `EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL, + `EventTypeName` varchar(512) NOT NULL, + `State` int NOT NULL, + `TimesSent` int NOT NULL, + `CreationTime` datetime(6) NOT NULL, + `Content` longtext NOT NULL, + `TransactionId` varchar(64) NOT NULL, + `LockExpiresAt` datetime(6) NULL, + PRIMARY KEY (`EventId`), + KEY `IX_OutboxEventLog_State_CreationTime` (`State`, `CreationTime`), + KEY `IX_OutboxEventLog_TransactionId` (`TransactionId`) + ) CHARACTER SET utf8mb4; + """, + cancellationToken); + + await db.Database.ExecuteSqlRawAsync( + """ + CREATE TABLE IF NOT EXISTS `InboxEventLog` ( + `EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL, + `EventTypeName` varchar(512) NOT NULL, + `State` int NOT NULL, + `ProcessedTime` datetime(6) NOT NULL, + `Content` longtext NOT NULL, + `ErrorMessage` longtext NULL, + PRIMARY KEY (`EventId`), + KEY `IX_InboxEventLog_ProcessedTime` (`ProcessedTime`) + ) CHARACTER SET utf8mb4; + """, + cancellationToken); + } +} diff --git a/src/SimpleAdmin.Api/appsettings.Development.json b/src/SimpleAdmin.Api/appsettings.Development.json index 1b28af5..03d2137 100644 --- a/src/SimpleAdmin.Api/appsettings.Development.json +++ b/src/SimpleAdmin.Api/appsettings.Development.json @@ -146,5 +146,13 @@ "RabbitMq": { "Enabled": false } + }, + "IntegrationEventLog": { + "OutboxPublisher": { + "StartupDelay": "00:00:05", + "ScanInterval": "00:00:10", + "BatchSize": 100, + "MaxDegreeOfParallelism": 8 + } } } -- 2.45.2