ideship/admin
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.06 #10

Merged
zzh merged 18 commits from test into master 2026-10-19 17:03:15 +08:00
Conversation 0 Commits 18 Files Changed 80 +256 -998
21 changed files with 256 additions and 998 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d72637e472 - Show all commits

28
src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackOrderNotifier.cs
View File

@ -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<IOrderParcelCreateEventPublisher>();
await publisher.PublishTrackUpsertAsync(order, request, customerId);
var eventBus = scope.ServiceProvider.GetRequiredService<IDistributedEventBus>();
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<IOrderSideEffectEventPublisher>();
await publisher.PublishLifecycleTrackAppendAsync(orderId, orderNo, trackCode, authorization);
if (orderId <= 0 || string.IsNullOrWhiteSpace(orderNo) || string.IsNullOrWhiteSpace(trackCode))
{
return;
}
var eventBus = scope.ServiceProvider.GetRequiredService<IDistributedEventBus>();
await eventBus.PublishAsync(new OrderLifecycleTrackAppendIntegrationEvent
{
OrderId = orderId,
OrderNo = orderNo.Trim(),
TrackCode = trackCode.Trim(),
Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim()
});
}
catch (Exception ex)
{

8
src/Oms.Application/Messaging/IOrderBatchJobPublisher.cs
View File

@ -1,8 +0,0 @@
namespace Oms.Application.Messaging;
public interface IOrderBatchJobPublisher
{
Task PublishAsync(OrderBatchJobMessage message, CancellationToken cancellationToken = default);
bool IsEnabled { get; }
}

57
src/Oms.Application/Messaging/OmsAfterCommit.cs
View File

@ -1,57 +0,0 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Fake.UnitOfWork;
namespace Oms.Application.Messaging;
/// <summary>
/// 把创单/预报之后的副作用延后到当前事务提交成功再执行，每一步单独提交。
/// </summary>
internal static class OmsAfterCommit
{
public static Task ScheduleAsync(IUnitOfWorkManager unitOfWorkManager, Func<Task> action)
{
var current = unitOfWorkManager.Current;
if (current != null && !current.IsCompleted)
{
// OnCompleted 仍处在创单事务里，此时再写 Outbox 征用不到连接，事件会丢。
// 等工作单元释放后再开独立事务。
EventHandler<UnitOfWorkEventArgs> handler = null!;
handler = (_, _) =>
{
current.Disposed -= handler;
_ = Task.Run(action);
};
current.Disposed += handler;
return Task.CompletedTask;
}
return action();
}
/// <summary>
/// 新开独立事务执行一步。失败只记日志，不抛回调用方。
/// </summary>
public static async Task<bool> RunStepAsync(
IServiceScopeFactory scopeFactory,
ILogger logger,
string step,
Func<IServiceProvider, Task> work)
{
try
{
using var scope = scopeFactory.CreateScope();
var unitOfWorkManager = scope.ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
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;
}
}
}

119
src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs
View File

@ -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;
/// <summary>
/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。
/// SqlSugar 与 EF 默认各持连接；若环境事务已由 SqlSugar 占用，则改用 RequiresNew 短事务只写 Outbox。
/// </summary>
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<bool> 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;
}
}
/// <summary>
/// 业务写库前先打开 Outbox DbContext，尽量让 EF 先占住环境事务连接，供 SqlSugar 征用。
/// </summary>
public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager)
{
var current = unitOfWorkManager.Current
?? throw new InvalidOperationException("No ambient unit of work");
var dbProvider = current.ServiceProvider
.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
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;
}
}

36
src/Oms.Application/Messaging/OrderBatchJobPublisher.cs
View File

@ -1,36 +0,0 @@
using Fake.EventBus.Distributed;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace Oms.Application.Messaging;
/// <summary>
/// 通过 Fake 分布式事件总线（Outbox）发布订单批量任务。
/// </summary>
public sealed class OrderBatchJobPublisher(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobPublisher> 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);
}
}

