From dcd985bca418b03fe914c86977369d683b1d089b Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Sat, 10 Oct 2026 10:54:24 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E5=88=9B=E5=8D=95/=E5=87=BA?= =?UTF-8?q?=E5=85=A5=E5=BA=93=E5=90=8E=E7=BD=AE=E5=89=AF=E4=BD=9C=E7=94=A8?= =?UTF-8?q?=E6=8E=A5=E5=85=A5=20Fake.EventBus=EF=BC=8C=E9=A2=84=E6=8A=A5?= =?UTF-8?q?=E8=AE=A1=E8=B4=B9=E6=8E=92=E8=B4=A7=E8=B5=B0=20MQ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将轨迹、面单、9610、揽收、自动触发与 ForecastBilling 异步化，缩短扫码与下单 RT；队列未启用时同步回落。 Co-authored-by: Cursor --- .../Dtos/Orders/OrderParcelCreateOutput.cs | 24 +- .../SmallBagTrackOrderNotifier.cs | 72 ++- .../IOrderSideEffectEventPublisher.cs | 22 + .../OrderAutoTriggerFireEventHandler.cs | 48 ++ .../OrderAutoTriggerFireIntegrationEvent.cs | 19 + .../OrderLifecycleTrackAppendEventHandler.cs | 49 ++ ...derLifecycleTrackAppendIntegrationEvent.cs | 18 + .../OrderSideEffectEventPublisher.cs | 112 ++++ .../IOrderParcelCreateEventPublisher.cs | 46 ++ .../OrderParcelCreateEventPublisher.cs | 198 +++++++ ...rderParcelCustoms9610UpsertEventHandler.cs | 44 ++ ...ParcelCustoms9610UpsertIntegrationEvent.cs | 14 + .../OrderParcelLabelGenerateEventHandler.cs | 38 ++ ...rderParcelLabelGenerateIntegrationEvent.cs | 14 + .../OrderParcelPickupAttachEventHandler.cs | 51 ++ ...OrderParcelPickupAttachIntegrationEvent.cs | 26 + .../OrderParcelTrackUpsertEventHandler.cs | 76 +++ .../OrderParcelTrackUpsertIntegrationEvent.cs | 28 + .../OrderBatchMessagingServiceExtensions.cs | 4 + .../Operations/OutboundRecordService.cs | 14 +- .../Orders/Batch/OrderBatchActionExecutor.cs | 1 + .../Orders/Batch/OrderBatchJobItemFeedback.cs | 1 + .../Batch/OrderBatchOrderActionHandler.cs | 8 + .../Issues/AutoTriggerRuntimeService.cs | 24 +- .../Orders/OrderService.ParcelBilling.cs | 485 +++++++++++++----- .../Services/Orders/OrderService.cs | 28 +- .../Consts/OrderBatchJobStatusCodes.cs | 2 + 27 files changed, 1259 insertions(+), 207 deletions(-) create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireEventHandler.cs create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireIntegrationEvent.cs create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendEventHandler.cs create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendIntegrationEvent.cs create mode 100644 src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertEventHandler.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertIntegrationEvent.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateEventHandler.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateIntegrationEvent.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachEventHandler.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachIntegrationEvent.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertEventHandler.cs create mode 100644 src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertIntegrationEvent.cs diff --git a/src/Oms.Application/Dtos/Orders/OrderParcelCreateOutput.cs b/src/Oms.Application/Dtos/Orders/OrderParcelCreateOutput.cs index 0f7f1fe..66875f0 100644 --- a/src/Oms.Application/Dtos/Orders/OrderParcelCreateOutput.cs +++ b/src/Oms.Application/Dtos/Orders/OrderParcelCreateOutput.cs @@ -41,12 +41,12 @@ public class OrderParcelCreateOutput public string? ConsigneeCountryName { get; set; } /// - /// 同步创建的揽收单号（上门揽收成功时）。 + /// 揽收单号（上门揽收；事件总线消费较快或同步回落时可能已有值）。 /// public string? PickupNo { get; set; } /// - /// 同步创建的揽收单 Id。 + /// 揽收单 Id。 /// public int? PickupOrderId { get; set; } @@ -59,4 +59,24 @@ public class OrderParcelCreateOutput /// 揽收单创建失败原因（运单已成功时仅作提示）。 /// public string? PickupCreateError { get; set; } + + /// + /// 揽收单已入事件总线、后台创建中（此时尚无 PickupNo）。 + /// + public bool PickupQueued { get; set; } + + /// + /// 订单状态（创单后可能仍为草稿，预报计费异步处理中）。 + /// + public string? OrderStatus { get; set; } + + /// + /// 预报失败/处理中原因。 + /// + public string? PredeclareFailReason { get; set; } + + /// + /// 预报计费/排货已入队或后台处理中。 + /// + public bool ForecastBillingQueued { get; set; } } diff --git a/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs b/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs index 2f728b6..84a3c24 100644 --- a/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs +++ b/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs @@ -1,6 +1,9 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Oms.Application.Messaging.OrderSideEffects; +using Oms.Application.Messaging.ParcelCreate; using Oms.Application.Services.Orders.Batch; +using Oms.Domain.OrdersAggregate.Repositories; namespace Oms.Application.Integrations.SmallBagTrack; @@ -15,31 +18,42 @@ internal sealed class SmallBagTrackOrderNotifier( try { using var scope = scopeFactory.CreateScope(); - var coordinator = scope.ServiceProvider.GetRequiredService(); - var published = await coordinator.PublishTrackUpsertAsync(request, orderId, customerId); - if (!published) + var orderRepository = scope.ServiceProvider.GetRequiredService(); + var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == orderId); + if (order == null) { - await FallbackDirectAsync(request, orderId); + return; } + + var publisher = scope.ServiceProvider.GetRequiredService(); + await publisher.PublishTrackUpsertAsync(order, request, customerId); } catch (Exception ex) { logger.LogWarning( ex, - "RabbitMQ track upsert publish failed, fallback to direct call. orderId={OrderId}", + "Track upsert publish failed, fallback via batch coordinator. orderId={OrderId}", orderId); - await FallbackDirectAsync(request, orderId); + await FallbackBatchOrDirectAsync(request, orderId, customerId); } }); } - private async Task FallbackDirectAsync(SmallBagTrackUpsertRequest request, int orderId) + private async Task FallbackBatchOrDirectAsync( + SmallBagTrackUpsertRequest request, + int orderId, + int? customerId) { try { using var scope = scopeFactory.CreateScope(); - var client = scope.ServiceProvider.GetRequiredService(); - await client.UpsertAsync(request); + var coordinator = scope.ServiceProvider.GetRequiredService(); + var published = await coordinator.PublishTrackUpsertAsync(request, orderId, customerId); + if (!published) + { + var client = scope.ServiceProvider.GetRequiredService(); + await client.UpsertAsync(request); + } } catch (Exception ex) { @@ -49,23 +63,29 @@ internal sealed class SmallBagTrackOrderNotifier( public void EnqueueLifecycleTrackAppend(string orderNo, string trackCode, int orderId, string? authorization = null) { - _ = Task.Run(async () => + _ = PublishLifecycleAsync(orderNo, trackCode, orderId, authorization); + } + + private async Task PublishLifecycleAsync( + string orderNo, + string trackCode, + int orderId, + string? authorization) + { + try { - try - { - using var scope = scopeFactory.CreateScope(); - var client = scope.ServiceProvider.GetRequiredService(); - await client.AppendLifecycleTrackAsync(orderNo, trackCode, authorization); - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "Async lifecycle track append failed. orderId={OrderId} orderNo={OrderNo} trackCode={TrackCode}", - orderId, - orderNo, - trackCode); - } - }); + using var scope = scopeFactory.CreateScope(); + var publisher = scope.ServiceProvider.GetRequiredService(); + await publisher.PublishLifecycleTrackAppendAsync(orderId, orderNo, trackCode, authorization); + } + catch (Exception ex) + { + logger.LogWarning( + ex, + "Lifecycle track append publish failed. orderId={OrderId} orderNo={OrderNo} trackCode={TrackCode}", + orderId, + orderNo, + trackCode); + } } } diff --git a/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs new file mode 100644 index 0000000..9cba9f9 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs @@ -0,0 +1,22 @@ +using Oms.Application.Services.Orders.Issues; + +namespace Oms.Application.Messaging.OrderSideEffects; + +/// +/// 订单侧后置副作用发布（生命周期轨迹、自动触发等）。 +/// +public interface IOrderSideEffectEventPublisher +{ + Task PublishLifecycleTrackAppendAsync( + int orderId, + string orderNo, + string trackCode, + string? authorization = null, + CancellationToken cancellationToken = default); + + Task PublishAutoTriggerFireAsync( + int orderId, + AutoTriggerTiming timing, + int operatorUserId, + CancellationToken cancellationToken = default); +} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireEventHandler.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireEventHandler.cs new file mode 100644 index 0000000..670c52e --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireEventHandler.cs @@ -0,0 +1,48 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Oms.Application.Services.Orders.Issues; + +namespace Oms.Application.Messaging.OrderSideEffects; + +public sealed class OrderAutoTriggerFireEventHandler( + AutoTriggerRuntimeService autoTriggerRuntimeService, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderAutoTriggerFireIntegrationEvent @event, + CancellationToken cancellationToken) + { + if (@event.OrderId <= 0 + || !Enum.TryParse(@event.Timing, ignoreCase: true, out var timing)) + { + logger.LogWarning( + "AutoTrigger fire skipped invalid payload OrderId={OrderId} Timing={Timing}", + @event.OrderId, + @event.Timing); + return; + } + + using var unitOfWork = unitOfWorkManager.Begin(); + try + { + await autoTriggerRuntimeService.FireByOrderIdAsync( + @event.OrderId, + timing, + @event.OperatorUserId); + } + catch (Exception ex) + { + logger.LogWarning( + ex, + "AutoTrigger fire failed OrderId={OrderId} Timing={Timing}", + @event.OrderId, + timing); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireIntegrationEvent.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireIntegrationEvent.cs new file mode 100644 index 0000000..f3d04c2 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireIntegrationEvent.cs @@ -0,0 +1,19 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.OrderSideEffects; + +/// +/// 订单自动触发（下单 / 入库 / 出库时机），走 Fake.EventBus.RabbitMQ。 +/// +[Serializable] +public class OrderAutoTriggerFireIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + /// + /// 枚举名。 + /// + public string Timing { get; set; } = string.Empty; + + public int OperatorUserId { get; set; } +} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendEventHandler.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendEventHandler.cs new file mode 100644 index 0000000..2114097 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendEventHandler.cs @@ -0,0 +1,49 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Oms.Application.Integrations.SmallBagTrack; + +namespace Oms.Application.Messaging.OrderSideEffects; + +public sealed class OrderLifecycleTrackAppendEventHandler( + ISmallBagTrackOrderClient trackClient, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderLifecycleTrackAppendIntegrationEvent @event, + CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(@event.OrderNo) || string.IsNullOrWhiteSpace(@event.TrackCode)) + { + logger.LogWarning( + "Lifecycle track append skipped empty OrderNo/TrackCode OrderId={OrderId}", + @event.OrderId); + return; + } + + using var unitOfWork = unitOfWorkManager.Begin(); + try + { + await trackClient.AppendLifecycleTrackAsync( + @event.OrderNo, + @event.TrackCode, + @event.Authorization, + cancellationToken); + } + catch (Exception ex) + { + // 轨迹失败不反复打满队列；与原 Task.Run 吞异常策略一致 + logger.LogWarning( + ex, + "Lifecycle track append failed OrderId={OrderId} OrderNo={OrderNo} TrackCode={TrackCode}", + @event.OrderId, + @event.OrderNo, + @event.TrackCode); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendIntegrationEvent.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendIntegrationEvent.cs new file mode 100644 index 0000000..6144b5e --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendIntegrationEvent.cs @@ -0,0 +1,18 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.OrderSideEffects; + +/// +/// 订单生命周期轨迹追加（如入库 A0301、出库等），走 Fake.EventBus.RabbitMQ。 +/// +[Serializable] +public class OrderLifecycleTrackAppendIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + public string OrderNo { get; set; } = string.Empty; + + public string TrackCode { get; set; } = string.Empty; + + public string? Authorization { get; set; } +} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs new file mode 100644 index 0000000..078be98 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs @@ -0,0 +1,112 @@ +using Fake.EventBus.Distributed; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Oms.Application.Services.Orders.Issues; + +namespace Oms.Application.Messaging.OrderSideEffects; + +/// +/// 优先 Fake.EventBus.RabbitMQ；发布失败则同进程同步执行 Handler。 +/// +public sealed class OrderSideEffectEventPublisher( + IDistributedEventBus eventBus, + IServiceScopeFactory scopeFactory, + ILogger logger) : IOrderSideEffectEventPublisher +{ + public async Task PublishLifecycleTrackAppendAsync( + int orderId, + string orderNo, + string trackCode, + string? authorization = null, + CancellationToken cancellationToken = default) + { + if (orderId <= 0 || string.IsNullOrWhiteSpace(orderNo) || string.IsNullOrWhiteSpace(trackCode)) + { + return; + } + + var @event = new OrderLifecycleTrackAppendIntegrationEvent + { + OrderId = orderId, + OrderNo = orderNo.Trim(), + TrackCode = trackCode.Trim(), + Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim() + }; + + if (await TryPublishAsync(@event, cancellationToken)) + { + return; + } + + await ExecuteInScopeAsync( + h => h.HandleAsync(@event, cancellationToken), + cancellationToken); + } + + public async Task PublishAutoTriggerFireAsync( + int orderId, + AutoTriggerTiming timing, + int operatorUserId, + CancellationToken cancellationToken = default) + { + if (orderId <= 0) + { + return; + } + + var @event = new OrderAutoTriggerFireIntegrationEvent + { + OrderId = orderId, + Timing = timing.ToString(), + 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); + } + } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs b/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs new file mode 100644 index 0000000..2545a2e --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs @@ -0,0 +1,46 @@ +using Oms.Application.Dtos.Orders; +using Oms.Application.Integrations.SmallBagTrack; +using Oms.Domain.OrdersAggregate.Entities; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置副作用：发布到 Fake.EventBus.RabbitMQ；发布失败时回落同步执行。 +/// +public interface IOrderParcelCreateEventPublisher +{ + /// + /// 轨迹订单 + 创建成功轨迹（A0101）合并为一个事件。 + /// + Task PublishTrackUpsertAsync( + Order order, + SmallBagTrackUpsertRequest request, + int? customerId, + CancellationToken cancellationToken = default); + + Task PublishLabelGenerateAsync( + Order order, + int operatorUserId = 0, + CancellationToken cancellationToken = default); + + Task PublishCustoms9610UpsertAsync( + Order order, + string? platformCode = null, + CancellationToken cancellationToken = default); + + /// + /// 发布揽收挂接事件。返回 true 表示已入队；false 表示已同步回落执行（或参数不足未执行）。 + /// + Task PublishPickupAttachAsync( + OrderParcelCreateInput input, + int orderId, + CancellationToken cancellationToken = default); + + /// + /// 同步执行揽收挂接（用于创单 HTTP 需要回写 PickupNo 的场景）。 + /// + Task ExecutePickupAttachAsync( + OrderParcelCreateInput input, + int orderId, + CancellationToken cancellationToken = default); +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs new file mode 100644 index 0000000..e265284 --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs @@ -0,0 +1,198 @@ +using Fake.EventBus.Distributed; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using Oms.Application.Dtos.Orders; +using Oms.Application.Integrations.SmallBagTrack; +using Oms.Application.Services.Orders; +using Oms.Domain.OrdersAggregate.Consts; +using Oms.Domain.OrdersAggregate.Entities; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置事件发布器：优先走 Fake.EventBus.RabbitMQ，失败则同进程同步执行 Handler 逻辑。 +/// +public sealed class OrderParcelCreateEventPublisher( + IDistributedEventBus eventBus, + IServiceScopeFactory scopeFactory, + ILogger logger) : IOrderParcelCreateEventPublisher +{ + public async Task PublishTrackUpsertAsync( + Order order, + SmallBagTrackUpsertRequest request, + int? customerId, + CancellationToken cancellationToken = default) + { + var @event = new OrderParcelTrackUpsertIntegrationEvent + { + OrderId = order.Id, + CustomerId = customerId, + PayloadJson = JsonConvert.SerializeObject(request), + AppendInitialTrackOnSuccess = request.AppendInitialTrackOnSuccess, + Authorization = request.Authorization + }; + + if (await TryPublishAsync(@event, cancellationToken)) + { + return; + } + + await ExecuteInScopeAsync( + h => h.HandleAsync(@event, cancellationToken), + cancellationToken); + } + + public async Task PublishLabelGenerateAsync( + Order order, + int operatorUserId = 0, + CancellationToken cancellationToken = default) + { + var @event = new OrderParcelLabelGenerateIntegrationEvent + { + OrderId = order.Id, + OperatorUserId = operatorUserId + }; + + if (await TryPublishAsync(@event, cancellationToken)) + { + return; + } + + await ExecuteInScopeAsync( + h => h.HandleAsync(@event, cancellationToken), + cancellationToken); + } + + public async Task PublishCustoms9610UpsertAsync( + Order order, + string? platformCode = null, + CancellationToken cancellationToken = default) + { + var @event = new OrderParcelCustoms9610UpsertIntegrationEvent + { + OrderId = order.Id, + PlatformCode = platformCode + }; + + if (await TryPublishAsync(@event, cancellationToken)) + { + return; + } + + await ExecuteInScopeAsync( + h => h.HandleAsync(@event, cancellationToken), + cancellationToken); + } + + public async Task PublishPickupAttachAsync( + OrderParcelCreateInput input, + int orderId, + CancellationToken cancellationToken = default) + { + var @event = BuildPickupEvent(input, orderId); + if (@event == null) + { + return false; + } + + // 揽收回落由调用方 ExecutePickupAttachAsync 同步执行，避免重复建单且便于回写单号。 + return await TryPublishAsync(@event, cancellationToken); + } + + public async Task ExecutePickupAttachAsync( + OrderParcelCreateInput input, + int orderId, + CancellationToken cancellationToken = default) + { + if (orderId <= 0 + || input.SenderAddressId is not > 0 + || input.ExpectPickupTime is null + || input.ExpectPickupTime == default + || input.WarehouseId is not > 0) + { + return null; + } + + using var scope = scopeFactory.CreateScope(); + var pickupService = scope.ServiceProvider.GetRequiredService(); + return await pickupService.CreateForParcelInternalAsync(new PickupOrderCreateInput + { + CustomerId = input.CustomerId, + SenderAddressId = input.SenderAddressId.Value, + Direction = PickupDirectionCodes.NormalizeOrDefault(input.PickupDirection), + WarehouseId = input.WarehouseId.Value, + CarrierCode = input.CarrierCode, + CarrierService = input.CarrierService, + ExpectPickupTime = input.ExpectPickupTime.Value, + OrderIds = [orderId], + Remark = null + }); + } + + private static OrderParcelPickupAttachIntegrationEvent? BuildPickupEvent( + OrderParcelCreateInput input, + int orderId) + { + if (orderId <= 0 + || input.SenderAddressId is not > 0 + || input.ExpectPickupTime is null + || input.ExpectPickupTime == default + || input.WarehouseId is not > 0) + { + return null; + } + + return new OrderParcelPickupAttachIntegrationEvent + { + OrderId = orderId, + CustomerId = input.CustomerId, + SenderAddressId = input.SenderAddressId.Value, + WarehouseId = input.WarehouseId.Value, + PickupDirection = input.PickupDirection, + CarrierCode = input.CarrierCode, + CarrierService = input.CarrierService, + 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/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertEventHandler.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertEventHandler.cs new file mode 100644 index 0000000..bb292d2 --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertEventHandler.cs @@ -0,0 +1,44 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Oms.Application.Services.Orders; +using Oms.Domain.OrdersAggregate.Repositories; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 处理创单后 9610 汇总申报订单写入事件。 +/// +public sealed class OrderParcelCustoms9610UpsertEventHandler( + IOrderRepository orderRepository, + Customs9610OrderWriter customs9610OrderWriter, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderParcelCustoms9610UpsertIntegrationEvent @event, + CancellationToken cancellationToken) + { + using var unitOfWork = unitOfWorkManager.Begin(); + var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == @event.OrderId, cancellationToken); + if (order == null) + { + logger.LogWarning("9610 upsert skipped, order missing OrderId={OrderId}", @event.OrderId); + return; + } + + try + { + await customs9610OrderWriter.UpsertFromOrderAsync(order, @event.PlatformCode, cancellationToken); + } + catch (Exception ex) + { + // 与原同步路径一致：9610 失败不抛到总线重试风暴 + logger.LogWarning(ex, "9610 upsert failed OrderId={OrderId}", @event.OrderId); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertIntegrationEvent.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertIntegrationEvent.cs new file mode 100644 index 0000000..ef933ef --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertIntegrationEvent.cs @@ -0,0 +1,14 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置：按 9610 汇总申报写入/刷新 t_9610_order（非 9610 时 Handler 内会按规则清理 pending）。 +/// +[Serializable] +public class OrderParcelCustoms9610UpsertIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + public string? PlatformCode { get; set; } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateEventHandler.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateEventHandler.cs new file mode 100644 index 0000000..8e29720 --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateEventHandler.cs @@ -0,0 +1,38 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Oms.Application.Services.Orders; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 处理创单后系统面单生成事件。 +/// +public sealed class OrderParcelLabelGenerateEventHandler( + IServiceScopeFactory scopeFactory, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderParcelLabelGenerateIntegrationEvent @event, + CancellationToken cancellationToken) + { + using var unitOfWork = unitOfWorkManager.Begin(); + using var scope = scopeFactory.CreateScope(); + var orderService = scope.ServiceProvider.GetRequiredService(); + var result = await orderService.GenerateLabelPdfForBatchAsync(@event.OrderId, cancellationToken); + if (!result.Success) + { + // 面单失败不阻断创单主链路；记录后正常 Ack，避免反复重试打满队列 + logger.LogWarning( + "Label generate failed OrderId={OrderId} Error={Error}", + @event.OrderId, + result.Error); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateIntegrationEvent.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateIntegrationEvent.cs new file mode 100644 index 0000000..a7a807b --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateIntegrationEvent.cs @@ -0,0 +1,14 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置：生成系统面单 PDF。 +/// +[Serializable] +public class OrderParcelLabelGenerateIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + public int OperatorUserId { get; set; } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachEventHandler.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachEventHandler.cs new file mode 100644 index 0000000..488e95c --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachEventHandler.cs @@ -0,0 +1,51 @@ +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Oms.Application.Dtos.Orders; +using Oms.Application.Services.Orders; +using Oms.Domain.OrdersAggregate.Consts; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 处理创单后上门揽收挂接事件。 +/// +public sealed class OrderParcelPickupAttachEventHandler( + PickupOrderService pickupOrderService, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderParcelPickupAttachIntegrationEvent @event, + CancellationToken cancellationToken) + { + using var unitOfWork = unitOfWorkManager.Begin(); + try + { + await pickupOrderService.CreateForParcelInternalAsync(new PickupOrderCreateInput + { + CustomerId = @event.CustomerId, + SenderAddressId = @event.SenderAddressId, + Direction = PickupDirectionCodes.NormalizeOrDefault(@event.PickupDirection), + WarehouseId = @event.WarehouseId, + CarrierCode = @event.CarrierCode, + CarrierService = @event.CarrierService, + ExpectPickupTime = @event.ExpectPickupTime, + OrderIds = [@event.OrderId], + Remark = null + }); + } + catch (Exception ex) + { + // 与原同步路径一致：揽收失败不回滚物流单，仅记日志（列表可补建） + logger.LogWarning( + ex, + "Pickup attach failed OrderId={OrderId}", + @event.OrderId); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachIntegrationEvent.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachIntegrationEvent.cs new file mode 100644 index 0000000..18afd3d --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachIntegrationEvent.cs @@ -0,0 +1,26 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置：上门揽收时创建并挂接揽收订单（可选再预报承运）。 +/// +[Serializable] +public class OrderParcelPickupAttachIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + public int? CustomerId { get; set; } + + public int SenderAddressId { get; set; } + + public int WarehouseId { get; set; } + + public string? PickupDirection { get; set; } + + public string? CarrierCode { get; set; } + + public string? CarrierService { get; set; } + + public DateTime ExpectPickupTime { get; set; } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertEventHandler.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertEventHandler.cs new file mode 100644 index 0000000..6347a3c --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertEventHandler.cs @@ -0,0 +1,76 @@ +using Fake.Application; +using Fake.DependencyInjection; +using Fake.EventBus; +using Fake.UnitOfWork; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using Oms.Application.Integrations.SmallBagTrack; +using Oms.Domain.OrdersAggregate.Repositories; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 处理「轨迹订单 + 创建成功轨迹」合并事件。 +/// +public sealed class OrderParcelTrackUpsertEventHandler( + IOrderRepository orderRepository, + ISmallBagTrackOrderClient trackClient, + ISmallBagTrackUpsertRequestBuilder requestBuilder, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger) + : IEventHandler, ITransientDependency +{ + public async Task HandleAsync( + OrderParcelTrackUpsertIntegrationEvent @event, + CancellationToken cancellationToken) + { + using var unitOfWork = unitOfWorkManager.Begin(); + + var request = await ResolveRequestAsync(@event, cancellationToken); + request.AppendInitialTrackOnSuccess = @event.AppendInitialTrackOnSuccess; + if (string.IsNullOrWhiteSpace(request.Authorization) + && !string.IsNullOrWhiteSpace(@event.Authorization)) + { + request.Authorization = @event.Authorization.Trim(); + } + + var result = await trackClient.UpsertDebugAsync(request, cancellationToken); + if (result.Skipped) + { + logger.LogInformation( + "Track upsert skipped OrderId={OrderId} Reason={Reason}", + @event.OrderId, + result.SkipReason); + } + else if (!result.Success) + { + throw new BusinessException( + $"轨迹订单创建失败: HTTP {result.StatusCode}"); + } + + await unitOfWork.CompleteAsync(cancellationToken); + } + + private async Task ResolveRequestAsync( + OrderParcelTrackUpsertIntegrationEvent @event, + CancellationToken cancellationToken) + { + if (!string.IsNullOrWhiteSpace(@event.PayloadJson)) + { + var fromPayload = JsonConvert.DeserializeObject(@event.PayloadJson); + if (fromPayload != null && !string.IsNullOrWhiteSpace(fromPayload.OrderNo)) + { + return fromPayload; + } + } + + var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == @event.OrderId, cancellationToken) + ?? throw new BusinessException("订单不存在"); + if (string.IsNullOrWhiteSpace(order.ServiceNo)) + { + throw new BusinessException("订单运单号为空，无法同步轨迹"); + } + + return await requestBuilder.BuildAsync(order, cancellationToken: cancellationToken); + } +} diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertIntegrationEvent.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertIntegrationEvent.cs new file mode 100644 index 0000000..214c668 --- /dev/null +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertIntegrationEvent.cs @@ -0,0 +1,28 @@ +using Fake.EventBus.Distributed; + +namespace Oms.Application.Messaging.ParcelCreate; + +/// +/// 创单后置：新增轨迹订单，并在成功后追加「创建订单成功」轨迹（A0101）。 +/// 二者合并为同一集成事件，由 Handler 内 UpsertSmallBag + AppendInitialTrackOnSuccess 一次完成。 +/// +[Serializable] +public class OrderParcelTrackUpsertIntegrationEvent : IntegrationEvent +{ + public int OrderId { get; set; } + + public int? CustomerId { get; set; } + + /// + /// 序列化后的 ； + /// 为空时 Handler 按 OrderId 重建。 + /// + public string? PayloadJson { get; set; } + + /// + /// 是否在 Upsert 成功后追加创单默认轨迹 A0101。 + /// + public bool AppendInitialTrackOnSuccess { get; set; } = true; + + public string? Authorization { get; set; } +} diff --git a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs index 24960a7..8131be4 100644 --- a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs +++ b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs @@ -1,6 +1,8 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Configuration; using Oms.Application.Messaging; +using Oms.Application.Messaging.OrderSideEffects; +using Oms.Application.Messaging.ParcelCreate; namespace Oms.Application; @@ -11,6 +13,8 @@ public static class OrderBatchMessagingServiceExtensions services.Configure(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName)); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); // 消费端由 Fake.EventBus.RabbitMQ（RabbitMqEventBus HostedService）托管 services.AddHostedService(); return services; diff --git a/src/Oms.Application/Services/Operations/OutboundRecordService.cs b/src/Oms.Application/Services/Operations/OutboundRecordService.cs index 30f84ff..b64d4b7 100644 --- a/src/Oms.Application/Services/Operations/OutboundRecordService.cs +++ b/src/Oms.Application/Services/Operations/OutboundRecordService.cs @@ -161,18 +161,8 @@ public class OutboundRecordService : ApplicationService var savedEntity = await OutboundRecordRepository.InsertAsync(entity); - try - { - var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim()); - if (refreshed != null) - { - await AutoTriggerRuntimeService.FireAsync(refreshed, AutoTriggerTiming.Outbound, operatorId); - } - } - catch - { - // 自动触发失败不影响出库主流程 - } + // 自动触发改走 Fake.EventBus，避免拉长出库扫码 RT + AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Outbound, operatorId); return ObjectMapper.Map(savedEntity); } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs index 913cd7a..d74cfe6 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs @@ -49,6 +49,7 @@ public sealed class OrderBatchActionExecutor( OrderBatchActionCodes.Print => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.GenerateLabelPdf => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.InboundBilling => await orderActionHandler.ExecuteAsync(message, cancellationToken), + OrderBatchActionCodes.ForecastBilling => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.Reallocate => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.MarkProblematic => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.MarkClaimed => await orderActionHandler.ExecuteAsync(message, cancellationToken), diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs index b8f75fd..703bf09 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs @@ -269,6 +269,7 @@ public static class OrderBatchJobItemFeedback OrderBatchActionCodes.Print => "打印成功", OrderBatchActionCodes.GenerateLabelPdf => "面单生成成功", OrderBatchActionCodes.InboundBilling => "入库计费/排货完成", + OrderBatchActionCodes.ForecastBilling => "预报计费/排货完成", OrderBatchActionCodes.Reallocate => "重新排货成功", OrderBatchActionCodes.MarkProblematic => "问题件创建成功", OrderBatchActionCodes.MarkClaimed => "理赔件创建成功", diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchOrderActionHandler.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchOrderActionHandler.cs index f578a60..f363cd8 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchOrderActionHandler.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchOrderActionHandler.cs @@ -67,6 +67,14 @@ public sealed class OrderBatchOrderActionHandler(IServiceScopeFactory scopeFacto await orderService.ProcessInboundBillingForBatchAsync(message.OrderId, cancellationToken); return OrderBatchActionExecuteResult.Ok(); + case OrderBatchActionCodes.ForecastBilling: + await orderService.ProcessForecastBillingForBatchAsync( + message.OrderId, + message.PayloadJson, + message.OperatorUserId, + cancellationToken); + return OrderBatchActionExecuteResult.Ok(); + case OrderBatchActionCodes.Reallocate: await orderService.ReallocateForBatchAsync( message.OrderId, diff --git a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs index e8276d1..9b70856 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -1,6 +1,6 @@ using Fake.DependencyInjection; -using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Oms.Application.Messaging.OrderSideEffects; using Oms.Domain.ChannelsAggregate.Consts; using Oms.Domain.ChannelsAggregate.Entities; using Oms.Domain.ChannelsAggregate.Repositories; @@ -16,13 +16,13 @@ public class AutoTriggerRuntimeService( OrderAutoTriggerService orderAutoTriggerService, OrderExceptionWorkflowService exceptionWorkflowService, OrderIssueNotificationService issueNotificationService, - IServiceScopeFactory scopeFactory, + IOrderSideEffectEventPublisher sideEffectEventPublisher, ILogger logger) : IScopedDependency { private const string InterceptOperation = "拦截"; /// - /// 后台异步触发，避免拉长入库/出库扫码请求。 + /// 后台异步触发（Fake.EventBus.RabbitMQ），避免拉长入库/出库/创单请求。 /// public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId) { @@ -31,23 +31,7 @@ public class AutoTriggerRuntimeService( return; } - _ = Task.Run(async () => - { - try - { - using var scope = scopeFactory.CreateScope(); - var runtime = scope.ServiceProvider.GetRequiredService(); - await runtime.FireByOrderIdAsync(orderId, timing, operatorUserId); - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "异步自动触发失败 OrderId={OrderId} Timing={Timing}", - orderId, - timing); - } - }); + _ = 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 8f7bc82..8dc9c39 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -9,6 +9,8 @@ using Oms.Application.Dtos.Orders; using Oms.Application.Integrations.SmallBagTrack; using Oms.Application.Labeling; using Oms.Application.Messaging; +using Oms.Application.Messaging.OrderSideEffects; +using Oms.Application.Messaging.ParcelCreate; using Oms.Application.Shared; using Oms.Application.Storage; using Oms.Application.Services.Orders.Batch; @@ -36,6 +38,8 @@ public partial class OrderService private const int ParcelBatchMaxCount = 2000; private const string InboundBillingProcessingReason = "入库后计费/排货处理中"; private const string InboundBillingEnqueueFailReason = "入库计费任务投递失败，请重跑入库计费"; + private const string ForecastBillingProcessingReason = "预报计费/排货处理中"; + private const string ForecastBillingEnqueueFailReason = "预报计费任务投递失败，请重新预报"; private static readonly ConcurrentDictionary InboundBillingInFlight = new(); @@ -143,10 +147,11 @@ public partial class OrderService // 客户主动「先保存为草稿」：落库草稿后直接返回，不跑预报计费。 if (input.SaveAsDraft) { - await TryUpsertCustoms9610Async(order, input.PlatformCode); + await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode); return await BuildParcelCreateOutputAsync(order); } + // 偏远标记轻量同步；计费重/应收/排货走 ForecastBilling MQ（失败不回滚建单）。 order.IsRemote = await IsRemoteHitAsync( product, snapshotBinding.Rules, @@ -154,62 +159,24 @@ public partial class OrderService input.Receiver?.Province, input.Receiver?.City, input.Receiver?.ZipCode) ? (byte)1 : (byte)0; - - var forecast = weights.First(x => x.Type == OrderWeightTypeCodes.Forecast); - await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast); - SyncWeightRows(weights, forecast); - await OrderRepository.ReplaceWeightsAsync(order.Id, weights); - await TryUpsertCustoms9610Async(order, input.PlatformCode); - - var customer = await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId); - var priceTierKey = await ResolvePriceTierKeyAsync(order.CustomerId); - var billing = await ResolveBillingContextAsync( - product, - snapshotBinding, - new OrderBillingAddress( - country.CountryCode, - input.Receiver?.Province, - input.Receiver?.City, - input.Receiver?.ZipCode), - forecast.ChargeableWeight ?? 0, - order.CargoAttribute ?? CargoAttributeCodes.General, - order.WarehouseId, - priceTierKey); - - await UpsertReceivableFeesAsync( - order, - billing, - OrderFeeChargingStageCodes.Forecast); - - // 预报阶段尝试排货写应付；失败不阻断创建（仅记录 AllocationPendingReason）。 - await TryConfirmServiceChannelAndPayableAsync( - order, - product, - billing, - OrderFeeChargingStageCodes.Forecast); - - await ApplyMarketingSnapshotOnCreateAsync(order, input.CouponInstanceId); - await CaptureForecastBillingSnapshotAsync(order); - - order.OrderStatus = OrderStatusCodes.Predeclared; - order.PredeclareFailReason = string.Empty; + order.OrderStatus = OrderStatusCodes.Draft; + order.PredeclareFailReason = ForecastBillingProcessingReason; + order.AllocationPendingReason = string.Empty; order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); - await EnqueueSmallBagTrackUpsertAsync(order, customer, input.Receiver, appendInitialTrack: true); - await EnqueueGenerateLabelPdfAsync(order); + await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode); + await EnqueueForecastBillingAsync( + order, + input.CouponInstanceId, + operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0)); - try - { - var triggerUserId = operatorUserId ?? CurrentUser.GetId(); - await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId); - } - catch - { - // 自动触发失败不影响预报主流程 - } - - return await BuildParcelCreateOutputAsync(order); + var output = await BuildParcelCreateOutputAsync(order); + output.ForecastBillingQueued = string.Equals( + order.PredeclareFailReason, + ForecastBillingProcessingReason, + StringComparison.Ordinal); + return output; } catch (Exception ex) { @@ -2287,8 +2254,7 @@ public partial class OrderService } /// - /// 草稿重新预报：按订单当前收件/件重尺/快照重跑黑名单、计重、分区与预报应收。 - /// 失败时保持草稿并写入 PredeclareFailReason。 + /// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货投递 ForecastBilling MQ。 /// internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null) { @@ -2319,63 +2285,52 @@ public partial class OrderService throw new BusinessException("收件地址命中黑名单区域"); } - order.IsRemote = await IsRemoteHitAsync( - product, - snapshotBinding.Rules, - address.CountryCode, - address.Province, - address.City, - address.ZipCode) ? (byte)1 : (byte)0; + var weights = await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]); + if (!weights.Any(x => + string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase))) + { + throw new BusinessException("订单缺少预报件重尺"); + } - var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) - .OrderBy(x => x.Id) - .ToList(); - var forecast = weights.FirstOrDefault(x => - string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase)) - ?? throw new BusinessException("订单缺少预报件重尺"); - - await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast); - SyncWeightRows(weights, forecast); - await OrderRepository.ReplaceWeightsAsync(order.Id, weights); - - var billing = await ResolveOrderBillingContextAsync( - order, - product, - forecast.ChargeableWeight ?? 0, - snapshotBinding); - - await UpsertReceivableFeesAsync( - order, - billing, - OrderFeeChargingStageCodes.Forecast, - replaceExistingReceivable: true); - - order.OrderStatus = OrderStatusCodes.Predeclared; - order.PredeclareFailReason = string.Empty; + order.OrderStatus = OrderStatusCodes.Draft; + order.PredeclareFailReason = ForecastBillingProcessingReason; order.AllocationPendingReason = string.Empty; order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); - var customer = order.CustomerId.HasValue - ? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) - : null; - await EnqueueSmallBagTrackUpsertAsync(order, customer, receiver: null, appendInitialTrack: true); - await EnqueueGenerateLabelPdfAsync(order); + await EnqueueForecastBillingAsync( + order, + order.CouponInstanceId, + operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0)); - try + // 同步回落失败时给出明确错误；入队成功则保持「处理中」由列表观察结果。 + var refreshed = await OrderRepository.FirstOrDefaultAsync(x => x.Id == order.Id); + if (refreshed != null + && string.Equals(refreshed.OrderStatus, OrderStatusCodes.Draft, StringComparison.OrdinalIgnoreCase) + && !string.IsNullOrWhiteSpace(refreshed.PredeclareFailReason) + && !string.Equals( + refreshed.PredeclareFailReason, + ForecastBillingProcessingReason, + StringComparison.Ordinal)) { - var triggerUserId = operatorUserId ?? CurrentUser.GetId(); - await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId); - } - catch - { - // 自动触发失败不影响预报主流程 + throw CreateDraftSavedBusinessException( + refreshed, + StripDraftSavedMarker(refreshed.PredeclareFailReason)); } } catch (BusinessException ex) { + if (string.Equals(ex.ErrorCode, OrderParcelDraftFailData.ErrorCodeValue, StringComparison.Ordinal)) + { + throw; + } + var cleanReason = StripDraftSavedMarker(ex.Message); - await MarkParcelDraftFailedAsync(order, cleanReason); + if (!string.Equals(cleanReason, ForecastBillingProcessingReason, StringComparison.Ordinal)) + { + await MarkParcelDraftFailedAsync(order, cleanReason); + } + throw CreateDraftSavedBusinessException(order, cleanReason); } catch (Exception ex) @@ -2464,7 +2419,7 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); - EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.InboundTrackCode); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.InboundTrackCode); await EnqueueInboundBillingAsync(order); } finally @@ -2646,6 +2601,256 @@ public partial class OrderService return text[..512]; } + /// + /// 预报计费/排货异步任务（创单与草稿预报共用）。 + /// + internal async Task ProcessForecastBillingForBatchAsync( + int orderId, + string? payloadJson = null, + int operatorUserId = 0, + CancellationToken cancellationToken = default) + { + var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId, cancellationToken); + if (order == null) + { + return; + } + + // 已预报成功则跳过（幂等） + if (string.Equals(order.OrderStatus, OrderStatusCodes.Predeclared, StringComparison.OrdinalIgnoreCase) + && string.IsNullOrWhiteSpace(order.PredeclareFailReason)) + { + return; + } + + int? couponInstanceId = order.CouponInstanceId; + if (!string.IsNullOrWhiteSpace(payloadJson)) + { + try + { + var payload = JsonConvert.DeserializeObject(payloadJson); + if (payload?.CouponInstanceId is > 0) + { + couponInstanceId = payload.CouponInstanceId; + } + } + catch + { + // ignore bad payload + } + } + + if (string.IsNullOrWhiteSpace(order.ProductCode)) + { + await MarkParcelDraftFailedAsync(order, "订单未绑定销售产品，无法预报"); + return; + } + + var product = await SalesProductRepository.GetByProductCodeAsync(order.ProductCode.Trim()); + if (product == null || product.IsActive != EnableStatus.Enabled) + { + await MarkParcelDraftFailedAsync(order, "销售产品不存在或未启用"); + return; + } + + ProductSnapshotBinding snapshotBinding; + OrderBillingAddress address; + try + { + snapshotBinding = await ResolveSnapshotBindingFromOrderAsync(product, order); + address = await ResolveOrderBillingAddressAsync(order); + order.IsRemote = await IsRemoteHitAsync( + product, + snapshotBinding.Rules, + address.CountryCode, + address.Province, + address.City, + address.ZipCode) ? (byte)1 : (byte)0; + } + catch (Exception ex) + { + await MarkParcelDraftFailedAsync( + order, + ex is BusinessException be ? be.Message : $"预报快照解析失败：{ex.Message}"); + return; + } + + var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) + .OrderBy(x => x.Id) + .ToList(); + var forecast = weights.FirstOrDefault(x => + string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase)); + if (forecast == null) + { + await MarkParcelDraftFailedAsync(order, "订单缺少预报件重尺"); + return; + } + + try + { + await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast); + SyncWeightRows(weights, forecast); + await OrderRepository.ReplaceWeightsAsync(order.Id, weights); + } + catch (Exception ex) + { + await MarkParcelDraftFailedAsync( + order, + ex is BusinessException be ? be.Message : $"计费重计算失败：{ex.Message}"); + return; + } + + OrderBillingContext billing; + try + { + billing = await ResolveOrderBillingContextAsync( + order, + product, + forecast.ChargeableWeight ?? 0, + snapshotBinding); + } + catch (Exception ex) + { + await MarkParcelDraftFailedAsync( + order, + ex is BusinessException be ? be.Message : $"计费上下文解析失败：{ex.Message}"); + return; + } + + try + { + await UpsertReceivableFeesAsync( + order, + billing, + OrderFeeChargingStageCodes.Forecast, + replaceExistingReceivable: true); + } + catch (Exception ex) + { + await MarkParcelDraftFailedAsync( + order, + ex is BusinessException be ? be.Message : $"预报应收计费失败：{ex.Message}"); + return; + } + + // 排货失败不阻断预报成功（与原创单同步路径一致） + await TryConfirmServiceChannelAndPayableAsync( + order, + product, + billing, + OrderFeeChargingStageCodes.Forecast); + + try + { + await ApplyMarketingSnapshotOnCreateAsync(order, couponInstanceId); + } + catch (Exception ex) + { + var logger = LazyServiceProvider.GetRequiredService() + .CreateLogger("Oms.ForecastBilling"); + logger.LogWarning(ex, "预报营销快照失败 OrderId={OrderId}", order.Id); + } + + try + { + await CaptureForecastBillingSnapshotAsync(order); + } + catch + { + // ignore + } + + order.OrderStatus = OrderStatusCodes.Predeclared; + order.PredeclareFailReason = string.Empty; + order.UpdateTime = DateTime.Now; + await OrderRepository.UpdateAsync(order); + + var customer = order.CustomerId.HasValue + ? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) + : null; + await PublishParcelCreateTrackAndLabelAsync( + order, + customer, + receiver: null, + appendInitialTrack: true, + operatorUserId); + + AutoTriggerRuntimeService.EnqueueFire( + order.Id, + AutoTriggerTiming.OrderCreate, + operatorUserId); + } + + private sealed class ForecastBillingJobPayload + { + public int? CouponInstanceId { get; set; } + } + + private async Task EnqueueForecastBillingAsync( + Order order, + int? couponInstanceId, + int operatorUserId) + { + var logger = LazyServiceProvider.GetRequiredService() + .CreateLogger("Oms.ForecastBilling"); + var payloadJson = JsonConvert.SerializeObject(new ForecastBillingJobPayload + { + CouponInstanceId = couponInstanceId + }); + + try + { + var publisher = LazyServiceProvider.GetRequiredService(); + if (!publisher.IsEnabled) + { + try + { + await ProcessForecastBillingForBatchAsync(order.Id, payloadJson, operatorUserId); + } + catch (Exception ex) + { + logger.LogError( + ex, + "预报计费同步执行失败 OrderId={OrderId} ServiceNo={ServiceNo}", + order.Id, + order.ServiceNo); + await MarkParcelDraftFailedAsync( + order, + TrimPendingReason($"预报计费失败：{ex.Message}")); + } + + return; + } + + var requestId = $"forecast-billing:{order.Id}:{Guid.NewGuid():N}"; + await BatchJobCoordinator.CreateAndEnqueueAsync( + new OrderBatchJobCreateInput + { + Action = OrderBatchActionCodes.ForecastBilling, + OrderIds = [order.Id], + RequestId = requestId, + PayloadJson = payloadJson + }, + operatorUserId); + } + catch (Exception ex) + { + logger.LogError( + ex, + "预报计费任务投递失败 OrderId={OrderId} ServiceNo={ServiceNo}", + order.Id, + order.ServiceNo); + try + { + await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason); + } + catch + { + // ignore + } + } + } + private async Task EnqueueInboundBillingAsync(Order order) { var logger = LazyServiceProvider.GetRequiredService() @@ -2819,7 +3024,7 @@ public partial class OrderService order.UpdateTime = now; await OrderRepository.UpdateAsync(order); - EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.OutboundTrackCode); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.OutboundTrackCode); } private async Task BuildParcelCreateOutputAsync(Order order) @@ -2877,46 +3082,56 @@ public partial class OrderService TrackNo = order.TrackNo, ConsigneeCountry = order.ConsigneeCountry, ConsigneeCountryName = order.ConsigneeCountryName, - LabelUrl = labelUrl + LabelUrl = labelUrl, + OrderStatus = order.OrderStatus, + PredeclareFailReason = order.PredeclareFailReason, + ForecastBillingQueued = string.Equals( + order.PredeclareFailReason, + ForecastBillingProcessingReason, + StringComparison.Ordinal) }; } - private ISmallBagTrackOrderNotifier SmallBagTrackOrderNotifier => - LazyServiceProvider.GetRequiredService(); + private IOrderParcelCreateEventPublisher ParcelCreateEventPublisher => + LazyServiceProvider.GetRequiredService(); - private async Task EnqueueGenerateLabelPdfAsync(Order order) + private IOrderSideEffectEventPublisher SideEffectEventPublisher => + LazyServiceProvider.GetRequiredService(); + + /// + /// 创单/草稿预报成功后：轨迹订单+A0101 合并事件，系统面单独立事件（Fake.EventBus.RabbitMQ）。 + /// + private async Task PublishParcelCreateTrackAndLabelAsync( + Order order, + Customer? customer, + OrderParcelReceiverInput? receiver, + bool appendInitialTrack, + int? operatorUserId = null) { - try + var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiver); + request.AppendInitialTrackOnSuccess = appendInitialTrack; + var authorization = LazyServiceProvider + .GetRequiredService() + .HttpContext? + .Request + .Headers + .Authorization + .ToString(); + if (!string.IsNullOrWhiteSpace(authorization)) { - var publisher = LazyServiceProvider.GetRequiredService(); - if (!publisher.IsEnabled) - { - // 本地未开队列：同步生成，避免 Pending 悬挂 - var mapped = (await MapOrderOutputsAsync([order])).First(); - await LabelPdfGenerationService.GenerateAndPersistAsync( - order, - mapped, - operatorUserId: CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); - return; - } + request.Authorization = authorization; + } - await BatchJobCoordinator.CreateAndEnqueueAsync( - new OrderBatchJobCreateInput - { - Action = OrderBatchActionCodes.GenerateLabelPdf, - OrderIds = [order.Id], - RequestId = $"label-pdf:{order.Id}:{Guid.NewGuid():N}" - }, - CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); - } - catch (Exception) - { - // 投递失败不阻断创单；打印时会补生成 - } + await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id); + await ParcelCreateEventPublisher.PublishLabelGenerateAsync( + order, + operatorUserId + ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0)); } /// /// 小包预报成功后异步同步轨迹订单（POST /ProviderService/Order/UpsertSmallBag）。 + /// 入库等非创单路径仍可调用；创单路径请走 。 /// private async Task EnqueueSmallBagTrackUpsertAsync( Order order, @@ -2938,10 +3153,10 @@ public partial class OrderService request.Authorization = authorization; } - SmallBagTrackOrderNotifier.Enqueue(request, order.Id, customer?.Id); + await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id); } - private void EnqueueLifecycleTrackAppend(Order order, string trackCode) + private async Task EnqueueLifecycleTrackAppendAsync(Order order, string trackCode) { if (string.IsNullOrWhiteSpace(order.ServiceNo)) { @@ -2956,10 +3171,10 @@ public partial class OrderService .Authorization .ToString(); - SmallBagTrackOrderNotifier.EnqueueLifecycleTrackAppend( + await SideEffectEventPublisher.PublishLifecycleTrackAppendAsync( + order.Id, order.ServiceNo, trackCode, - order.Id, string.IsNullOrWhiteSpace(authorization) ? null : authorization); } diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 9d33c6c..35a155d 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -11,6 +11,7 @@ using Oms.Application.Labeling; using Oms.Application.Shared; using Oms.Application.Storage; using Newtonsoft.Json; +using Oms.Application.Messaging.ParcelCreate; using Oms.Application.Services.Orders.Batch; using Oms.Domain.ChannelsAggregate.Repositories; using Oms.Domain.CustomersAggregate.Repositories; @@ -660,21 +661,24 @@ public partial class OrderService : ApplicationService return; } + var publisher = LazyServiceProvider.GetRequiredService(); try { - var pickupService = LazyServiceProvider.GetRequiredService(); - var pickup = await pickupService.CreateForParcelInternalAsync(new PickupOrderCreateInput + // 优先入 Fake.EventBus；发布失败则同步回落建揽收并回写单号。 + var queued = await publisher.PublishPickupAttachAsync(input, output.OrderId); + if (queued) { - CustomerId = input.CustomerId, - SenderAddressId = input.SenderAddressId.Value, - Direction = PickupDirectionCodes.NormalizeOrDefault(input.PickupDirection), - WarehouseId = input.WarehouseId.Value, - CarrierCode = input.CarrierCode, - CarrierService = input.CarrierService, - ExpectPickupTime = input.ExpectPickupTime.Value, - OrderIds = [output.OrderId], - Remark = null - }); + output.PickupQueued = true; + return; + } + + var pickup = await publisher.ExecutePickupAttachAsync(input, output.OrderId); + if (pickup == null) + { + output.PickupCreateError = "揽收单创建失败"; + return; + } + output.PickupNo = pickup.PickupNo; output.PickupOrderId = pickup.Id; output.CarrierWaybillNo = pickup.CarrierWaybillNo; diff --git a/src/Oms.Domain/OrdersAggregate/Consts/OrderBatchJobStatusCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderBatchJobStatusCodes.cs index da96b64..e6250a6 100644 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderBatchJobStatusCodes.cs +++ b/src/Oms.Domain/OrdersAggregate/Consts/OrderBatchJobStatusCodes.cs @@ -33,6 +33,8 @@ public static class OrderBatchActionCodes public const string Print = "print"; public const string GenerateLabelPdf = "generate-label-pdf"; public const string InboundBilling = "inbound-billing"; + /// 创单/草稿预报：预报应收计费 + 排货应付。 + public const string ForecastBilling = "forecast-billing"; public const string Reallocate = "reallocate"; public const string Export = "export"; public const string ParcelCreate = "parcel-create"; -- 2.45.2 From 64737284f3d5d51c2f53bb733530baef00e7e6b6 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Sat, 10 Oct 2026 11:46:33 +0800 Subject: [PATCH 2/4] =?UTF-8?q?feat:=20=E8=BF=90=E5=8D=95=E6=93=8D?= =?UTF-8?q?=E4=BD=9C=E6=97=A5=E5=BF=97=E6=94=B9=E4=B8=BA=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E6=91=98=E8=A6=81=EF=BC=8C=E8=A1=A5=E9=BD=90=E8=B4=B9=E7=94=A8?= =?UTF-8?q?/=E6=94=AF=E4=BB=98/=E9=A2=84=E6=8A=A5/=E6=89=93=E5=8D=B0?= =?UTF-8?q?=E5=9F=8B=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- sql/migrate_order_operation_log.sql | 2 +- .../Dtos/Orders/OrderOperationLogDtos.cs | 5 ++ .../Orders/Batch/OrderBatchActionExecutor.cs | 26 ++++++ .../Orders/OrderOperationLogSnapshot.cs | 89 +++++++++++++++++++ .../Orders/OrderOperationLogWriter.cs | 87 +++++++++++++++--- .../Services/Orders/OrderService.OrderEdit.cs | 32 ++++++- .../Orders/OrderService.ParcelBilling.cs | 14 +++ .../Services/Orders/OrderService.cs | 39 ++++++++ .../Consts/OrderOperationActionCodes.cs | 12 +++ 9 files changed, 289 insertions(+), 17 deletions(-) diff --git a/sql/migrate_order_operation_log.sql b/sql/migrate_order_operation_log.sql index cd0aba0..094b880 100644 --- a/sql/migrate_order_operation_log.sql +++ b/sql/migrate_order_operation_log.sql @@ -17,7 +17,7 @@ SET @sql := IF( `Id` INT NOT NULL AUTO_INCREMENT COMMENT ''主键'', `OrderId` INT NOT NULL COMMENT ''运单Id'', `ServiceNo` VARCHAR(64) NULL COMMENT ''运单号冗余'', - `Action` VARCHAR(32) NOT NULL COMMENT ''动作码 editBasic/editWeight/recalcReceivable/recalcPayable/reallocate'', + `Action` VARCHAR(32) NOT NULL COMMENT ''动作码 editBasic/editWeight/recalcReceivable/recalcPayable/reallocate/bindSnapshotVersions/updateAdditionalServices/addFee/pay/preDeclare/downstreamForward/print'', `ActionName` VARCHAR(64) NOT NULL COMMENT ''动作中文名'', `OperatorId` INT NOT NULL DEFAULT 0 COMMENT ''操作人用户Id'', `OperatorName` VARCHAR(64) NULL COMMENT ''操作人展示名'', diff --git a/src/Oms.Application/Dtos/Orders/OrderOperationLogDtos.cs b/src/Oms.Application/Dtos/Orders/OrderOperationLogDtos.cs index 3b787e0..c5a3c84 100644 --- a/src/Oms.Application/Dtos/Orders/OrderOperationLogDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderOperationLogDtos.cs @@ -35,6 +35,11 @@ public class OrderOperationLogOutput public string? Remark { get; set; } + /// + /// 列表「内容」列：由备注 + 字段变更摘要拼装。 + /// + public string Summary { get; set; } = string.Empty; + public DateTime CreateTime { get; set; } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs index d74cfe6..b5c9283 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs @@ -26,6 +26,7 @@ public sealed class OrderBatchActionExecutor( OrderDownstreamCredentialResolver downstreamCredentialResolver, IItfxForwardClient itfxForwardClient, OrderFileManager orderFileManager, + OrderOperationLogWriter operationLogWriter, OrderBatchCreateActionHandler createActionHandler, OrderBatchOrderActionHandler orderActionHandler) : IOrderBatchActionExecutor, IScopedDependency { @@ -225,6 +226,8 @@ public sealed class OrderBatchActionExecutor( businessMessage)); } + var beforeSnap = OrderOperationLogSnapshot.CaptureOrder(order); + if (ItfxCreateOrderOneDataParser.TryParse(forward.ResponseBody, out var data) && data != null) { @@ -243,6 +246,29 @@ public sealed class OrderBatchActionExecutor( } } + var afterSnap = OrderOperationLogSnapshot.CaptureOrder(order); + var remarkParts = new List { $"供应商 {creds.SupplierCode}" }; + if (!string.IsNullOrWhiteSpace(order.DownstreamServiceNo)) + { + remarkParts.Add($"服务商单号 {order.DownstreamServiceNo}"); + } + + if (!string.IsNullOrWhiteSpace(order.TrackNo)) + { + remarkParts.Add($"跟踪号 {order.TrackNo}"); + } + + await operationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.DownstreamForward, + beforeSnap, + afterSnap, + string.Join("；", remarkParts), + message.OperatorUserId > 0 ? message.OperatorUserId : null, + operatorName: null, + operatorType: "staff", + source: "job"); + return OrderBatchActionExecuteResult.Ok( serviceNo: order.ServiceNo, trackNo: order.TrackNo, diff --git a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs index 527471a..f37c5d4 100644 --- a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs +++ b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs @@ -1,4 +1,6 @@ using System.Globalization; +using Newtonsoft.Json; +using Oms.Application.Shared; using Oms.Domain.OrdersAggregate.Consts; using Oms.Domain.OrdersAggregate.Entities; @@ -9,6 +11,8 @@ namespace Oms.Application.Services.Orders; /// public static class OrderOperationLogSnapshot { + public const int SummaryMaxChangeItems = 5; + private static readonly Dictionary Labels = new(StringComparer.OrdinalIgnoreCase) { ["orderStatus"] = "订单状态", @@ -31,6 +35,7 @@ public static class OrderOperationLogSnapshot ["salesProductRulesVersion"] = "规则版本", ["salesProductPriceVersion"] = "价格版本", ["isPaid"] = "是否已支付", + ["additionalServices"] = "附加服务", ["sender.name"] = "发件人姓名", ["sender.phone"] = "发件人电话", ["sender.country"] = "发件国家", @@ -75,6 +80,7 @@ public static class OrderOperationLogSnapshot ["salesProductRulesVersion"] = Norm(order.SalesProductRulesVersion), ["salesProductPriceVersion"] = Norm(order.SalesProductPriceVersion), ["isPaid"] = FormatByte(order.IsPaid), + ["additionalServices"] = FormatAdditionalServices(order.AdditionalServices), }; } @@ -225,6 +231,89 @@ public static class OrderOperationLogSnapshot return path; } + /// + /// 拼装列表「内容」列：备注优先，再跟最多 N 条字段变更。 + /// + public static string BuildSummary( + IReadOnlyList changes, + string? remark, + int maxItems = SummaryMaxChangeItems) + { + var parts = new List(); + if (!string.IsNullOrWhiteSpace(remark)) + { + parts.Add(remark.Trim()); + } + + if (changes.Count > 0) + { + var take = Math.Max(0, maxItems); + foreach (var change in changes.Take(take)) + { + var label = string.IsNullOrWhiteSpace(change.Label) ? change.Path : change.Label; + parts.Add($"{label}：{DisplayValue(change.Path, change.Before)}→{DisplayValue(change.Path, change.After)}"); + } + + if (changes.Count > take) + { + parts.Add($"等{changes.Count}项变更"); + } + } + + return parts.Count == 0 ? "—" : string.Join("；", parts); + } + + public readonly record struct OrderOperationLogChangeView( + string Path, + string Label, + string? Before, + string? After); + + private static string DisplayValue(string path, string? value) + { + if (string.IsNullOrWhiteSpace(value)) + { + return "空"; + } + + if (string.Equals(path, "isPaid", StringComparison.OrdinalIgnoreCase) + || string.Equals(path, "isCod", StringComparison.OrdinalIgnoreCase)) + { + return value == "1" ? "是" : "否"; + } + + if (string.Equals(path, "orderStatus", StringComparison.OrdinalIgnoreCase)) + { + return OmsDocHelper.GetOrderStatusName(value) ?? value; + } + + return value; + } + + private static string? FormatAdditionalServices(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + { + return null; + } + + try + { + var list = JsonConvert.DeserializeObject>(json) ?? []; + var codes = list + .Where(x => !string.IsNullOrWhiteSpace(x)) + .Select(x => x.Trim()) + .Distinct(StringComparer.OrdinalIgnoreCase) + .OrderBy(x => x, StringComparer.OrdinalIgnoreCase) + .ToList(); + return codes.Count == 0 ? null : string.Join("、", codes); + } + catch + { + return Norm(json); + } + } + private static string? Norm(string? value) { if (string.IsNullOrWhiteSpace(value)) diff --git a/src/Oms.Application/Services/Orders/OrderOperationLogWriter.cs b/src/Oms.Application/Services/Orders/OrderOperationLogWriter.cs index d5b3aea..3a6cd6b 100644 --- a/src/Oms.Application/Services/Orders/OrderOperationLogWriter.cs +++ b/src/Oms.Application/Services/Orders/OrderOperationLogWriter.cs @@ -33,23 +33,29 @@ public class OrderOperationLogWriter : IScopedDependency _logger = logger; } - public async Task TryWriteAsync( + public Task TryWriteAsync( Order order, string action, IReadOnlyDictionary before, IReadOnlyDictionary after, string? remark = null) + => TryWriteAsync(order, action, before, after, remark, operatorId: null, operatorName: null); + + public async Task TryWriteAsync( + Order order, + string action, + IReadOnlyDictionary before, + IReadOnlyDictionary after, + string? remark, + int? operatorId, + string? operatorName, + string operatorType = "staff", + string source = "pc") { try { var changes = OrderOperationLogSnapshot.Diff(before, after); - var operatorId = 0; - string? operatorName = null; - if (_currentUser.IsAuthenticated) - { - operatorId = _currentUser.GetId(); - operatorName = await _userDisplayNameResolver.ResolveAsync(operatorId); - } + var (resolvedId, resolvedName) = await ResolveOperatorAsync(operatorId, operatorName); var entity = new OrderOperationLog { @@ -57,10 +63,10 @@ public class OrderOperationLogWriter : IScopedDependency ServiceNo = order.ServiceNo, Action = action, ActionName = OrderOperationActionCodes.GetActionName(action), - OperatorId = operatorId, - OperatorName = operatorName, - OperatorType = "staff", - Source = "pc", + OperatorId = resolvedId, + OperatorName = resolvedName, + OperatorType = string.IsNullOrWhiteSpace(operatorType) ? "staff" : operatorType.Trim(), + Source = string.IsNullOrWhiteSpace(source) ? "pc" : source.Trim(), ChangesJson = JsonConvert.SerializeObject( changes.Select(x => new OrderOperationLogChangeItem { @@ -70,7 +76,7 @@ public class OrderOperationLogWriter : IScopedDependency After = x.After }).ToList()), Remark = string.IsNullOrWhiteSpace(remark) ? null : remark.Trim(), - CreateUserId = operatorId, + CreateUserId = resolvedId, CreateTime = DateTime.Now }; @@ -86,6 +92,28 @@ public class OrderOperationLogWriter : IScopedDependency } } + /// + /// 无字段 diff 时写入叙事日志（打印、下游预报成功等）。 + /// + public Task TryWriteSimpleAsync( + Order order, + string action, + string? remark, + int? operatorId = null, + string? operatorName = null, + string operatorType = "staff", + string source = "pc") + => TryWriteAsync( + order, + action, + new Dictionary(StringComparer.OrdinalIgnoreCase), + new Dictionary(StringComparer.OrdinalIgnoreCase), + remark, + operatorId, + operatorName, + operatorType, + source); + public async Task ListByOrderIdAsync(int orderId) { var rows = await _repository.GetListAsync(x => x.OrderId == orderId); @@ -97,6 +125,30 @@ public class OrderOperationLogWriter : IScopedDependency return new OrderOperationLogListOutput { Items = items }; } + private async Task<(int OperatorId, string? OperatorName)> ResolveOperatorAsync( + int? operatorId, + string? operatorName) + { + if (operatorId is > 0) + { + var name = string.IsNullOrWhiteSpace(operatorName) + ? await _userDisplayNameResolver.ResolveAsync(operatorId.Value) + : operatorName.Trim(); + return (operatorId.Value, name); + } + + if (_currentUser.IsAuthenticated) + { + var id = _currentUser.GetId(); + var name = string.IsNullOrWhiteSpace(operatorName) + ? await _userDisplayNameResolver.ResolveAsync(id) + : operatorName.Trim(); + return (id, name); + } + + return (0, string.IsNullOrWhiteSpace(operatorName) ? null : operatorName.Trim()); + } + private static OrderOperationLogOutput MapOutput(OrderOperationLog row) { var changes = new List(); @@ -113,6 +165,14 @@ public class OrderOperationLogWriter : IScopedDependency } } + var views = changes + .Select(x => new OrderOperationLogSnapshot.OrderOperationLogChangeView( + x.Path, + x.Label, + x.Before, + x.After)) + .ToList(); + return new OrderOperationLogOutput { Id = row.Id, @@ -126,6 +186,7 @@ public class OrderOperationLogWriter : IScopedDependency Source = row.Source, Changes = changes, Remark = row.Remark, + Summary = OrderOperationLogSnapshot.BuildSummary(views, row.Remark), CreateTime = row.CreateTime }; } diff --git a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs index 9049e59..0a98566 100644 --- a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs +++ b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs @@ -1,3 +1,4 @@ +using System.Globalization; using Fake.Application; using Microsoft.AspNetCore.Mvc; using Newtonsoft.Json; @@ -323,7 +324,7 @@ public partial class OrderService "规则"); if (!string.Equals(order.SalesProductRulesVersion, resolved, StringComparison.Ordinal)) { - remarkParts.Add($"rules:{order.SalesProductRulesVersion ?? "-"}→{resolved}"); + remarkParts.Add($"规则版本 {order.SalesProductRulesVersion ?? "-"}→{resolved}"); order.SalesProductRulesVersion = resolved; } } @@ -337,7 +338,7 @@ public partial class OrderService "价格"); if (!string.Equals(order.SalesProductPriceVersion, resolved, StringComparison.Ordinal)) { - remarkParts.Add($"price:{order.SalesProductPriceVersion ?? "-"}→{resolved}"); + remarkParts.Add($"价格版本 {order.SalesProductPriceVersion ?? "-"}→{resolved}"); order.SalesProductPriceVersion = resolved; } } @@ -354,11 +355,13 @@ public partial class OrderService { EnsureReceivableMutable(order); await RecalculateReceivableCoreAsync(order); + remarkParts.Add("并重算应收"); } if (input.RecalculatePayable) { await RecalculatePayableCoreAsync(order); + remarkParts.Add("并重算应付"); } order = await GetOrderOrThrowAsync(order.Id); @@ -368,7 +371,7 @@ public partial class OrderService OrderOperationActionCodes.BindSnapshotVersions, beforeSnap, afterSnap, - remarkParts.Count == 0 ? null : string.Join(';', remarkParts)); + remarkParts.Count == 0 ? null : string.Join('；', remarkParts)); return (await MapOrderOutputsAsync([order])).First(); } @@ -785,6 +788,7 @@ public partial class OrderService { var order = await GetOrderOrThrowAsync(input.Id); EnsureReceivableMutable(order); + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true); if (string.IsNullOrWhiteSpace(order.ProductCode)) { @@ -803,6 +807,15 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); + + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.UpdateAdditionalServices, + beforeSnap, + afterSnap); + return (await MapOrderOutputsAsync([order])).First(); } @@ -892,6 +905,8 @@ public partial class OrderService throw new BusinessException("计费阶段无效"); } + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true); + var fee = new OrderFee { Type = type, @@ -912,6 +927,17 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true); + var typeName = type == OrderFeeTypeCodes.Receivable ? "应收" : "应付"; + var amountText = input.Amount.ToString("0.####", CultureInfo.InvariantCulture); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.AddFee, + beforeSnap, + afterSnap, + $"新增{typeName}「{costItem}」{currency} {amountText}"); + return (await MapOrderOutputsAsync([order])).First(); } diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index 8dc9c39..fbd528a 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -2623,6 +2623,8 @@ public partial class OrderService return; } + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true); + int? couponInstanceId = order.CouponInstanceId; if (!string.IsNullOrWhiteSpace(payloadJson)) { @@ -2765,6 +2767,18 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.PreDeclare, + beforeSnap, + afterSnap, + "预报计费/排货完成", + operatorUserId > 0 ? operatorUserId : null, + operatorName: null, + operatorType: "staff", + source: "job"); + var customer = order.CustomerId.HasValue ? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) : null; diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 35a155d..5e90860 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -1,3 +1,4 @@ +using System.Globalization; using Fake.Application; using Fake.Domain.Exceptions; using Fake.ObjectStorage; @@ -147,6 +148,7 @@ public partial class OrderService : ApplicationService throw new BusinessException("订单未关联客户，无法支付"); } + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order); var fees = await OrderRepository.GetFeesByOrderIdsAsync([order.Id]); var payAmount = order.PayableAmount ?? fees .Where(x => x.Type == OrderFeeTypeCodes.Receivable) @@ -189,6 +191,16 @@ public partial class OrderService : ApplicationService order.IsPaid = 1; order.OrderStatus = OrderStatusCodes.PendingShip; await UpdateOrderAsync(order); + + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order); + var amountText = payAmount.ToString("0.####", CultureInfo.InvariantCulture); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.Pay, + beforeSnap, + afterSnap, + $"支付金额 {amountText}"); } /// @@ -346,6 +358,10 @@ public partial class OrderService : ApplicationService var existingUrl = await OrderFileManager.GetSystemLabelUrlAsync(order.Id); if (!string.IsNullOrWhiteSpace(existingUrl)) { + await OperationLogWriter.TryWriteSimpleAsync( + order, + OrderOperationActionCodes.Print, + "打印面单（使用已有文件）"); return new OrderPrintOutput { PrintUrl = existingUrl!, @@ -365,6 +381,10 @@ public partial class OrderService : ApplicationService { // 兼容回落：仍无模板时给 HTML stub，避免彻底不可打 var fallbackUrl = await CreateOrderPrintFileAsync(mapped); + await OperationLogWriter.TryWriteSimpleAsync( + order, + OrderOperationActionCodes.Print, + "打印面单（回落 HTML）"); return new OrderPrintOutput { PrintUrl = fallbackUrl, @@ -373,6 +393,10 @@ public partial class OrderService : ApplicationService }; } + await OperationLogWriter.TryWriteSimpleAsync( + order, + OrderOperationActionCodes.Print, + "打印面单"); return new OrderPrintOutput { PrintUrl = result.LabelUrl!, @@ -420,6 +444,21 @@ public partial class OrderService : ApplicationService PrintUrl = url, LabelGenerateStatus = LabelGenerateStatusCodes.Ready }); + try + { + var existingOrder = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); + if (existingOrder != null) + { + await OperationLogWriter.TryWriteSimpleAsync( + existingOrder, + OrderOperationActionCodes.Print, + "批量打印（使用已有文件）"); + } + } + catch + { + // 日志失败不影响打印 + } } else { diff --git a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs index 17d6f8b..d6c482b 100644 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs +++ b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs @@ -8,6 +8,12 @@ public static class OrderOperationActionCodes public const string RecalcPayable = "recalcPayable"; public const string Reallocate = "reallocate"; public const string BindSnapshotVersions = "bindSnapshotVersions"; + public const string UpdateAdditionalServices = "updateAdditionalServices"; + public const string AddFee = "addFee"; + public const string Pay = "pay"; + public const string PreDeclare = "preDeclare"; + public const string DownstreamForward = "downstreamForward"; + public const string Print = "print"; public static string GetActionName(string action) => action switch { @@ -17,6 +23,12 @@ public static class OrderOperationActionCodes RecalcPayable => "重算应付", Reallocate => "重新排货", BindSnapshotVersions => "切换价格/规则版本", + UpdateAdditionalServices => "更新附加服务", + AddFee => "添加费用", + Pay => "支付", + PreDeclare => "重新预报", + DownstreamForward => "预报下游", + Print => "打印面单", _ => action }; } -- 2.45.2 From 8df0f371655d79d267d3c4f4c1fee7c8d42cd205 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Sat, 10 Oct 2026 16:27:49 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20=E8=AE=A2=E5=8D=95=E7=BC=96?= =?UTF-8?q?=E8=BE=91=E6=94=AF=E6=8C=81=E6=9C=8D=E5=8A=A1=E5=95=86=E5=8D=95?= =?UTF-8?q?=E5=8F=B7=E5=B9=B6=E5=AE=8C=E5=96=84=E6=93=8D=E4=BD=9C=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 开放 DownstreamServiceNo 更新，补充包裹计费相关操作码与字段快照，便于运单详情编辑留痕。 Co-authored-by: Cursor --- .../Dtos/Orders/OrderUpdateInput.cs | 6 ++++ .../Orders/OrderOperationLogSnapshot.cs | 32 +++++++++++++++++++ .../Services/Orders/OrderService.OrderEdit.cs | 11 ++++++- .../Orders/OrderService.ParcelBilling.cs | 31 ++++++++++++++++++ .../Consts/OrderOperationActionCodes.cs | 4 +++ 5 files changed, 83 insertions(+), 1 deletion(-) diff --git a/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs b/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs index d9bc676..58cff86 100644 --- a/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs +++ b/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs @@ -25,6 +25,12 @@ public class OrderUpdateInput [StringLength(64)] public string? TrackNo { get; set; } + /// + /// 服务商单号。 + /// + [StringLength(64)] + public string? DownstreamServiceNo { get; set; } + /// /// 渠道编码。 /// diff --git a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs index f37c5d4..e25b977 100644 --- a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs +++ b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs @@ -37,21 +37,37 @@ public static class OrderOperationLogSnapshot ["isPaid"] = "是否已支付", ["additionalServices"] = "附加服务", ["sender.name"] = "发件人姓名", + ["sender.company"] = "发件公司", ["sender.phone"] = "发件人电话", + ["sender.telePhone"] = "发件人固话", + ["sender.email"] = "发件人邮箱", ["sender.country"] = "发件国家", ["sender.province"] = "发件省/州", ["sender.city"] = "发件城市", ["sender.zipCode"] = "发件邮编", ["sender.address1"] = "发件地址1", ["sender.address2"] = "发件地址2", + ["sender.address3"] = "发件地址3", + ["sender.houseNumber"] = "发件门牌号", ["receiver.name"] = "收件人姓名", + ["receiver.company"] = "收件公司", ["receiver.phone"] = "收件人电话", + ["receiver.telePhone"] = "收件人固话", + ["receiver.email"] = "收件人邮箱", ["receiver.country"] = "收件国家", ["receiver.province"] = "收件省/州", ["receiver.city"] = "收件城市", ["receiver.zipCode"] = "收件邮编", + ["receiver.streetNumber"] = "收件街道号", ["receiver.address1"] = "收件地址1", ["receiver.address2"] = "收件地址2", + ["receiver.address3"] = "收件地址3", + ["receiver.taxType"] = "收件税号类型", + ["receiver.taxNo"] = "收件税号", + ["receiver.certificateType"] = "收件证件类型", + ["receiver.certificateNo"] = "收件证件号", + ["receiver.eori"] = "收件EORI", + ["receiver.vat"] = "收件VAT", ["fee.receivableTotal"] = "应收合计", ["fee.payableTotal"] = "应付合计", }; @@ -94,13 +110,18 @@ public static class OrderOperationLogSnapshot return new Dictionary(StringComparer.OrdinalIgnoreCase) { ["sender.name"] = Norm(sender.SenderName), + ["sender.company"] = Norm(sender.Company), ["sender.phone"] = Norm(sender.Phone), + ["sender.telePhone"] = Norm(sender.TelePhone), + ["sender.email"] = Norm(sender.Email), ["sender.country"] = FormatInt(sender.Country), ["sender.province"] = Norm(sender.Province), ["sender.city"] = Norm(sender.City), ["sender.zipCode"] = Norm(sender.ZipCode), ["sender.address1"] = Norm(sender.StreetAddress1), ["sender.address2"] = Norm(sender.StreetAddress2), + ["sender.address3"] = Norm(sender.StreetAddress3), + ["sender.houseNumber"] = Norm(sender.HouseNumber), }; } @@ -114,13 +135,24 @@ public static class OrderOperationLogSnapshot return new Dictionary(StringComparer.OrdinalIgnoreCase) { ["receiver.name"] = Norm(recipient.RecipientsName), + ["receiver.company"] = Norm(recipient.CompanyName), ["receiver.phone"] = Norm(recipient.Phone), + ["receiver.telePhone"] = Norm(recipient.TelePhone), + ["receiver.email"] = Norm(recipient.Email), ["receiver.country"] = FormatInt(recipient.Country), ["receiver.province"] = Norm(recipient.Province), ["receiver.city"] = Norm(recipient.City), ["receiver.zipCode"] = Norm(recipient.ZipCode), + ["receiver.streetNumber"] = Norm(recipient.StreetNumber), ["receiver.address1"] = Norm(recipient.StreetAddress1), ["receiver.address2"] = Norm(recipient.StreetAddress2), + ["receiver.address3"] = Norm(recipient.StreetAddress3), + ["receiver.taxType"] = FormatInt(recipient.TaxType), + ["receiver.taxNo"] = Norm(recipient.TaxNo), + ["receiver.certificateType"] = FormatInt(recipient.CertificateType), + ["receiver.certificateNo"] = Norm(recipient.CertificateNo), + ["receiver.eori"] = Norm(recipient.Eori), + ["receiver.vat"] = Norm(recipient.Vat), }; } diff --git a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs index 0a98566..076e3f3 100644 --- a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs +++ b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs @@ -53,6 +53,13 @@ public partial class OrderService order.TrackNo = string.IsNullOrWhiteSpace(input.TrackNo) ? null : input.TrackNo.Trim(); } + if (input.DownstreamServiceNo != null) + { + order.DownstreamServiceNo = string.IsNullOrWhiteSpace(input.DownstreamServiceNo) + ? null + : input.DownstreamServiceNo.Trim(); + } + if (input.ChannelCode != null) { order.ChannelCode = string.IsNullOrWhiteSpace(input.ChannelCode) ? null : input.ChannelCode.Trim(); @@ -138,7 +145,9 @@ public partial class OrderService order = await GetOrderOrThrowAsync(order.Id); var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeParties: true); - var remark = input.Items != null ? "申报信息已更新" : null; + // 前端保存基本信息时总会带 items；仅在无字段级 diff 时用备注兜底，避免掩盖真实字段变更 + var fieldChanges = OrderOperationLogSnapshot.Diff(beforeSnap, afterSnap); + var remark = input.Items != null && fieldChanges.Count == 0 ? "申报信息已更新" : null; await OperationLogWriter.TryWriteAsync( order, OrderOperationActionCodes.EditBasic, diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index fbd528a..15e2e53 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -2382,6 +2382,7 @@ public partial class OrderService try { + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) .OrderBy(x => x.Id) .ToList(); @@ -2419,6 +2420,16 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); + // 操作日志同步落库（件重尺已写完）；计费/排货仍走 MQ，不把审计塞进队列 + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.Inbound, + beforeSnap, + afterSnap, + FormatMeasuredDimensionsRemark("收货", length!.Value, width!.Value, height!.Value, actualWeight!.Value)); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.InboundTrackCode); await EnqueueInboundBillingAsync(order); } @@ -3016,6 +3027,7 @@ public partial class OrderService EnsureBillingProductActive(product); + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) .OrderBy(x => x.Id) .ToList(); @@ -3038,9 +3050,28 @@ public partial class OrderService order.UpdateTime = now; await OrderRepository.UpdateAsync(order); + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.Outbound, + beforeSnap, + afterSnap, + FormatMeasuredDimensionsRemark("发货", length!.Value, width!.Value, height!.Value, actualWeight!.Value)); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.OutboundTrackCode); } + private static string FormatMeasuredDimensionsRemark( + string stage, + decimal length, + decimal width, + decimal height, + decimal actualWeight) + { + return $"{stage} 长×宽×高×重：{length:0.####}×{width:0.####}×{height:0.####}×{actualWeight:0.####}"; + } + private async Task BuildParcelCreateOutputAsync(Order order) { // 预报已成功时，映射/面单/本地存储失败不得回滚订单（常见：容器 /app/wwwroot 不可写）。 diff --git a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs index d6c482b..104d465 100644 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs +++ b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs @@ -14,6 +14,8 @@ public static class OrderOperationActionCodes public const string PreDeclare = "preDeclare"; public const string DownstreamForward = "downstreamForward"; public const string Print = "print"; + public const string Inbound = "inbound"; + public const string Outbound = "outbound"; public static string GetActionName(string action) => action switch { @@ -29,6 +31,8 @@ public static class OrderOperationActionCodes PreDeclare => "重新预报", DownstreamForward => "预报下游", Print => "打印面单", + Inbound => "入库", + Outbound => "出库", _ => action }; } -- 2.45.2 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 4/4] 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