diff --git a/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs b/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs index 84a3c24..d73df52 100644 --- a/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs +++ b/src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs @@ -1,5 +1,7 @@ +using Fake.EventBus.Distributed; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Newtonsoft.Json; using Oms.Application.Messaging.OrderSideEffects; using Oms.Application.Messaging.ParcelCreate; using Oms.Application.Services.Orders.Batch; @@ -25,8 +27,15 @@ internal sealed class SmallBagTrackOrderNotifier( return; } - var publisher = scope.ServiceProvider.GetRequiredService(); - await publisher.PublishTrackUpsertAsync(order, request, customerId); + var eventBus = scope.ServiceProvider.GetRequiredService(); + await eventBus.PublishAsync(new OrderParcelTrackUpsertIntegrationEvent + { + OrderId = order.Id, + CustomerId = customerId, + PayloadJson = JsonConvert.SerializeObject(request), + AppendInitialTrackOnSuccess = request.AppendInitialTrackOnSuccess, + Authorization = request.Authorization + }); } catch (Exception ex) { @@ -75,8 +84,19 @@ internal sealed class SmallBagTrackOrderNotifier( try { using var scope = scopeFactory.CreateScope(); - var publisher = scope.ServiceProvider.GetRequiredService(); - await publisher.PublishLifecycleTrackAppendAsync(orderId, orderNo, trackCode, authorization); + if (orderId <= 0 || string.IsNullOrWhiteSpace(orderNo) || string.IsNullOrWhiteSpace(trackCode)) + { + return; + } + + var eventBus = scope.ServiceProvider.GetRequiredService(); + await eventBus.PublishAsync(new OrderLifecycleTrackAppendIntegrationEvent + { + OrderId = orderId, + OrderNo = orderNo.Trim(), + TrackCode = trackCode.Trim(), + Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim() + }); } catch (Exception ex) { diff --git a/src/Oms.Application/Messaging/IOrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/IOrderBatchJobPublisher.cs deleted file mode 100644 index 40054e3..0000000 --- a/src/Oms.Application/Messaging/IOrderBatchJobPublisher.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace Oms.Application.Messaging; - -public interface IOrderBatchJobPublisher -{ - Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default); - - bool IsEnabled { get; } -} diff --git a/src/Oms.Application/Messaging/OmsAfterCommit.cs b/src/Oms.Application/Messaging/OmsAfterCommit.cs deleted file mode 100644 index 9b31ee7..0000000 --- a/src/Oms.Application/Messaging/OmsAfterCommit.cs +++ /dev/null @@ -1,57 +0,0 @@ -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Fake.UnitOfWork; - -namespace Oms.Application.Messaging; - -/// -/// 把创单/预报之后的副作用延后到当前事务提交成功再执行，每一步单独提交。 -/// -internal static class OmsAfterCommit -{ - public static Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func action) - { - var current = unitOfWorkManager.Current; - if (current != null && !current.IsCompleted) - { - // OnCompleted 仍处在创单事务里，此时再写 Outbox 征用不到连接，事件会丢。 - // 等工作单元释放后再开独立事务。 - EventHandler handler = null!; - handler = (_, _) => - { - current.Disposed -= handler; - _ = Task.Run(action); - }; - current.Disposed += handler; - return Task.CompletedTask; - } - - return action(); - } - - /// - /// 新开独立事务执行一步。失败只记日志，不抛回调用方。 - /// - public static async Task RunStepAsync( - IServiceScopeFactory scopeFactory, - ILogger logger, - string step, - Func work) - { - try - { - using var scope = scopeFactory.CreateScope(); - var unitOfWorkManager = scope.ServiceProvider.GetRequiredService(); - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager); - await work(scope.ServiceProvider); - await unitOfWork.CompleteAsync(); - return true; - } - catch (Exception ex) - { - logger.LogError(ex, "订单后置步骤失败 Step={Step}", step); - return false; - } - } -} diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs deleted file mode 100644 index b572be4..0000000 --- a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs +++ /dev/null @@ -1,119 +0,0 @@ -using System.Data; -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 投递）。 -/// SqlSugar 与 EF 默认各持连接；若环境事务已由 SqlSugar 占用，则改用 RequiresNew 短事务只写 Outbox。 -/// -internal static class OmsIntegrationEventPublisher -{ - public static async Task PublishAsync( - IDistributedEventBus eventBus, - IUnitOfWorkManager unitOfWorkManager, - ILogger logger, - IntegrationEvent @event, - CancellationToken cancellationToken = default) - { - if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) - { - try - { - await EnsureOutboxDbContextAsync(unitOfWorkManager); - await eventBus.PublishAsync(@event, cancellationToken); - logger.LogDebug( - "Outbox enqueued {EventName} Id={EventId} (ambient TX)", - @event.GetType().Name, - @event.Id); - return; - } - catch (Exception ex) when (ShouldUseIndependentOutbox(ex)) - { - // 环境事务已提交、连接对不上，或还没有可征用的数据库事务时，改走独立短事务。 - logger.LogWarning( - ex, - "Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}", - @event.GetType().Name, - @event.Id); - } - } - else - { - logger.LogWarning( - "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX", - @event.GetType().Name, - @event.Id); - } - - using var uow = unitOfWorkManager.Begin(new UnitOfWorkAttribute - { - IsTransactional = true, - RequiresNew = true, - IsolationLevel = IsolationLevel.ReadCommitted - }); - await EnsureOutboxDbContextAsync(unitOfWorkManager); - 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; - } - } - - /// - /// 业务写库前先打开 Outbox DbContext，尽量让 EF 先占住环境事务连接，供 SqlSugar 征用。 - /// - public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager) - { - var current = unitOfWorkManager.Current - ?? throw new InvalidOperationException("No ambient unit of work"); - var dbProvider = current.ServiceProvider - .GetRequiredService>(); - return dbProvider.GetDbContextAsync(); - } - - private static bool ShouldUseIndependentOutbox(Exception ex) - { - for (var e = ex; e != null; e = e.InnerException!) - { - var message = e.Message; - if (message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase) - || message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase) - || message.Contains("可征用的数据库事务", StringComparison.Ordinal) - || message.Contains("returned null", StringComparison.OrdinalIgnoreCase) - || message.Contains("No ambient transaction", StringComparison.OrdinalIgnoreCase) - || message.Contains("已经回滚", StringComparison.Ordinal) - || message.Contains("已被调用", StringComparison.Ordinal)) - { - return true; - } - } - - return false; - } -} diff --git a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs deleted file mode 100644 index a007ac4..0000000 --- a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs +++ /dev/null @@ -1,36 +0,0 @@ -using Fake.EventBus.Distributed; -using Fake.UnitOfWork; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; - -namespace Oms.Application.Messaging; - -/// -/// 通过 Fake 分布式事件总线（Outbox）发布订单批量任务。 -/// -public sealed class OrderBatchJobPublisher( - IDistributedEventBus eventBus, - IUnitOfWorkManager unitOfWorkManager, - IOptions options, - ILogger logger) : IOrderBatchJobPublisher -{ - public bool IsEnabled => options.Value.Enabled; - - public async Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default) - { - if (!options.Value.Enabled) - { - throw new InvalidOperationException("Order batch RabbitMQ is disabled."); - } - - var @event = OrderBatchJobIntegrationEvent.FromMessage(message); - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, cancellationToken); - - logger.LogDebug( - "Outbox enqueued order batch event {EventId} messageId={MessageId} action={Action}", - @event.Id, - message.MessageId, - message.Action); - } -} diff --git a/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs deleted file mode 100644 index 9cba9f9..0000000 --- a/src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs +++ /dev/null @@ -1,22 +0,0 @@ -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/OrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs deleted file mode 100644 index 9322fc4..0000000 --- a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs +++ /dev/null @@ -1,61 +0,0 @@ -using Fake.EventBus.Distributed; -using Fake.UnitOfWork; -using Microsoft.Extensions.Logging; -using Oms.Application.Services.Orders.Issues; - -namespace Oms.Application.Messaging.OrderSideEffects; - -/// -/// 订单旁路副作用事件：事务内写 Outbox，Commit 后投递 RabbitMQ。 -/// -public sealed class OrderSideEffectEventPublisher( - IDistributedEventBus eventBus, - IUnitOfWorkManager unitOfWorkManager, - 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() - }; - - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, 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 - }; - - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, cancellationToken); - } -} diff --git a/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs b/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs deleted file mode 100644 index 2545a2e..0000000 --- a/src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs +++ /dev/null @@ -1,46 +0,0 @@ -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 deleted file mode 100644 index cf35e80..0000000 --- a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs +++ /dev/null @@ -1,143 +0,0 @@ -using Fake.EventBus.Distributed; -using Fake.UnitOfWork; -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; - -/// -/// 创单后置事件发布器：事务内写 Outbox（Fake.EventBus.IntegrationEventLog），Commit 后投递 RabbitMQ。 -/// 揽收 Publish 失败时由调用方同步 ExecutePickupAttachAsync 回落。 -/// -public sealed class OrderParcelCreateEventPublisher( - IDistributedEventBus eventBus, - IUnitOfWorkManager unitOfWorkManager, - 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 - }; - - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, cancellationToken); - } - - public async Task PublishLabelGenerateAsync( - Order order, - int operatorUserId = 0, - CancellationToken cancellationToken = default) - { - var @event = new OrderParcelLabelGenerateIntegrationEvent - { - OrderId = order.Id, - OperatorUserId = operatorUserId - }; - - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, cancellationToken); - } - - public async Task PublishCustoms9610UpsertAsync( - Order order, - string? platformCode = null, - CancellationToken cancellationToken = default) - { - var @event = new OrderParcelCustoms9610UpsertIntegrationEvent - { - OrderId = order.Id, - PlatformCode = platformCode - }; - - await OmsIntegrationEventPublisher.PublishAsync( - eventBus, unitOfWorkManager, logger, @event, cancellationToken); - } - - public async Task PublishPickupAttachAsync( - OrderParcelCreateInput input, - int orderId, - CancellationToken cancellationToken = default) - { - var @event = BuildPickupEvent(input, orderId); - if (@event == null) - { - return false; - } - - return await OmsIntegrationEventPublisher.TryPublishAsync( - eventBus, unitOfWorkManager, logger, @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 - }; - } -} diff --git a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs index 8131be4..6772161 100644 --- a/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs +++ b/src/Oms.Application/OrderBatchMessagingServiceExtensions.cs @@ -1,8 +1,6 @@ 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,10 +9,7 @@ public static class OrderBatchMessagingServiceExtensions public static IServiceCollection AddOmsOrderBatchMessaging(this IServiceCollection services, IConfiguration configuration) { 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/Orders/Batch/OrderBatchActionExecutor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs index 1c0ef56..4008592 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs @@ -1,5 +1,6 @@ using Fake.Application; using Fake.DependencyInjection; +using Fake.EventBus.Distributed; using Fake.ObjectStorage; using Newtonsoft.Json; using Oms.Application.Dtos.Orders; @@ -24,7 +25,7 @@ public sealed class OrderBatchActionExecutor( IObjectStorage fileStorage, ISmallBagTrackOrderClient smallBagTrackOrderClient, ISmallBagTrackUpsertRequestBuilder smallBagTrackUpsertRequestBuilder, - IOrderParcelCreateEventPublisher parcelCreateEventPublisher, + IDistributedEventBus eventBus, OrderDownstreamCredentialResolver downstreamCredentialResolver, IItfxForwardClient itfxForwardClient, OrderFileManager orderFileManager, @@ -252,11 +253,14 @@ public sealed class OrderBatchActionExecutor( trackRequest.Authorization = authorization.Trim(); } - await parcelCreateEventPublisher.PublishTrackUpsertAsync( - order, - trackRequest, - customerId: order.CustomerId, - cancellationToken); + await eventBus.PublishAsync(new OrderParcelTrackUpsertIntegrationEvent + { + OrderId = order.Id, + CustomerId = order.CustomerId, + PayloadJson = JsonConvert.SerializeObject(trackRequest), + AppendInitialTrackOnSuccess = trackRequest.AppendInitialTrackOnSuccess, + Authorization = trackRequest.Authorization + }, cancellationToken); } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs index 3cac21b..493334e 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs @@ -1,6 +1,6 @@ using Fake.Application; using Fake.DependencyInjection; -using Fake.UnitOfWork; +using Fake.EventBus.Distributed; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; @@ -29,9 +29,8 @@ public sealed class OrderBatchJobCoordinator( ICountryRepository countryRepository, IServiceChannelRepository serviceChannelRepository, IObjectStorage fileStorage, - IOrderBatchJobPublisher publisher, + IDistributedEventBus eventBus, OrderBatchJobProcessor batchJobProcessor, - IUnitOfWorkManager unitOfWorkManager, IOptions batchOptions, ILogger logger) : IScopedDependency { @@ -44,7 +43,7 @@ public sealed class OrderBatchJobCoordinator( string? authorization = null, CancellationToken cancellationToken = default) { - if (!publisher.IsEnabled) + if (!batchOptions.Value.Enabled) { throw new BusinessException("订单批量队列未启用，请检查 RabbitMQ 配置"); } @@ -78,12 +77,6 @@ public sealed class OrderBatchJobCoordinator( } } - // 先打开 EF Outbox 上下文，尽量让环境事务由 EF 占连接，避免 SqlSugar 先开事务后 Outbox 征用失败。 - if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) - { - await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager); - } - var now = DateTime.Now; var job = new OrderBatchJob { @@ -161,7 +154,7 @@ public sealed class OrderBatchJobCoordinator( MessageId = item.MessageId ?? $"{job.Id}:{item.OrderId}:{action}", }; - await publisher.PublishAsync(message, cancellationToken); + await eventBus.PublishAsync(OrderBatchJobIntegrationEvent.FromMessage(message), cancellationToken); } logger.LogInformation( @@ -181,7 +174,7 @@ public sealed class OrderBatchJobCoordinator( int operatorUserId, CancellationToken cancellationToken = default) { - if (!publisher.IsEnabled) + if (!batchOptions.Value.Enabled) { throw new BusinessException("订单批量队列未启用，请检查 RabbitMQ 配置"); } @@ -194,8 +187,9 @@ public sealed class OrderBatchJobCoordinator( for (var index = 0; index < items.Count; index++) { - await publisher.PublishAsync( - BuildPayloadMessage(job, items[index], itemsInput[index], input.CustomerId, operatorUserId), + await eventBus.PublishAsync( + OrderBatchJobIntegrationEvent.FromMessage( + BuildPayloadMessage(job, items[index], itemsInput[index], input.CustomerId, operatorUserId)), cancellationToken); } @@ -383,7 +377,7 @@ public sealed class OrderBatchJobCoordinator( public async Task PublishTrackUpsertAsync(SmallBagTrackUpsertRequest request, int orderId, int? customerId) { - if (!publisher.IsEnabled) + if (!batchOptions.Value.Enabled) { logger.LogDebug("RabbitMQ disabled, track upsert will use direct call. orderId={OrderId}", orderId); return false; @@ -401,7 +395,7 @@ public sealed class OrderBatchJobCoordinator( MessageId = $"track:{orderId}:{Guid.NewGuid():N}", }; - await publisher.PublishAsync(message); + await eventBus.PublishAsync(OrderBatchJobIntegrationEvent.FromMessage(message)); return true; } @@ -437,7 +431,7 @@ public sealed class OrderBatchJobCoordinator( int operatorUserId, CancellationToken cancellationToken = default) { - if (!publisher.IsEnabled) + if (!batchOptions.Value.Enabled) { throw new BusinessException("订单批量队列未启用，请检查 RabbitMQ 配置"); } @@ -472,7 +466,9 @@ public sealed class OrderBatchJobCoordinator( item.UpdateUserId = operatorUserId; await batchJobRepository.UpdateItemAsync(item, cancellationToken); - await publisher.PublishAsync(BuildMessageFromJobItem(job, item, operatorUserId), cancellationToken); + await eventBus.PublishAsync( + OrderBatchJobIntegrationEvent.FromMessage(BuildMessageFromJobItem(job, item, operatorUserId)), + cancellationToken); } job.Status = OrderBatchJobStatusCodes.Running; diff --git a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs index 34d536e..fec95be 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -1,4 +1,5 @@ using Fake.DependencyInjection; +using Fake.EventBus.Distributed; using Microsoft.Extensions.Logging; using Oms.Application.Messaging.OrderSideEffects; using Oms.Domain.ChannelsAggregate.Consts; @@ -16,7 +17,7 @@ public class AutoTriggerRuntimeService( OrderAutoTriggerService orderAutoTriggerService, OrderExceptionWorkflowService exceptionWorkflowService, OrderIssueNotificationService issueNotificationService, - IOrderSideEffectEventPublisher sideEffectEventPublisher, + IDistributedEventBus eventBus, ILogger logger) : IScopedDependency { private const string InterceptOperation = "拦截"; @@ -27,7 +28,12 @@ public class AutoTriggerRuntimeService( public Task EnqueueFireAsync(int orderId, AutoTriggerTiming timing, int operatorUserId) => orderId <= 0 ? Task.CompletedTask - : sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId); + : eventBus.PublishAsync(new OrderAutoTriggerFireIntegrationEvent + { + OrderId = orderId, + Timing = timing.ToString(), + OperatorUserId = operatorUserId + }); public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId) { diff --git a/src/Oms.Application/Services/Orders/OrderService.BatchCreate.cs b/src/Oms.Application/Services/Orders/OrderService.BatchCreate.cs index 6649f53..ce3b856 100644 --- a/src/Oms.Application/Services/Orders/OrderService.BatchCreate.cs +++ b/src/Oms.Application/Services/Orders/OrderService.BatchCreate.cs @@ -1,4 +1,6 @@ using Fake.Application; +using Fake.EventBus.Distributed; +using Microsoft.Extensions.Options; using Newtonsoft.Json; using Oms.Application.Dtos.Orders; using Oms.Application.Messaging; @@ -10,8 +12,11 @@ namespace Oms.Application.Services.Orders; public partial class OrderService { - private IOrderBatchJobPublisher BatchJobPublisher => - LazyServiceProvider.GetRequiredService(); + private IDistributedEventBus EventBus => + LazyServiceProvider.GetRequiredService(); + + private bool BatchQueueEnabled => + LazyServiceProvider.GetRequiredService>().Value.Enabled; /// /// 批量 Worker 导入单行 Excel 数据（内部方法，勿暴露为 HTTP API）。 diff --git a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs index 40f81fa..6728828 100644 --- a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs +++ b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs @@ -554,7 +554,7 @@ public partial class OrderService var batchInput = await BuildImportBatchJobInputAsync(input); var operatorUserId = CurrentUser.GetId(); - if (BatchJobPublisher.IsEnabled) + if (BatchQueueEnabled) { var job = await BatchJobCoordinator.CreatePayloadJobAsync(batchInput, operatorUserId); return new OrderUploadOutput { JobId = job.JobId, Async = true }; diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index 56ef739..9199c16 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -41,7 +41,6 @@ public partial class OrderService private const string InboundBillingProcessingReason = "入库后计费/排货处理中"; private const string InboundBillingEnqueueFailReason = "入库计费任务投递失败，请重跑入库计费"; internal const string ForecastBillingProcessingReason = "预报计费/排货处理中"; - private const string ForecastBillingEnqueueFailReason = "预报计费任务投递失败，请重新预报"; private static readonly ConcurrentDictionary InboundBillingInFlight = new(); @@ -100,11 +99,15 @@ public partial class OrderService }; } + /// + /// /order/parcel/create 的实际步骤。调用方已开启事务；这里抛出的异常会回滚订单和已写入的 Outbox。 + /// private async Task CreateParcelCoreAsync( OrderParcelCreateInput input, int? operatorUserId = null, ParcelCreateContext? createContext = null) { + // 1. 幂等：同一 RequestId 直接返回已有订单，不再建单、不再发事件。 if (!string.IsNullOrWhiteSpace(input.RequestId)) { var existed = await OrderRepository.GetByRequestIdAsync(input.RequestId); @@ -118,6 +121,7 @@ public partial class OrderService NormalizeParcelCreateInput(input); createContext ??= BuildParcelCreateContext(input); + // 2. 硬校验。拦截类自动触发在落库前执行，命中即失败，订单不会留下。 var sessionCustomer = await ResolveEffectiveCustomerAsync(input.CustomerId); input.CustomerId = sessionCustomer.Id; @@ -151,112 +155,74 @@ public partial class OrderService country.CountryCode, country.ChineseName); + // 3. 落库订单、件重尺和收发件等子表。此后才有 OrderId，才能写 Outbox。 var serviceNo = await ResolveServiceNoAsync(); var operatorId = operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); - Order? created = null; - try + var order = BuildDraftOrderEntity( + input, + product, + country.CountryCode, + country.ChineseName, + snapshotBinding, + serviceNo, + createContext); + order = await OrderRepository.InsertAsync(order); + + var weights = BuildInitialWeightRows(order.Id, input); + await SaveParcelChildrenAsync(order, input, weights, country); + + // 4. 保存草稿：不做预报计费。9610、上门揽收与订单一起提交。 + if (input.SaveAsDraft) { - var order = BuildDraftOrderEntity( - input, - product, - country.CountryCode, - country.ChineseName, - snapshotBinding, - serviceNo, - createContext); - order = await OrderRepository.InsertAsync(order); - created = order; - - var weights = BuildInitialWeightRows(order.Id, input); - await SaveParcelChildrenAsync(order, input, weights, country); - - // 客户主动「先保存为草稿」：落库后返回。9610 / 揽收在事务提交后各自执行。 - if (input.SaveAsDraft) + await EventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent { - var draftOutput = await BuildParcelCreateOutputAsync(order); - await ScheduleCreateSideEffectsAsync(order.Id, input, draftOutput, includeForecast: false, operatorId); - return draftOutput; - } - - // 偏远标记轻量同步；计费重/应收/排货走 ForecastBilling MQ（失败不回滚建单）。 - order.IsRemote = await IsRemoteHitAsync( - product, - snapshotBinding.Rules, - country.CountryCode, - input.Receiver?.Province, - input.Receiver?.City, - input.Receiver?.ZipCode) ? (byte)1 : (byte)0; - order.OrderStatus = OrderStatusCodes.Draft; - order.PredeclareFailReason = ForecastBillingProcessingReason; - order.AllocationPendingReason = string.Empty; - order.UpdateTime = DateTime.Now; - await OrderRepository.UpdateAsync(order); - - var output = await BuildParcelCreateOutputAsync(order); - await ScheduleCreateSideEffectsAsync(order.Id, input, output, includeForecast: true, operatorId); - return output; - } - catch (BusinessException) - { - await TryRollbackFailedParcelCreateAsync(created); - throw; - } - catch (Exception ex) - { - // 创建失败不落库草稿：回滚已插入订单，由前端让客户选择继续修改或主动保存草稿。 - await TryRollbackFailedParcelCreateAsync(created); - - var cleanReason = StripDraftSavedMarker(ex.Message); - if (string.IsNullOrWhiteSpace(cleanReason)) - { - cleanReason = "预报未完成"; - } - - throw new BusinessException(cleanReason); - } - } - - private async Task TryRollbackFailedParcelCreateAsync(Order? created) - { - if (created == null) - { - return; + OrderId = order.Id, + PlatformCode = input.PlatformCode + }); + var draftOutput = await BuildParcelCreateOutputAsync(order); + await TryAttachPickupAfterParcelCreateAsync(input, draftOutput); + return draftOutput; } - try - { - await RollbackFailedParcelCreateAsync(created); - } - catch - { - // 回滚失败不覆盖原始业务错误；半成品订单需运维清理 - } - } - - /// - /// 创建流程中途失败时删除已落库的半成品订单，避免自动变成草稿。 - /// - private async Task RollbackFailedParcelCreateAsync(Order order) - { - try - { - var writer = LazyServiceProvider.GetRequiredService(); - // 强制清掉可能已写入的 pending 9610（ExportCustomsMode 临时置空触发删除逻辑） - var mode = order.ExportCustomsMode; - order.ExportCustomsMode = null; - await writer.UpsertFromOrderAsync(order); - order.ExportCustomsMode = mode; - } - catch - { - // ignore - } - - order.OrderStatus = OrderStatusCodes.Deleted; - order.PredeclareFailReason = string.Empty; + // 5. 确认下单：偏远标记当场算完；计费重/应收/排货留给预报任务。 + // 状态先停在草稿 +「预报计费/排货处理中」，接口即可返回。 + order.IsRemote = await IsRemoteHitAsync( + product, + snapshotBinding.Rules, + country.CountryCode, + input.Receiver?.Province, + input.Receiver?.City, + input.Receiver?.ZipCode) ? (byte)1 : (byte)0; + order.OrderStatus = OrderStatusCodes.Draft; + order.PredeclareFailReason = ForecastBillingProcessingReason; + order.AllocationPendingReason = string.Empty; order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); - await OrderRepository.DeleteAsync(order); + + // 6. 与订单同一事务写入 Outbox。Commit 后由 Fake 投递；任一条失败则整单回滚。 + // 轨迹不在此列：要等预报消费端排货拿到 ChannelCode 再发。 + var authorization = CaptureRequestAuthorization(); + await EventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent + { + OrderId = order.Id, + PlatformCode = input.PlatformCode + }); + await EnqueueForecastBillingAsync(order, input.CouponInstanceId, operatorId, authorization); + await EventBus.PublishAsync(new OrderParcelLabelGenerateIntegrationEvent + { + OrderId = order.Id, + OperatorUserId = operatorId + }); + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.OrderCreate, operatorId); + + var output = await BuildParcelCreateOutputAsync(order); + output.ForecastBillingQueued = string.Equals( + order.PredeclareFailReason, + ForecastBillingProcessingReason, + StringComparison.Ordinal); + // 揽收失败只回写 PickupCreateError，不让整单回滚。 + await TryAttachPickupAfterParcelCreateAsync(input, output); + return output; } private async Task TryUpsertCustoms9610Async(Order order, string? platformCode = null) @@ -2330,28 +2296,13 @@ public partial class OrderService var operatorId = operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); var authorization = CaptureRequestAuthorization(); - var publisher = LazyServiceProvider.GetRequiredService(); - if (!publisher.IsEnabled) + await EnqueueForecastBillingAsync(order, order.CouponInstanceId, operatorId, authorization); + await EventBus.PublishAsync(new OrderParcelLabelGenerateIntegrationEvent { - var payloadJson = JsonConvert.SerializeObject(new ForecastBillingJobPayload - { - CouponInstanceId = order.CouponInstanceId, - Authorization = authorization - }); - await ProcessForecastBillingForBatchAsync( - order.Id, - payloadJson, - operatorId, - authorization: authorization); - } - else - { - var orderId = order.Id; - var couponInstanceId = order.CouponInstanceId; - await ScheduleForecastEnqueueAsync(orderId, couponInstanceId, operatorId, authorization); - } - - await ScheduleForecastSideEffectsAsync(order.Id, operatorId); + OrderId = order.Id, + OperatorUserId = operatorId + }); + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.OrderCreate, operatorId); // 同步回落失败时给出明确错误；入队成功则保持「处理中」由列表观察结果。 var refreshed = await OrderRepository.FirstOrDefaultAsync(x => x.Id == order.Id); @@ -2846,7 +2797,7 @@ public partial class OrderService // 消费端没有 HttpContext，须用创单时捕获的 Authorization，否则轨迹网关返回 401。 if (allocated && !string.IsNullOrWhiteSpace(order.ChannelCode)) { - await ScheduleTrackAfterAllocationAsync(order.Id, trackAuthorization); + await PublishForecastTrackAsync(order.Id, trackAuthorization); } var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true); @@ -2872,154 +2823,6 @@ public partial class OrderService public string? Authorization { get; set; } } - /// - /// 创单事务提交后，9610、预报计费、揽收、面单、自动触发各自提交。 - /// 轨迹订单等排货成功后再创建。 - /// - private async Task ScheduleCreateSideEffectsAsync( - int orderId, - OrderParcelCreateInput input, - OrderParcelCreateOutput output, - bool includeForecast, - int operatorUserId) - { - var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); - var scopeFactory = LazyServiceProvider.GetRequiredService(); - var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.OrderParcelCreate"); - var platformCode = input.PlatformCode; - var couponInstanceId = input.CouponInstanceId; - // 事务提交后 HttpContext 已不可用，这里先捕获登录态供轨迹网关使用。 - var authorization = CaptureRequestAuthorization(); - - await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () => - { - try - { - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp => - sp.GetRequiredService().PublishCustomsForOrderAsync(orderId, platformCode)); - - var forecastQueued = true; - if (includeForecast) - { - forecastQueued = await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp => - { - var service = sp.GetRequiredService(); - await service.EnqueueForecastBillingByIdAsync( - orderId, - couponInstanceId, - operatorUserId, - authorization); - }); - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => - sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)); - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => - sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)); - } - - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp => - sp.GetRequiredService().TryAttachPickupAfterParcelCreateAsync(input, output)); - - if (includeForecast && !forecastQueued) - { - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp => - sp.GetRequiredService().MarkForecastEnqueueFailedIfStillProcessingAsync(orderId)); - } - } - catch (Exception ex) - { - logger.LogError(ex, "创单后置步骤调度失败 OrderId={OrderId}", orderId); - } - }); - } - - private async Task ScheduleForecastEnqueueAsync( - int orderId, - int? couponInstanceId, - int operatorUserId, - string? authorization) - { - var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); - var scopeFactory = LazyServiceProvider.GetRequiredService(); - var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); - await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => - OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", sp => - sp.GetRequiredService().EnqueueForecastBillingByIdAsync( - orderId, - couponInstanceId, - operatorUserId, - authorization))); - } - - /// - /// 草稿重新预报提交后，面单和下单自动触发各自入队。轨迹等排货成功后再创建。 - /// - private async Task ScheduleForecastSideEffectsAsync(int orderId, int operatorUserId) - { - var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); - var scopeFactory = LazyServiceProvider.GetRequiredService(); - var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); - await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () => - { - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => - sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)); - await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => - sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)); - }); - } - - /// - /// 预报计费事务提交后，再创建轨迹订单（此时已有渠道和 ApiCode）。 - /// - private async Task ScheduleTrackAfterAllocationAsync(int orderId, string? authorization) - { - var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); - var scopeFactory = LazyServiceProvider.GetRequiredService(); - var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); - await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => - OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp => - sp.GetRequiredService().PublishForecastTrackAsync(orderId, authorization))); - } - - private async Task PublishCustomsForOrderAsync(int orderId, string? platformCode) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) - { - return; - } - - await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, platformCode); - } - - private async Task EnqueueForecastBillingByIdAsync( - int orderId, - int? couponInstanceId, - int operatorUserId, - string? authorization) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) - { - return; - } - - await EnqueueForecastBillingAsync(order, couponInstanceId, operatorUserId, authorization); - } - - private async Task MarkForecastEnqueueFailedIfStillProcessingAsync(int orderId) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || !string.Equals(order.PredeclareFailReason, ForecastBillingProcessingReason, StringComparison.Ordinal)) - { - return; - } - - await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason); - } - private async Task PublishForecastTrackAsync(int orderId, string? authorization = null) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); @@ -3036,41 +2839,26 @@ public partial class OrderService var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiverInput: null); request.AppendInitialTrackOnSuccess = true; ApplyTrackAuthorization(request, authorization); - await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id); - } - - private async Task PublishForecastLabelAsync(int orderId, int operatorUserId) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) + await EventBus.PublishAsync(new OrderParcelTrackUpsertIntegrationEvent { - return; - } - - await ParcelCreateEventPublisher.PublishLabelGenerateAsync(order, operatorUserId); - } - - private async Task PublishForecastAutoTriggerAsync(int orderId, int operatorUserId) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) - { - return; - } - - await AutoTriggerRuntimeService.EnqueueFireAsync(orderId, AutoTriggerTiming.OrderCreate, operatorUserId); + OrderId = order.Id, + CustomerId = customer?.Id, + PayloadJson = JsonConvert.SerializeObject(request), + AppendInitialTrackOnSuccess = request.AppendInitialTrackOnSuccess, + Authorization = request.Authorization + }); } + /// + /// 预报计费入队。队列开着写 ForecastBilling Outbox；关着则在当前事务里同步跑同一套计费。 + /// Authorization 放进任务载荷，供排货成功后转发轨迹网关。 + /// private async Task EnqueueForecastBillingAsync( Order order, int? couponInstanceId, int operatorUserId, string? authorization = null) { - var logger = LazyServiceProvider.GetRequiredService() - .CreateLogger("Oms.ForecastBilling"); authorization ??= CaptureRequestAuthorization(); var payloadJson = JsonConvert.SerializeObject(new ForecastBillingJobPayload { @@ -3078,62 +2866,27 @@ public partial class OrderService Authorization = authorization }); - try + if (!BatchQueueEnabled) { - var publisher = LazyServiceProvider.GetRequiredService(); - if (!publisher.IsEnabled) - { - try - { - await ProcessForecastBillingForBatchAsync( - order.Id, - payloadJson, - operatorUserId, - authorization: authorization); - } - 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, - authorization); - } - catch (Exception ex) - { - logger.LogError( - ex, - "预报计费任务投递失败 OrderId={OrderId} ServiceNo={ServiceNo}", + await ProcessForecastBillingForBatchAsync( order.Id, - order.ServiceNo); - try - { - await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason); - } - catch - { - // ignore - } + payloadJson, + operatorUserId, + authorization: authorization); + 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, + authorization); } private async Task EnqueueInboundBillingAsync(Order order) @@ -3142,8 +2895,7 @@ public partial class OrderService .CreateLogger("Oms.InboundBilling"); try { - var publisher = LazyServiceProvider.GetRequiredService(); - if (!publisher.IsEnabled) + if (!BatchQueueEnabled) { // 本地未开队列：同步跑异步计费逻辑，仍不向外抛（入库主流程已完成） try @@ -3397,12 +3149,6 @@ public partial class OrderService }; } - private IOrderParcelCreateEventPublisher ParcelCreateEventPublisher => - LazyServiceProvider.GetRequiredService(); - - private IOrderSideEffectEventPublisher SideEffectEventPublisher => - LazyServiceProvider.GetRequiredService(); - /// /// 小包预报成功后异步同步轨迹订单（POST /ProviderService/Order/UpsertSmallBag）。 /// @@ -3415,7 +3161,14 @@ public partial class OrderService var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiver); request.AppendInitialTrackOnSuccess = appendInitialTrack; ApplyTrackAuthorization(request); - await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id); + await EventBus.PublishAsync(new OrderParcelTrackUpsertIntegrationEvent + { + OrderId = order.Id, + CustomerId = customer?.Id, + PayloadJson = JsonConvert.SerializeObject(request), + AppendInitialTrackOnSuccess = request.AppendInitialTrackOnSuccess, + Authorization = request.Authorization + }); } private void ApplyTrackAuthorization(SmallBagTrackUpsertRequest request, string? authorization = null) @@ -3486,11 +3239,13 @@ public partial class OrderService .Authorization .ToString(); - await SideEffectEventPublisher.PublishLifecycleTrackAppendAsync( - order.Id, - order.ServiceNo, - trackCode, - string.IsNullOrWhiteSpace(authorization) ? null : authorization); + await EventBus.PublishAsync(new OrderLifecycleTrackAppendIntegrationEvent + { + OrderId = order.Id, + OrderNo = order.ServiceNo, + TrackCode = trackCode, + Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization + }); } /// diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 1932297..7022a57 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -603,34 +603,20 @@ public partial class OrderService : ApplicationService } /// - /// 创建订单包裹（PC/H5 确认下单或保存草稿）。 + /// 创建订单包裹（PC/H5 确认下单或保存草稿）。整段在一个事务里，异常直接抛出并回滚。 /// - /// 【同步，本接口事务内】RequestId 幂等 → 校验入参/销售产品 → 黑名单 → - /// 自动触发「拦截」规则（命中直接失败）→ 生成运单号 → 落库订单与子表。 - /// 确认下单：同步写偏远标记，状态草稿 +「预报计费/排货处理中」后返回； - /// 保存草稿：仅落库，不做预报计费。失败会回滚订单，不落草稿。 + /// 【本请求内】RequestId 幂等 → 校验 → 黑名单 → 下单前拦截规则 → 落库订单与子表。 + /// 保存草稿只再写 9610 与上门揽收；确认下单额外标记「预报处理中」，并写入预报计费、面单、自动触发。 /// /// - /// 【创单事务提交后各自投递 MQ（互不等待；失败不回滚订单）】 - /// 1) 9610 订单 Upsert（customs-9610）— 草稿与确认下单都会投； - /// 2) 预报计费/排货任务（forecast-billing 批量任务）— 仅确认下单； - /// 队列关闭时在后台同步跑同一套计费逻辑； - /// 3) 系统面单 PDF（label-generate）— 仅确认下单； - /// 4) 下单自动触发（OrderCreate：警告/问题件等，不含拦截）— 仅确认下单； - /// 5) 上门揽收挂载（pickup-attach）— 仅上门揽收且地址/时间/仓库齐全时； - /// 投递失败则同步回落建揽收单。 - /// - /// - /// 【预报计费消费后】计费重、应收、排货应付、营销快照 → 订单改已预报。 - /// 排货成功且已有 ChannelCode 时，再投递轨迹订单 + A0101（track-upsert）； - /// 轨迹网关需要登录 Authorization，创单时写入预报任务 Payload，消费端原样转发； - /// 排货失败不阻断预报，也不建轨迹（可页面补做）。 + /// 【同一事务的 Outbox，Commit 后才投递】9610、预报计费、面单、OrderCreate 自动触发、上门揽收。 + /// 队列关闭时预报计费在本事务内同步执行。轨迹不在这里发，等预报消费端排货成功后再发。 /// /// /// 入参：订单包裹创建参数。SaveAsDraft=true 为仅保存草稿。 /// - /// 订单包裹创建结果。确认下单时 ForecastBillingQueued 表示预报计费已排队； - /// 上门揽收时 PickupQueued / PickupNo 视投递或同步回落结果而定。 + /// 订单包裹创建结果。确认下单时 ForecastBillingQueued 表示预报计费已入队； + /// 上门揽收入队成功时 PickupQueued=true，失败只写 PickupCreateError，不回滚订单。 /// [HttpPost("/order/parcel/create")] [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] @@ -680,6 +666,9 @@ public partial class OrderService : ApplicationService } } + /// + /// 仅上门揽收。地址、期望时间和仓库齐全才入队；发布失败写入返回值，不抛回创单事务。 + /// private async Task TryAttachPickupAfterParcelCreateAsync( OrderParcelCreateInput input, OrderParcelCreateOutput output) @@ -715,27 +704,20 @@ public partial class OrderService : ApplicationService return; } - var publisher = LazyServiceProvider.GetRequiredService(); try { - // 优先入 Fake.EventBus；发布失败则同步回落建揽收并回写单号。 - var queued = await publisher.PublishPickupAttachAsync(input, output.OrderId); - if (queued) + await EventBus.PublishAsync(new OrderParcelPickupAttachIntegrationEvent { - 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; + OrderId = output.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 + }); + output.PickupQueued = true; } catch (Exception ex) { @@ -762,7 +744,7 @@ public partial class OrderService : ApplicationService throw new BusinessException($"单次批量创建不能超过{ParcelBatchMaxCount}条"); } - if (BatchJobPublisher.IsEnabled) + if (BatchQueueEnabled) { return await BatchCreateParcelViaQueueAsync(input); } diff --git a/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs index da907e3..eeac041 100644 --- a/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs +++ b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs @@ -6,8 +6,6 @@ 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; @@ -18,8 +16,8 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools _dbProvider; - private readonly OrderBatchJobPublisher _publisher; protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) { @@ -29,13 +27,9 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools(); + _eventBus = ServiceProvider.GetRequiredService(); _physicalBus = ServiceProvider.GetRequiredService(); _dbProvider = ServiceProvider.GetRequiredService>(); - _publisher = new OrderBatchJobPublisher( - ServiceProvider.GetRequiredService(), - _uowManager, - Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }), - NullLogger.Instance); } [Fact] @@ -56,7 +50,7 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools(), - _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 index f4d036e..6cda93d 100644 --- a/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs +++ b/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs @@ -6,7 +6,7 @@ using Fake.Testing; using Fake.UnitOfWork; using FluentAssertions; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging.Abstractions; +using Newtonsoft.Json; using Oms.Application.Integrations.SmallBagTrack; using Oms.Application.Messaging; using Oms.Application.Messaging.ParcelCreate; @@ -22,7 +22,6 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools _dbProvider; - private readonly OrderParcelCreateEventPublisher _publisher; protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) { @@ -36,11 +35,6 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools(); _outbox = ServiceProvider.GetRequiredService(); _dbProvider = ServiceProvider.GetRequiredService>(); - _publisher = new OrderParcelCreateEventPublisher( - _distributedEventBus, - _uowManager, - ServiceProvider.GetRequiredService(), - NullLogger.Instance); } [Fact] @@ -58,7 +52,11 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools _distributedEventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent + { + OrderId = 404, + PlatformCode = "shopee" + }); - _physicalBus.Published.Should().ContainSingle() - .Which.Should().BeOfType() - .Which.OrderId.Should().Be(404); + await act.Should().ThrowAsync(); } } diff --git a/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs index 8f3abf8..84fd172 100644 --- a/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs +++ b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs @@ -6,7 +6,6 @@ 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; @@ -20,7 +19,6 @@ public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools _dbProvider; - private readonly OrderSideEffectEventPublisher _publisher; protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) { @@ -33,10 +31,6 @@ public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools(); _physicalBus = ServiceProvider.GetRequiredService(); _dbProvider = ServiceProvider.GetRequiredService>(); - _publisher = new OrderSideEffectEventPublisher( - _distributedEventBus, - _uowManager, - NullLogger.Instance); } [Fact] @@ -47,8 +41,18 @@ public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools _countryRepository = new(MockBehavior.Loose); private readonly Mock _serviceChannelRepository = new(MockBehavior.Loose); private readonly Mock _fileStorage = new(MockBehavior.Loose); - private readonly Mock _publisher = new(MockBehavior.Strict); + private readonly Mock _eventBus = new(MockBehavior.Strict); private readonly OrderBatchJobCoordinator _coordinator; public OrderBatchJobCoordinatorGovernanceTests() { - _publisher.SetupGet(x => x.IsEnabled).Returns(true); - var batchOptions = Options.Create(new OrderBatchRabbitMqOptions()); + var batchOptions = Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }); var actionExecutor = new Mock(MockBehavior.Loose); var processor = new OrderBatchJobProcessor( _batchJobRepository.Object, @@ -37,17 +36,14 @@ public class OrderBatchJobCoordinatorGovernanceTests actionExecutor.Object, batchOptions, NullLogger.Instance); - var unitOfWorkManager = new Mock(MockBehavior.Loose); - unitOfWorkManager.SetupGet(x => x.Current).Returns((IUnitOfWork?)null); _coordinator = new OrderBatchJobCoordinator( _batchJobRepository.Object, _orderRepository.Object, _countryRepository.Object, _serviceChannelRepository.Object, _fileStorage.Object, - _publisher.Object, + _eventBus.Object, processor, - unitOfWorkManager.Object, batchOptions, NullLogger.Instance); } @@ -71,8 +67,8 @@ public class OrderBatchJobCoordinatorGovernanceTests _batchJobRepository .Setup(x => x.UpdateJobAsync(job, It.IsAny())) .Returns(Task.CompletedTask); - _publisher - .Setup(x => x.PublishAsync(It.IsAny(), It.IsAny())) + _eventBus + .Setup(x => x.PublishAsync(It.IsAny(), It.IsAny())) .Returns(Task.CompletedTask); var output = await _coordinator.RetryFailedAsync(job.Id, null, 99); @@ -83,9 +79,9 @@ public class OrderBatchJobCoordinatorGovernanceTests failedItem.RetryCount.Should().Be(0); failedItem.FailReason.Should().BeNull(); job.Status.Should().Be(OrderBatchJobStatusCodes.Running); - _publisher.Verify( + _eventBus.Verify( x => x.PublishAsync( - It.Is(m => m.JobItemId == failedItem.Id), + It.Is(m => m.JobItemId == failedItem.Id), It.IsAny()), Times.Once); }