22
src/Oms.Application/Messaging/OrderSideEffects/IOrderSideEffectEventPublisher.cs
View File

@ -1,22 +0,0 @@
using Oms.Application.Services.Orders.Issues;
namespace Oms.Application.Messaging.OrderSideEffects;
/// <summary>
/// 订单侧后置副作用发布（生命周期轨迹、自动触发等）。
/// </summary>
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);
}

61
src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs
View File

@ -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;
/// <summary>
/// 订单旁路副作用事件：事务内写 Outbox，Commit 后投递 RabbitMQ。
/// </summary>
public sealed class OrderSideEffectEventPublisher(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderSideEffectEventPublisher> 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);
}
}

46
src/Oms.Application/Messaging/ParcelCreate/IOrderParcelCreateEventPublisher.cs
View File

@ -1,46 +0,0 @@
using Oms.Application.Dtos.Orders;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置副作用：发布到 Fake.EventBus.RabbitMQ；发布失败时回落同步执行。
/// </summary>
public interface IOrderParcelCreateEventPublisher
{
/// <summary>
/// 轨迹订单 + 创建成功轨迹（A0101）合并为一个事件。
/// </summary>
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);
/// <summary>
/// 发布揽收挂接事件。返回 true 表示已入队；false 表示已同步回落执行（或参数不足未执行）。
/// </summary>
Task<bool> PublishPickupAttachAsync(
OrderParcelCreateInput input,
int orderId,
CancellationToken cancellationToken = default);
/// <summary>
/// 同步执行揽收挂接（用于创单 HTTP 需要回写 PickupNo 的场景）。
/// </summary>
Task<PickupOrderOutput?> ExecutePickupAttachAsync(
OrderParcelCreateInput input,
int orderId,
CancellationToken cancellationToken = default);
}

143
src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs
View File

@ -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;
/// <summary>
/// 创单后置事件发布器：事务内写 Outbox（Fake.EventBus.IntegrationEventLog），Commit 后投递 RabbitMQ。
/// 揽收 Publish 失败时由调用方同步 ExecutePickupAttachAsync 回落。
/// </summary>
public sealed class OrderParcelCreateEventPublisher(
IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
IServiceScopeFactory scopeFactory,
ILogger<OrderParcelCreateEventPublisher> 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<bool> 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<PickupOrderOutput?> 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<PickupOrderService>();
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
};
}
}

5
src/Oms.Application/OrderBatchMessagingServiceExtensions.cs
View File

@ -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<OrderBatchRabbitMqOptions>(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName));
services.AddSingleton<IOrderBatchJobPublisher, OrderBatchJobPublisher>();
services.AddSingleton<IOrderBatchQueueInspector, OrderBatchQueueInspector>();
services.AddSingleton<IOrderParcelCreateEventPublisher, OrderParcelCreateEventPublisher>();
services.AddSingleton<IOrderSideEffectEventPublisher, OrderSideEffectEventPublisher>();
// 消费端由 Fake.EventBus.RabbitMQ（RabbitMqEventBus HostedService）托管
services.AddHostedService<OrderBatchDlxMonitorHostedService>();
return services;

16
src/Oms.Application/Services/Orders/Batch/OrderBatchActionExecutor.cs
View File

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

32
src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
View File

@ -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<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobCoordinator> 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<bool> 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;

10
src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs
View File

@ -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<AutoTriggerRuntimeService> 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)
{

9
src/Oms.Application/Services/Orders/OrderService.BatchCreate.cs
View File

@ -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<IOrderBatchJobPublisher>();
private IDistributedEventBus EventBus =>
LazyServiceProvider.GetRequiredService<IDistributedEventBus>();
private bool BatchQueueEnabled =>
LazyServiceProvider.GetRequiredService<IOptions<OrderBatchRabbitMqOptions>>().Value.Enabled;
/// <summary>
/// 批量 Worker 导入单行 Excel 数据（内部方法，勿暴露为 HTTP API）。

2
src/Oms.Application/Services/Orders/OrderService.Helpers.cs
View File

@ -554,7 +554,7 @@ public partial class OrderService
var batchInput = await BuildImportBatchJobInputAsync(input);
var operatorUserId = CurrentUser.GetId<int>();
if (BatchJobPublisher.IsEnabled)
if (BatchQueueEnabled)
{
var job = await BatchJobCoordinator.CreatePayloadJobAsync(batchInput, operatorUserId);
return new OrderUploadOutput { JobId = job.JobId, Async = true };

479
src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs
View File

@ -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<int, byte> InboundBillingInFlight = new();
@ -100,11 +99,15 @@ public partial class OrderService
};
}
/// <summary>
/// /order/parcel/create 的实际步骤。调用方已开启事务；这里抛出的异常会回滚订单和已写入的 Outbox。
/// </summary>
private async Task<OrderParcelCreateOutput> 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<int>() : 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
{
// 回滚失败不覆盖原始业务错误；半成品订单需运维清理
}
}
/// <summary>
/// 创建流程中途失败时删除已落库的半成品订单，避免自动变成草稿。
/// </summary>
private async Task RollbackFailedParcelCreateAsync(Order order)
{
try
{
var writer = LazyServiceProvider.GetRequiredService<Customs9610OrderWriter>();
// 强制清掉可能已写入的 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<int>() : 0);
var authorization = CaptureRequestAuthorization();
var publisher = LazyServiceProvider.GetRequiredService<IOrderBatchJobPublisher>();
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; }
}
/// <summary>
/// 创单事务提交后，9610、预报计费、揽收、面单、自动触发各自提交。
/// 轨迹订单等排货成功后再创建。
/// </summary>
private async Task ScheduleCreateSideEffectsAsync(
int orderId,
OrderParcelCreateInput input,
OrderParcelCreateOutput output,
bool includeForecast,
int operatorUserId)
{
var unitOfWorkManager = LazyServiceProvider.GetRequiredService<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().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<OrderService>().PublishCustomsForOrderAsync(orderId, platformCode));
var forecastQueued = true;
if (includeForecast)
{
forecastQueued = await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp =>
{
var service = sp.GetRequiredService<OrderService>();
await service.EnqueueForecastBillingByIdAsync(
orderId,
couponInstanceId,
operatorUserId,
authorization);
});
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
sp.GetRequiredService<OrderService>().PublishForecastAutoTriggerAsync(orderId, operatorUserId));
}
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp =>
sp.GetRequiredService<OrderService>().TryAttachPickupAfterParcelCreateAsync(input, output));
if (includeForecast && !forecastQueued)
{
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp =>
sp.GetRequiredService<OrderService>().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<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.ForecastBilling");
await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () =>
OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", sp =>
sp.GetRequiredService<OrderService>().EnqueueForecastBillingByIdAsync(
orderId,
couponInstanceId,
operatorUserId,
authorization)));
}
/// <summary>
/// 草稿重新预报提交后，面单和下单自动触发各自入队。轨迹等排货成功后再创建。
/// </summary>
private async Task ScheduleForecastSideEffectsAsync(int orderId, int operatorUserId)
{
var unitOfWorkManager = LazyServiceProvider.GetRequiredService<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.ForecastBilling");
await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () =>
{
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
sp.GetRequiredService<OrderService>().PublishForecastAutoTriggerAsync(orderId, operatorUserId));
});
}
/// <summary>
/// 预报计费事务提交后，再创建轨迹订单（此时已有渠道和 ApiCode）。
/// </summary>
private async Task ScheduleTrackAfterAllocationAsync(int orderId, string? authorization)
{
var unitOfWorkManager = LazyServiceProvider.GetRequiredService<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.ForecastBilling");
await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () =>
OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp =>
sp.GetRequiredService<OrderService>().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
});
}
/// <summary>
/// 预报计费入队。队列开着写 ForecastBilling Outbox；关着则在当前事务里同步跑同一套计费。
/// Authorization 放进任务载荷，供排货成功后转发轨迹网关。
/// </summary>
private async Task EnqueueForecastBillingAsync(
Order order,
int? couponInstanceId,
int operatorUserId,
string? authorization = null)
{
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>()
.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<IOrderBatchJobPublisher>();
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<IOrderBatchJobPublisher>();
if (!publisher.IsEnabled)
if (!BatchQueueEnabled)
{
// 本地未开队列：同步跑异步计费逻辑，仍不向外抛（入库主流程已完成）
try
@ -3397,12 +3149,6 @@ public partial class OrderService
};
}
private IOrderParcelCreateEventPublisher ParcelCreateEventPublisher =>
LazyServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
private IOrderSideEffectEventPublisher SideEffectEventPublisher =>
LazyServiceProvider.GetRequiredService<IOrderSideEffectEventPublisher>();
/// <summary>
/// 小包预报成功后异步同步轨迹订单（POST /ProviderService/Order/UpsertSmallBag）。
/// </summary>
@ -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
});
}
/// <summary>

62
src/Oms.Application/Services/Orders/OrderService.cs
View File

@ -603,34 +603,20 @@ public partial class OrderService : ApplicationService
}
/// <summary>
/// 创建订单包裹（PC/H5 确认下单或保存草稿）。
/// 创建订单包裹（PC/H5 确认下单或保存草稿）。整段在一个事务里，异常直接抛出并回滚。
/// <para>
/// 【同步，本接口事务内】RequestId 幂等 → 校验入参/销售产品 → 黑名单 →
/// 自动触发「拦截」规则（命中直接失败）→ 生成运单号 → 落库订单与子表。
/// 确认下单：同步写偏远标记，状态草稿 +「预报计费/排货处理中」后返回；
/// 保存草稿：仅落库，不做预报计费。失败会回滚订单，不落草稿。
/// 【本请求内】RequestId 幂等 → 校验 → 黑名单 → 下单前拦截规则 → 落库订单与子表。
/// 保存草稿只再写 9610 与上门揽收；确认下单额外标记「预报处理中」，并写入预报计费、面单、自动触发。
/// </para>
/// <para>
/// 【创单事务提交后各自投递 MQ（互不等待；失败不回滚订单）】
/// 1) 9610 订单 Upsert（customs-9610）— 草稿与确认下单都会投；
/// 2) 预报计费/排货任务（forecast-billing 批量任务）— 仅确认下单；
/// 队列关闭时在后台同步跑同一套计费逻辑；
/// 3) 系统面单 PDF（label-generate）— 仅确认下单；
/// 4) 下单自动触发（OrderCreate：警告/问题件等，不含拦截）— 仅确认下单；
/// 5) 上门揽收挂载（pickup-attach）— 仅上门揽收且地址/时间/仓库齐全时；
/// 投递失败则同步回落建揽收单。
/// </para>
/// <para>
/// 【预报计费消费后】计费重、应收、排货应付、营销快照 → 订单改已预报。
/// 排货成功且已有 ChannelCode 时，再投递轨迹订单 + A0101（track-upsert）；
/// 轨迹网关需要登录 Authorization，创单时写入预报任务 Payload，消费端原样转发；
/// 排货失败不阻断预报，也不建轨迹（可页面补做）。
/// 【同一事务的 Outbox，Commit 后才投递】9610、预报计费、面单、OrderCreate 自动触发、上门揽收。
/// 队列关闭时预报计费在本事务内同步执行。轨迹不在这里发，等预报消费端排货成功后再发。
/// </para>
/// </summary>
/// <param name="input">入参：订单包裹创建参数。SaveAsDraft=true 为仅保存草稿。</param>
/// <returns>
/// 订单包裹创建结果。确认下单时 ForecastBillingQueued 表示预报计费已排队；
/// 上门揽收时 PickupQueued / PickupNo 视投递或同步回落结果而定。
/// 订单包裹创建结果。确认下单时 ForecastBillingQueued 表示预报计费已入队；
/// 上门揽收入队成功时 PickupQueued=true，失败只写 PickupCreateError，不回滚订单。
/// </returns>
[HttpPost("/order/parcel/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
@ -680,6 +666,9 @@ public partial class OrderService : ApplicationService
}
}
/// <summary>
/// 仅上门揽收。地址、期望时间和仓库齐全才入队；发布失败写入返回值，不抛回创单事务。
/// </summary>
private async Task TryAttachPickupAfterParcelCreateAsync(
OrderParcelCreateInput input,
OrderParcelCreateOutput output)
@ -715,27 +704,20 @@ public partial class OrderService : ApplicationService
return;
}
var publisher = LazyServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
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);
}

31
src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs
View File

@ -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<OmsOut
{
private readonly IUnitOfWorkManager _uowManager;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IDistributedEventBus _eventBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderBatchJobPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
@ -29,13 +27,9 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools<OmsOut
public OrderBatchJobOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_eventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderBatchJobPublisher(
ServiceProvider.GetRequiredService<IDistributedEventBus>(),
_uowManager,
Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }),
NullLogger<OrderBatchJobPublisher>.Instance);
}
[Fact]
@ -56,7 +50,7 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools<OmsOut
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAsync(message);
await _eventBus.PublishAsync(OrderBatchJobIntegrationEvent.FromMessage(message));
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
@ -68,23 +62,4 @@ public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools<OmsOut
evt.MessageId.Should().Be(message.MessageId);
}
[Fact]
public async Task 队列未启用时_Publish应抛错()
{
var disabled = new OrderBatchJobPublisher(
ServiceProvider.GetRequiredService<IDistributedEventBus>(),
_uowManager,
Options.Create(new OrderBatchRabbitMqOptions { Enabled = false }),
NullLogger<OrderBatchJobPublisher>.Instance);
var act = () => disabled.PublishAsync(new OrderBatchJobMessage
{
Action = OrderBatchActionCodes.ForecastBilling,
OrderId = 1,
MessageId = "x"
});
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*disabled*");
}
}

48
src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs
View File

@ -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<Om
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IOutboxEventLogService _outbox;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderParcelCreateEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
@ -36,11 +35,6 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<Om
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_outbox = ServiceProvider.GetRequiredService<IOutboxEventLogService>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderParcelCreateEventPublisher(
_distributedEventBus,
_uowManager,
ServiceProvider.GetRequiredService<IServiceScopeFactory>(),
NullLogger<OrderParcelCreateEventPublisher>.Instance);
}
[Fact]
@ -58,7 +52,11 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<Om
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order, "lazada");
await _distributedEventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent
{
OrderId = order.Id,
PlatformCode = "lazada"
});
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
@ -86,8 +84,19 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<Om
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishTrackUpsertAsync(order, request, customerId: 9);
await _publisher.PublishLabelGenerateAsync(order, operatorUserId: 7);
await _distributedEventBus.PublishAsync(new OrderParcelTrackUpsertIntegrationEvent
{
OrderId = order.Id,
CustomerId = 9,
PayloadJson = JsonConvert.SerializeObject(request),
AppendInitialTrackOnSuccess = request.AppendInitialTrackOnSuccess,
Authorization = request.Authorization
});
await _distributedEventBus.PublishAsync(new OrderParcelLabelGenerateIntegrationEvent
{
OrderId = order.Id,
OperatorUserId = 7
});
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
@ -108,7 +117,10 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<Om
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order);
await _distributedEventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent
{
OrderId = order.Id
});
await uow.RollbackAsync();
}
@ -116,16 +128,16 @@ public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<Om
}
[Fact]
public async Task 无环境事务时_短事务仍能写入并投递()
public async Task 无环境事务时_总线应拒绝发布()
{
_physicalBus.Clear();
_uowManager.Current.Should().BeNull();
var order = new Order { Id = 404, ServiceNo = "SN404" };
await _publisher.PublishCustoms9610UpsertAsync(order, "shopee");
var act = () => _distributedEventBus.PublishAsync(new OrderParcelCustoms9610UpsertIntegrationEvent
{
OrderId = 404,
PlatformCode = "shopee"
});
_physicalBus.Published.Should().ContainSingle()
.Which.Should().BeOfType<OrderParcelCustoms9610UpsertIntegrationEvent>()
.Which.OrderId.Should().Be(404);
await act.Should().ThrowAsync<InvalidOperationException>();
}
}

20
src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs
View File

@ -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<OmsO
private readonly IDistributedEventBus _distributedEventBus;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderSideEffectEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
@ -33,10 +31,6 @@ public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools<OmsO
_distributedEventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderSideEffectEventPublisher(
_distributedEventBus,
_uowManager,
NullLogger<OrderSideEffectEventPublisher>.Instance);
}
[Fact]
@ -47,8 +41,18 @@ public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools<OmsO
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAutoTriggerFireAsync(11, AutoTriggerTiming.OrderCreate, 3);
await _publisher.PublishLifecycleTrackAppendAsync(11, "SN11", "A0301");
await _distributedEventBus.PublishAsync(new OrderAutoTriggerFireIntegrationEvent
{
OrderId = 11,
Timing = AutoTriggerTiming.OrderCreate.ToString(),
OperatorUserId = 3
});
await _distributedEventBus.PublishAsync(new OrderLifecycleTrackAppendIntegrationEvent
{
OrderId = 11,
OrderNo = "SN11",
TrackCode = "A0301"
});
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}

20
src/Oms.Tests/OrderBatchJobCoordinatorGovernanceTests.cs
View File

@ -1,7 +1,7 @@
using FluentAssertions;
using Fake.Application;
using Fake.EventBus.Distributed;
using Fake.ObjectStorage;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Moq;
@ -23,13 +23,12 @@ public class OrderBatchJobCoordinatorGovernanceTests
private readonly Mock<ICountryRepository> _countryRepository = new(MockBehavior.Loose);
private readonly Mock<IServiceChannelRepository> _serviceChannelRepository = new(MockBehavior.Loose);
private readonly Mock<IObjectStorage> _fileStorage = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchJobPublisher> _publisher = new(MockBehavior.Strict);
private readonly Mock<IDistributedEventBus> _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<IOrderBatchActionExecutor>(MockBehavior.Loose);
var processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
@ -37,17 +36,14 @@ public class OrderBatchJobCoordinatorGovernanceTests
actionExecutor.Object,
batchOptions,
NullLogger<OrderBatchJobProcessor>.Instance);
var unitOfWorkManager = new Mock<IUnitOfWorkManager>(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<OrderBatchJobCoordinator>.Instance);
}
@ -71,8 +67,8 @@ public class OrderBatchJobCoordinatorGovernanceTests
_batchJobRepository
.Setup(x => x.UpdateJobAsync(job, It.IsAny<CancellationToken>()))
.Returns(Task.CompletedTask);
_publisher
.Setup(x => x.PublishAsync(It.IsAny<OrderBatchJobMessage>(), It.IsAny<CancellationToken>()))
_eventBus
.Setup(x => x.PublishAsync(It.IsAny<OrderBatchJobIntegrationEvent>(), It.IsAny<CancellationToken>()))
.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<OrderBatchJobMessage>(m => m.JobItemId == failedItem.Id),
It.Is<OrderBatchJobIntegrationEvent>(m => m.JobItemId == failedItem.Id),
It.IsAny<CancellationToken>()),
Times.Once);
}