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

feat: 创单/出入库后置副作用接入 Fake.EventBus，预报计费排货走 MQ

Browse Source 
将轨迹、面单、9610、揽收、自动触发与 ForecastBilling 异步化，缩短扫码与下单 RT；队列未启用时同步回落。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zzh 2026-10-10 10:54:24 +08:00
parent da27fa4377
commit dcd985bca4
27 changed files with 1259 additions and 207 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/Oms.Application/Dtos/Orders/OrderParcelCreateOutput.cs
View File

@ -41,12 +41,12 @@ public class OrderParcelCreateOutput
public string? ConsigneeCountryName { get; set; }
/// <summary>
/// 同步创建的揽收单号（上门揽收成功时）。
/// 揽收单号（上门揽收；事件总线消费较快或同步回落时可能已有值）。
/// </summary>
public string? PickupNo { get; set; }
/// <summary>
/// 同步创建的揽收单 Id。
/// 揽收单 Id。
/// </summary>
public int? PickupOrderId { get; set; }
@ -59,4 +59,24 @@ public class OrderParcelCreateOutput
/// 揽收单创建失败原因（运单已成功时仅作提示）。
/// </summary>
public string? PickupCreateError { get; set; }
/// <summary>
/// 揽收单已入事件总线、后台创建中（此时尚无 PickupNo）。
/// </summary>
public bool PickupQueued { get; set; }
/// <summary>
/// 订单状态（创单后可能仍为草稿，预报计费异步处理中）。
/// </summary>
public string? OrderStatus { get; set; }
/// <summary>
/// 预报失败/处理中原因。
/// </summary>
public string? PredeclareFailReason { get; set; }
/// <summary>
/// 预报计费/排货已入队或后台处理中。
/// </summary>
public bool ForecastBillingQueued { get; set; }
}

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

@ -1,6 +1,9 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Application.Services.Orders.Batch;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Integrations.SmallBagTrack;
@ -15,31 +18,42 @@ internal sealed class SmallBagTrackOrderNotifier(
try
{
using var scope = scopeFactory.CreateScope();
var coordinator = scope.ServiceProvider.GetRequiredService<OrderBatchJobCoordinator>();
var published = await coordinator.PublishTrackUpsertAsync(request, orderId, customerId);
if (!published)
var orderRepository = scope.ServiceProvider.GetRequiredService<IOrderRepository>();
var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == orderId);
if (order == null)
{
await FallbackDirectAsync(request, orderId);
return;
}
var publisher = scope.ServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
await publisher.PublishTrackUpsertAsync(order, request, customerId);
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"RabbitMQ track upsert publish failed, fallback to direct call. orderId={OrderId}",
"Track upsert publish failed, fallback via batch coordinator. orderId={OrderId}",
orderId);
await FallbackDirectAsync(request, orderId);
await FallbackBatchOrDirectAsync(request, orderId, customerId);
}
});
}
private async Task FallbackDirectAsync(SmallBagTrackUpsertRequest request, int orderId)
private async Task FallbackBatchOrDirectAsync(
SmallBagTrackUpsertRequest request,
int orderId,
int? customerId)
{
try
{
using var scope = scopeFactory.CreateScope();
var client = scope.ServiceProvider.GetRequiredService<ISmallBagTrackOrderClient>();
await client.UpsertAsync(request);
var coordinator = scope.ServiceProvider.GetRequiredService<OrderBatchJobCoordinator>();
var published = await coordinator.PublishTrackUpsertAsync(request, orderId, customerId);
if (!published)
{
var client = scope.ServiceProvider.GetRequiredService<ISmallBagTrackOrderClient>();
await client.UpsertAsync(request);
}
}
catch (Exception ex)
{
@ -49,23 +63,29 @@ internal sealed class SmallBagTrackOrderNotifier(
public void EnqueueLifecycleTrackAppend(string orderNo, string trackCode, int orderId, string? authorization = null)
{
_ = Task.Run(async () =>
_ = PublishLifecycleAsync(orderNo, trackCode, orderId, authorization);
}
private async Task PublishLifecycleAsync(
string orderNo,
string trackCode,
int orderId,
string? authorization)
{
try
{
try
{
using var scope = scopeFactory.CreateScope();
var client = scope.ServiceProvider.GetRequiredService<ISmallBagTrackOrderClient>();
await client.AppendLifecycleTrackAsync(orderNo, trackCode, authorization);
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Async lifecycle track append failed. orderId={OrderId} orderNo={OrderNo} trackCode={TrackCode}",
orderId,
orderNo,
trackCode);
}
});
using var scope = scopeFactory.CreateScope();
var publisher = scope.ServiceProvider.GetRequiredService<IOrderSideEffectEventPublisher>();
await publisher.PublishLifecycleTrackAppendAsync(orderId, orderNo, trackCode, authorization);
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Lifecycle track append publish failed. orderId={OrderId} orderNo={OrderNo} trackCode={TrackCode}",
orderId,
orderNo,
trackCode);
}
}
}

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

@ -0,0 +1,22 @@
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);
}

48
src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireEventHandler.cs Normal file
View File

@ -0,0 +1,48 @@
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders.Issues;
namespace Oms.Application.Messaging.OrderSideEffects;
public sealed class OrderAutoTriggerFireEventHandler(
AutoTriggerRuntimeService autoTriggerRuntimeService,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderAutoTriggerFireEventHandler> logger)
: IEventHandler<OrderAutoTriggerFireIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderAutoTriggerFireIntegrationEvent @event,
CancellationToken cancellationToken)
{
if (@event.OrderId <= 0
|| !Enum.TryParse<AutoTriggerTiming>(@event.Timing, ignoreCase: true, out var timing))
{
logger.LogWarning(
"AutoTrigger fire skipped invalid payload OrderId={OrderId} Timing={Timing}",
@event.OrderId,
@event.Timing);
return;
}
using var unitOfWork = unitOfWorkManager.Begin();
try
{
await autoTriggerRuntimeService.FireByOrderIdAsync(
@event.OrderId,
timing,
@event.OperatorUserId);
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"AutoTrigger fire failed OrderId={OrderId} Timing={Timing}",
@event.OrderId,
timing);
}
await unitOfWork.CompleteAsync(cancellationToken);
}
}

19
src/Oms.Application/Messaging/OrderSideEffects/OrderAutoTriggerFireIntegrationEvent.cs Normal file
View File

@ -0,0 +1,19 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.OrderSideEffects;
/// <summary>
/// 订单自动触发（下单 / 入库 / 出库时机），走 Fake.EventBus.RabbitMQ。
/// </summary>
[Serializable]
public class OrderAutoTriggerFireIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
/// <summary>
/// <see cref="Services.Orders.Issues.AutoTriggerTiming"/> 枚举名。
/// </summary>
public string Timing { get; set; } = string.Empty;
public int OperatorUserId { get; set; }
}

49
src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendEventHandler.cs Normal file
View File

@ -0,0 +1,49 @@
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Integrations.SmallBagTrack;
namespace Oms.Application.Messaging.OrderSideEffects;
public sealed class OrderLifecycleTrackAppendEventHandler(
ISmallBagTrackOrderClient trackClient,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderLifecycleTrackAppendEventHandler> logger)
: IEventHandler<OrderLifecycleTrackAppendIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderLifecycleTrackAppendIntegrationEvent @event,
CancellationToken cancellationToken)
{
if (string.IsNullOrWhiteSpace(@event.OrderNo) || string.IsNullOrWhiteSpace(@event.TrackCode))
{
logger.LogWarning(
"Lifecycle track append skipped empty OrderNo/TrackCode OrderId={OrderId}",
@event.OrderId);
return;
}
using var unitOfWork = unitOfWorkManager.Begin();
try
{
await trackClient.AppendLifecycleTrackAsync(
@event.OrderNo,
@event.TrackCode,
@event.Authorization,
cancellationToken);
}
catch (Exception ex)
{
// 轨迹失败不反复打满队列；与原 Task.Run 吞异常策略一致
logger.LogWarning(
ex,
"Lifecycle track append failed OrderId={OrderId} OrderNo={OrderNo} TrackCode={TrackCode}",
@event.OrderId,
@event.OrderNo,
@event.TrackCode);
}
await unitOfWork.CompleteAsync(cancellationToken);
}
}

18
src/Oms.Application/Messaging/OrderSideEffects/OrderLifecycleTrackAppendIntegrationEvent.cs Normal file
View File

@ -0,0 +1,18 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.OrderSideEffects;
/// <summary>
/// 订单生命周期轨迹追加（如入库 A0301、出库等），走 Fake.EventBus.RabbitMQ。
/// </summary>
[Serializable]
public class OrderLifecycleTrackAppendIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
public string OrderNo { get; set; } = string.Empty;
public string TrackCode { get; set; } = string.Empty;
public string? Authorization { get; set; }
}

112
src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs Normal file
View File

@ -0,0 +1,112 @@
using Fake.EventBus.Distributed;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders.Issues;
namespace Oms.Application.Messaging.OrderSideEffects;
/// <summary>
/// 优先 Fake.EventBus.RabbitMQ；发布失败则同进程同步执行 Handler。
/// </summary>
public sealed class OrderSideEffectEventPublisher(
IDistributedEventBus eventBus,
IServiceScopeFactory scopeFactory,
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()
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderLifecycleTrackAppendEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
public async Task PublishAutoTriggerFireAsync(
int orderId,
AutoTriggerTiming timing,
int operatorUserId,
CancellationToken cancellationToken = default)
{
if (orderId <= 0)
{
return;
}
var @event = new OrderAutoTriggerFireIntegrationEvent
{
OrderId = orderId,
Timing = timing.ToString(),
OperatorUserId = operatorUserId
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderAutoTriggerFireEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
private async Task<bool> TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken)
{
try
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Published side-effect event {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return true;
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Publish side-effect event {EventName} failed, fallback to sync. EventId={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
private async Task ExecuteInScopeAsync<THandler>(
Func<THandler, Task> action,
CancellationToken cancellationToken)
where THandler : notnull
{
try
{
using var scope = scopeFactory.CreateScope();
var handler = scope.ServiceProvider.GetRequiredService<THandler>();
await action(handler);
}
catch (Exception ex)
{
logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name);
}
}
}

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

@ -0,0 +1,46 @@
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);
}

198
src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs Normal file
View File

@ -0,0 +1,198 @@
using Fake.EventBus.Distributed;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Oms.Application.Dtos.Orders;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Application.Services.Orders;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置事件发布器：优先走 Fake.EventBus.RabbitMQ，失败则同进程同步执行 Handler 逻辑。
/// </summary>
public sealed class OrderParcelCreateEventPublisher(
IDistributedEventBus eventBus,
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
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelTrackUpsertEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
public async Task PublishLabelGenerateAsync(
Order order,
int operatorUserId = 0,
CancellationToken cancellationToken = default)
{
var @event = new OrderParcelLabelGenerateIntegrationEvent
{
OrderId = order.Id,
OperatorUserId = operatorUserId
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelLabelGenerateEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
public async Task PublishCustoms9610UpsertAsync(
Order order,
string? platformCode = null,
CancellationToken cancellationToken = default)
{
var @event = new OrderParcelCustoms9610UpsertIntegrationEvent
{
OrderId = order.Id,
PlatformCode = platformCode
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync<OrderParcelCustoms9610UpsertEventHandler>(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
public async Task<bool> PublishPickupAttachAsync(
OrderParcelCreateInput input,
int orderId,
CancellationToken cancellationToken = default)
{
var @event = BuildPickupEvent(input, orderId);
if (@event == null)
{
return false;
}
// 揽收回落由调用方 ExecutePickupAttachAsync 同步执行，避免重复建单且便于回写单号。
return await TryPublishAsync(@event, cancellationToken);
}
public async Task<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
};
}
private async Task<bool> TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken)
{
try
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Published parcel-create event {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return true;
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Publish parcel-create event {EventName} failed, fallback to sync. EventId={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
private async Task ExecuteInScopeAsync<THandler>(
Func<THandler, Task> action,
CancellationToken cancellationToken)
where THandler : notnull
{
try
{
using var scope = scopeFactory.CreateScope();
var handler = scope.ServiceProvider.GetRequiredService<THandler>();
await action(handler);
}
catch (Exception ex)
{
// 同步回落失败也不抛回创单主流程
logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name);
}
}
}

44
src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertEventHandler.cs Normal file
View File

@ -0,0 +1,44 @@
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 处理创单后 9610 汇总申报订单写入事件。
/// </summary>
public sealed class OrderParcelCustoms9610UpsertEventHandler(
IOrderRepository orderRepository,
Customs9610OrderWriter customs9610OrderWriter,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderParcelCustoms9610UpsertEventHandler> logger)
: IEventHandler<OrderParcelCustoms9610UpsertIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderParcelCustoms9610UpsertIntegrationEvent @event,
CancellationToken cancellationToken)
{
using var unitOfWork = unitOfWorkManager.Begin();
var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == @event.OrderId, cancellationToken);
if (order == null)
{
logger.LogWarning("9610 upsert skipped, order missing OrderId={OrderId}", @event.OrderId);
return;
}
try
{
await customs9610OrderWriter.UpsertFromOrderAsync(order, @event.PlatformCode, cancellationToken);
}
catch (Exception ex)
{
// 与原同步路径一致：9610 失败不抛到总线重试风暴
logger.LogWarning(ex, "9610 upsert failed OrderId={OrderId}", @event.OrderId);
}
await unitOfWork.CompleteAsync(cancellationToken);
}
}

14
src/Oms.Application/Messaging/ParcelCreate/OrderParcelCustoms9610UpsertIntegrationEvent.cs Normal file
View File

@ -0,0 +1,14 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置：按 9610 汇总申报写入/刷新 t_9610_order（非 9610 时 Handler 内会按规则清理 pending）。
/// </summary>
[Serializable]
public class OrderParcelCustoms9610UpsertIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
public string? PlatformCode { get; set; }
}

38
src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateEventHandler.cs Normal file
View File

@ -0,0 +1,38 @@
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 处理创单后系统面单生成事件。
/// </summary>
public sealed class OrderParcelLabelGenerateEventHandler(
IServiceScopeFactory scopeFactory,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderParcelLabelGenerateEventHandler> logger)
: IEventHandler<OrderParcelLabelGenerateIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderParcelLabelGenerateIntegrationEvent @event,
CancellationToken cancellationToken)
{
using var unitOfWork = unitOfWorkManager.Begin();
using var scope = scopeFactory.CreateScope();
var orderService = scope.ServiceProvider.GetRequiredService<OrderService>();
var result = await orderService.GenerateLabelPdfForBatchAsync(@event.OrderId, cancellationToken);
if (!result.Success)
{
// 面单失败不阻断创单主链路；记录后正常 Ack，避免反复重试打满队列
logger.LogWarning(
"Label generate failed OrderId={OrderId} Error={Error}",
@event.OrderId,
result.Error);
}
await unitOfWork.CompleteAsync(cancellationToken);
}
}

14
src/Oms.Application/Messaging/ParcelCreate/OrderParcelLabelGenerateIntegrationEvent.cs Normal file
View File

@ -0,0 +1,14 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置：生成系统面单 PDF。
/// </summary>
[Serializable]
public class OrderParcelLabelGenerateIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
public int OperatorUserId { get; set; }
}

51
src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachEventHandler.cs Normal file
View File

@ -0,0 +1,51 @@
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Dtos.Orders;
using Oms.Application.Services.Orders;
using Oms.Domain.OrdersAggregate.Consts;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 处理创单后上门揽收挂接事件。
/// </summary>
public sealed class OrderParcelPickupAttachEventHandler(
PickupOrderService pickupOrderService,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderParcelPickupAttachEventHandler> logger)
: IEventHandler<OrderParcelPickupAttachIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderParcelPickupAttachIntegrationEvent @event,
CancellationToken cancellationToken)
{
using var unitOfWork = unitOfWorkManager.Begin();
try
{
await pickupOrderService.CreateForParcelInternalAsync(new PickupOrderCreateInput
{
CustomerId = @event.CustomerId,
SenderAddressId = @event.SenderAddressId,
Direction = PickupDirectionCodes.NormalizeOrDefault(@event.PickupDirection),
WarehouseId = @event.WarehouseId,
CarrierCode = @event.CarrierCode,
CarrierService = @event.CarrierService,
ExpectPickupTime = @event.ExpectPickupTime,
OrderIds = [@event.OrderId],
Remark = null
});
}
catch (Exception ex)
{
// 与原同步路径一致：揽收失败不回滚物流单，仅记日志（列表可补建）
logger.LogWarning(
ex,
"Pickup attach failed OrderId={OrderId}",
@event.OrderId);
}
await unitOfWork.CompleteAsync(cancellationToken);
}
}

26
src/Oms.Application/Messaging/ParcelCreate/OrderParcelPickupAttachIntegrationEvent.cs Normal file
View File

@ -0,0 +1,26 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置：上门揽收时创建并挂接揽收订单（可选再预报承运）。
/// </summary>
[Serializable]
public class OrderParcelPickupAttachIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
public int? CustomerId { get; set; }
public int SenderAddressId { get; set; }
public int WarehouseId { get; set; }
public string? PickupDirection { get; set; }
public string? CarrierCode { get; set; }
public string? CarrierService { get; set; }
public DateTime ExpectPickupTime { get; set; }
}

76
src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertEventHandler.cs Normal file
View File

@ -0,0 +1,76 @@
using Fake.Application;
using Fake.DependencyInjection;
using Fake.EventBus;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 处理「轨迹订单 + 创建成功轨迹」合并事件。
/// </summary>
public sealed class OrderParcelTrackUpsertEventHandler(
IOrderRepository orderRepository,
ISmallBagTrackOrderClient trackClient,
ISmallBagTrackUpsertRequestBuilder requestBuilder,
IUnitOfWorkManager unitOfWorkManager,
ILogger<OrderParcelTrackUpsertEventHandler> logger)
: IEventHandler<OrderParcelTrackUpsertIntegrationEvent>, ITransientDependency
{
public async Task HandleAsync(
OrderParcelTrackUpsertIntegrationEvent @event,
CancellationToken cancellationToken)
{
using var unitOfWork = unitOfWorkManager.Begin();
var request = await ResolveRequestAsync(@event, cancellationToken);
request.AppendInitialTrackOnSuccess = @event.AppendInitialTrackOnSuccess;
if (string.IsNullOrWhiteSpace(request.Authorization)
&& !string.IsNullOrWhiteSpace(@event.Authorization))
{
request.Authorization = @event.Authorization.Trim();
}
var result = await trackClient.UpsertDebugAsync(request, cancellationToken);
if (result.Skipped)
{
logger.LogInformation(
"Track upsert skipped OrderId={OrderId} Reason={Reason}",
@event.OrderId,
result.SkipReason);
}
else if (!result.Success)
{
throw new BusinessException(
$"轨迹订单创建失败: HTTP {result.StatusCode}");
}
await unitOfWork.CompleteAsync(cancellationToken);
}
private async Task<SmallBagTrackUpsertRequest> ResolveRequestAsync(
OrderParcelTrackUpsertIntegrationEvent @event,
CancellationToken cancellationToken)
{
if (!string.IsNullOrWhiteSpace(@event.PayloadJson))
{
var fromPayload = JsonConvert.DeserializeObject<SmallBagTrackUpsertRequest>(@event.PayloadJson);
if (fromPayload != null && !string.IsNullOrWhiteSpace(fromPayload.OrderNo))
{
return fromPayload;
}
}
var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == @event.OrderId, cancellationToken)
?? throw new BusinessException("订单不存在");
if (string.IsNullOrWhiteSpace(order.ServiceNo))
{
throw new BusinessException("订单运单号为空，无法同步轨迹");
}
return await requestBuilder.BuildAsync(order, cancellationToken: cancellationToken);
}
}

28
src/Oms.Application/Messaging/ParcelCreate/OrderParcelTrackUpsertIntegrationEvent.cs Normal file
View File

@ -0,0 +1,28 @@
using Fake.EventBus.Distributed;
namespace Oms.Application.Messaging.ParcelCreate;
/// <summary>
/// 创单后置：新增轨迹订单，并在成功后追加「创建订单成功」轨迹（A0101）。
/// 二者合并为同一集成事件，由 Handler 内 UpsertSmallBag + AppendInitialTrackOnSuccess 一次完成。
/// </summary>
[Serializable]
public class OrderParcelTrackUpsertIntegrationEvent : IntegrationEvent
{
public int OrderId { get; set; }
public int? CustomerId { get; set; }
/// <summary>
/// 序列化后的 <see cref="Integrations.SmallBagTrack.SmallBagTrackUpsertRequest"/>；
/// 为空时 Handler 按 OrderId 重建。
/// </summary>
public string? PayloadJson { get; set; }
/// <summary>
/// 是否在 Upsert 成功后追加创单默认轨迹 A0101。
/// </summary>
public bool AppendInitialTrackOnSuccess { get; set; } = true;
public string? Authorization { get; set; }
}

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

@ -1,6 +1,8 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Configuration;
using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Messaging.ParcelCreate;
namespace Oms.Application;
@ -11,6 +13,8 @@ public static class OrderBatchMessagingServiceExtensions
services.Configure<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;

14
src/Oms.Application/Services/Operations/OutboundRecordService.cs
View File

@ -161,18 +161,8 @@ public class OutboundRecordService : ApplicationService
var savedEntity = await OutboundRecordRepository.InsertAsync(entity);
try
{
var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim());
if (refreshed != null)
{
await AutoTriggerRuntimeService.FireAsync(refreshed, AutoTriggerTiming.Outbound, operatorId);
}
}
catch
{
// 自动触发失败不影响出库主流程
}
// 自动触发改走 Fake.EventBus，避免拉长出库扫码 RT
AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Outbound, operatorId);
return ObjectMapper.Map<OutboundRecord, OutboundRecordOutput>(savedEntity);
}

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

@ -49,6 +49,7 @@ public sealed class OrderBatchActionExecutor(
OrderBatchActionCodes.Print => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.GenerateLabelPdf => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.InboundBilling => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.ForecastBilling => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.Reallocate => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.MarkProblematic => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.MarkClaimed => await orderActionHandler.ExecuteAsync(message, cancellationToken),

1
src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs
View File

@ -269,6 +269,7 @@ public static class OrderBatchJobItemFeedback
OrderBatchActionCodes.Print => "打印成功",
OrderBatchActionCodes.GenerateLabelPdf => "面单生成成功",
OrderBatchActionCodes.InboundBilling => "入库计费/排货完成",
OrderBatchActionCodes.ForecastBilling => "预报计费/排货完成",
OrderBatchActionCodes.Reallocate => "重新排货成功",
OrderBatchActionCodes.MarkProblematic => "问题件创建成功",
OrderBatchActionCodes.MarkClaimed => "理赔件创建成功",

8
src/Oms.Application/Services/Orders/Batch/OrderBatchOrderActionHandler.cs
View File

@ -67,6 +67,14 @@ public sealed class OrderBatchOrderActionHandler(IServiceScopeFactory scopeFacto
await orderService.ProcessInboundBillingForBatchAsync(message.OrderId, cancellationToken);
return OrderBatchActionExecuteResult.Ok();
case OrderBatchActionCodes.ForecastBilling:
await orderService.ProcessForecastBillingForBatchAsync(
message.OrderId,
message.PayloadJson,
message.OperatorUserId,
cancellationToken);
return OrderBatchActionExecuteResult.Ok();
case OrderBatchActionCodes.Reallocate:
await orderService.ReallocateForBatchAsync(
message.OrderId,

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

@ -1,6 +1,6 @@
using Fake.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Domain.ChannelsAggregate.Consts;
using Oms.Domain.ChannelsAggregate.Entities;
using Oms.Domain.ChannelsAggregate.Repositories;
@ -16,13 +16,13 @@ public class AutoTriggerRuntimeService(
OrderAutoTriggerService orderAutoTriggerService,
OrderExceptionWorkflowService exceptionWorkflowService,
OrderIssueNotificationService issueNotificationService,
IServiceScopeFactory scopeFactory,
IOrderSideEffectEventPublisher sideEffectEventPublisher,
ILogger<AutoTriggerRuntimeService> logger) : IScopedDependency
{
private const string InterceptOperation = "拦截";
/// <summary>
/// 后台异步触发，避免拉长入库/出库扫码请求。
/// 后台异步触发（Fake.EventBus.RabbitMQ），避免拉长入库/出库/创单请求。
/// </summary>
public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId)
{
@ -31,23 +31,7 @@ public class AutoTriggerRuntimeService(
return;
}
_ = Task.Run(async () =>
{
try
{
using var scope = scopeFactory.CreateScope();
var runtime = scope.ServiceProvider.GetRequiredService<AutoTriggerRuntimeService>();
await runtime.FireByOrderIdAsync(orderId, timing, operatorUserId);
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"异步自动触发失败 OrderId={OrderId} Timing={Timing}",
orderId,
timing);
}
});
_ = sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId);
}
public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId)

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

@ -9,6 +9,8 @@ using Oms.Application.Dtos.Orders;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Application.Labeling;
using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Application.Shared;
using Oms.Application.Storage;
using Oms.Application.Services.Orders.Batch;
@ -36,6 +38,8 @@ public partial class OrderService
private const int ParcelBatchMaxCount = 2000;
private const string InboundBillingProcessingReason = "入库后计费/排货处理中";
private const string InboundBillingEnqueueFailReason = "入库计费任务投递失败，请重跑入库计费";
private const string ForecastBillingProcessingReason = "预报计费/排货处理中";
private const string ForecastBillingEnqueueFailReason = "预报计费任务投递失败，请重新预报";
private static readonly ConcurrentDictionary<int, byte> InboundBillingInFlight = new();
@ -143,10 +147,11 @@ public partial class OrderService
// 客户主动「先保存为草稿」：落库草稿后直接返回，不跑预报计费。
if (input.SaveAsDraft)
{
await TryUpsertCustoms9610Async(order, input.PlatformCode);
await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode);
return await BuildParcelCreateOutputAsync(order);
}
// 偏远标记轻量同步；计费重/应收/排货走 ForecastBilling MQ（失败不回滚建单）。
order.IsRemote = await IsRemoteHitAsync(
product,
snapshotBinding.Rules,
@ -154,62 +159,24 @@ public partial class OrderService
input.Receiver?.Province,
input.Receiver?.City,
input.Receiver?.ZipCode) ? (byte)1 : (byte)0;
var forecast = weights.First(x => x.Type == OrderWeightTypeCodes.Forecast);
await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast);
SyncWeightRows(weights, forecast);
await OrderRepository.ReplaceWeightsAsync(order.Id, weights);
await TryUpsertCustoms9610Async(order, input.PlatformCode);
var customer = await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId);
var priceTierKey = await ResolvePriceTierKeyAsync(order.CustomerId);
var billing = await ResolveBillingContextAsync(
product,
snapshotBinding,
new OrderBillingAddress(
country.CountryCode,
input.Receiver?.Province,
input.Receiver?.City,
input.Receiver?.ZipCode),
forecast.ChargeableWeight ?? 0,
order.CargoAttribute ?? CargoAttributeCodes.General,
order.WarehouseId,
priceTierKey);
await UpsertReceivableFeesAsync(
order,
billing,
OrderFeeChargingStageCodes.Forecast);
// 预报阶段尝试排货写应付；失败不阻断创建（仅记录 AllocationPendingReason）。
await TryConfirmServiceChannelAndPayableAsync(
order,
product,
billing,
OrderFeeChargingStageCodes.Forecast);
await ApplyMarketingSnapshotOnCreateAsync(order, input.CouponInstanceId);
await CaptureForecastBillingSnapshotAsync(order);
order.OrderStatus = OrderStatusCodes.Predeclared;
order.PredeclareFailReason = string.Empty;
order.OrderStatus = OrderStatusCodes.Draft;
order.PredeclareFailReason = ForecastBillingProcessingReason;
order.AllocationPendingReason = string.Empty;
order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order);
await EnqueueSmallBagTrackUpsertAsync(order, customer, input.Receiver, appendInitialTrack: true);
await EnqueueGenerateLabelPdfAsync(order);
await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode);
await EnqueueForecastBillingAsync(
order,
input.CouponInstanceId,
operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
try
{
var triggerUserId = operatorUserId ?? CurrentUser.GetId<int>();
await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId);
}
catch
{
// 自动触发失败不影响预报主流程
}
return await BuildParcelCreateOutputAsync(order);
var output = await BuildParcelCreateOutputAsync(order);
output.ForecastBillingQueued = string.Equals(
order.PredeclareFailReason,
ForecastBillingProcessingReason,
StringComparison.Ordinal);
return output;
}
catch (Exception ex)
{
@ -2287,8 +2254,7 @@ public partial class OrderService
}
/// <summary>
/// 草稿重新预报：按订单当前收件/件重尺/快照重跑黑名单、计重、分区与预报应收。
/// 失败时保持草稿并写入 PredeclareFailReason。
/// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货投递 ForecastBilling MQ。
/// </summary>
internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null)
{
@ -2319,63 +2285,52 @@ public partial class OrderService
throw new BusinessException("收件地址命中黑名单区域");
}
order.IsRemote = await IsRemoteHitAsync(
product,
snapshotBinding.Rules,
address.CountryCode,
address.Province,
address.City,
address.ZipCode) ? (byte)1 : (byte)0;
var weights = await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]);
if (!weights.Any(x =>
string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase)))
{
throw new BusinessException("订单缺少预报件重尺");
}
var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]))
.OrderBy(x => x.Id)
.ToList();
var forecast = weights.FirstOrDefault(x =>
string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase))
?? throw new BusinessException("订单缺少预报件重尺");
await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast);
SyncWeightRows(weights, forecast);
await OrderRepository.ReplaceWeightsAsync(order.Id, weights);
var billing = await ResolveOrderBillingContextAsync(
order,
product,
forecast.ChargeableWeight ?? 0,
snapshotBinding);
await UpsertReceivableFeesAsync(
order,
billing,
OrderFeeChargingStageCodes.Forecast,
replaceExistingReceivable: true);
order.OrderStatus = OrderStatusCodes.Predeclared;
order.PredeclareFailReason = string.Empty;
order.OrderStatus = OrderStatusCodes.Draft;
order.PredeclareFailReason = ForecastBillingProcessingReason;
order.AllocationPendingReason = string.Empty;
order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order);
var customer = order.CustomerId.HasValue
? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value)
: null;
await EnqueueSmallBagTrackUpsertAsync(order, customer, receiver: null, appendInitialTrack: true);
await EnqueueGenerateLabelPdfAsync(order);
await EnqueueForecastBillingAsync(
order,
order.CouponInstanceId,
operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
try
// 同步回落失败时给出明确错误；入队成功则保持「处理中」由列表观察结果。
var refreshed = await OrderRepository.FirstOrDefaultAsync(x => x.Id == order.Id);
if (refreshed != null
&& string.Equals(refreshed.OrderStatus, OrderStatusCodes.Draft, StringComparison.OrdinalIgnoreCase)
&& !string.IsNullOrWhiteSpace(refreshed.PredeclareFailReason)
&& !string.Equals(
refreshed.PredeclareFailReason,
ForecastBillingProcessingReason,
StringComparison.Ordinal))
{
var triggerUserId = operatorUserId ?? CurrentUser.GetId<int>();
await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId);
}
catch
{
// 自动触发失败不影响预报主流程
throw CreateDraftSavedBusinessException(
refreshed,
StripDraftSavedMarker(refreshed.PredeclareFailReason));
}
}
catch (BusinessException ex)
{
if (string.Equals(ex.ErrorCode, OrderParcelDraftFailData.ErrorCodeValue, StringComparison.Ordinal))
{
throw;
}
var cleanReason = StripDraftSavedMarker(ex.Message);
await MarkParcelDraftFailedAsync(order, cleanReason);
if (!string.Equals(cleanReason, ForecastBillingProcessingReason, StringComparison.Ordinal))
{
await MarkParcelDraftFailedAsync(order, cleanReason);
}
throw CreateDraftSavedBusinessException(order, cleanReason);
}
catch (Exception ex)
@ -2464,7 +2419,7 @@ public partial class OrderService
order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order);
EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.InboundTrackCode);
await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.InboundTrackCode);
await EnqueueInboundBillingAsync(order);
}
finally
@ -2646,6 +2601,256 @@ public partial class OrderService
return text[..512];
}
/// <summary>
/// 预报计费/排货异步任务（创单与草稿预报共用）。
/// </summary>
internal async Task ProcessForecastBillingForBatchAsync(
int orderId,
string? payloadJson = null,
int operatorUserId = 0,
CancellationToken cancellationToken = default)
{
var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId, cancellationToken);
if (order == null)
{
return;
}
// 已预报成功则跳过（幂等）
if (string.Equals(order.OrderStatus, OrderStatusCodes.Predeclared, StringComparison.OrdinalIgnoreCase)
&& string.IsNullOrWhiteSpace(order.PredeclareFailReason))
{
return;
}
int? couponInstanceId = order.CouponInstanceId;
if (!string.IsNullOrWhiteSpace(payloadJson))
{
try
{
var payload = JsonConvert.DeserializeObject<ForecastBillingJobPayload>(payloadJson);
if (payload?.CouponInstanceId is > 0)
{
couponInstanceId = payload.CouponInstanceId;
}
}
catch
{
// ignore bad payload
}
}
if (string.IsNullOrWhiteSpace(order.ProductCode))
{
await MarkParcelDraftFailedAsync(order, "订单未绑定销售产品，无法预报");
return;
}
var product = await SalesProductRepository.GetByProductCodeAsync(order.ProductCode.Trim());
if (product == null || product.IsActive != EnableStatus.Enabled)
{
await MarkParcelDraftFailedAsync(order, "销售产品不存在或未启用");
return;
}
ProductSnapshotBinding snapshotBinding;
OrderBillingAddress address;
try
{
snapshotBinding = await ResolveSnapshotBindingFromOrderAsync(product, order);
address = await ResolveOrderBillingAddressAsync(order);
order.IsRemote = await IsRemoteHitAsync(
product,
snapshotBinding.Rules,
address.CountryCode,
address.Province,
address.City,
address.ZipCode) ? (byte)1 : (byte)0;
}
catch (Exception ex)
{
await MarkParcelDraftFailedAsync(
order,
ex is BusinessException be ? be.Message : $"预报快照解析失败：{ex.Message}");
return;
}
var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]))
.OrderBy(x => x.Id)
.ToList();
var forecast = weights.FirstOrDefault(x =>
string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase));
if (forecast == null)
{
await MarkParcelDraftFailedAsync(order, "订单缺少预报件重尺");
return;
}
try
{
await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast);
SyncWeightRows(weights, forecast);
await OrderRepository.ReplaceWeightsAsync(order.Id, weights);
}
catch (Exception ex)
{
await MarkParcelDraftFailedAsync(
order,
ex is BusinessException be ? be.Message : $"计费重计算失败：{ex.Message}");
return;
}
OrderBillingContext billing;
try
{
billing = await ResolveOrderBillingContextAsync(
order,
product,
forecast.ChargeableWeight ?? 0,
snapshotBinding);
}
catch (Exception ex)
{
await MarkParcelDraftFailedAsync(
order,
ex is BusinessException be ? be.Message : $"计费上下文解析失败：{ex.Message}");
return;
}
try
{
await UpsertReceivableFeesAsync(
order,
billing,
OrderFeeChargingStageCodes.Forecast,
replaceExistingReceivable: true);
}
catch (Exception ex)
{
await MarkParcelDraftFailedAsync(
order,
ex is BusinessException be ? be.Message : $"预报应收计费失败：{ex.Message}");
return;
}
// 排货失败不阻断预报成功（与原创单同步路径一致）
await TryConfirmServiceChannelAndPayableAsync(
order,
product,
billing,
OrderFeeChargingStageCodes.Forecast);
try
{
await ApplyMarketingSnapshotOnCreateAsync(order, couponInstanceId);
}
catch (Exception ex)
{
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>()
.CreateLogger("Oms.ForecastBilling");
logger.LogWarning(ex, "预报营销快照失败 OrderId={OrderId}", order.Id);
}
try
{
await CaptureForecastBillingSnapshotAsync(order);
}
catch
{
// ignore
}
order.OrderStatus = OrderStatusCodes.Predeclared;
order.PredeclareFailReason = string.Empty;
order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order);
var customer = order.CustomerId.HasValue
? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value)
: null;
await PublishParcelCreateTrackAndLabelAsync(
order,
customer,
receiver: null,
appendInitialTrack: true,
operatorUserId);
AutoTriggerRuntimeService.EnqueueFire(
order.Id,
AutoTriggerTiming.OrderCreate,
operatorUserId);
}
private sealed class ForecastBillingJobPayload
{
public int? CouponInstanceId { get; set; }
}
private async Task EnqueueForecastBillingAsync(
Order order,
int? couponInstanceId,
int operatorUserId)
{
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>()
.CreateLogger("Oms.ForecastBilling");
var payloadJson = JsonConvert.SerializeObject(new ForecastBillingJobPayload
{
CouponInstanceId = couponInstanceId
});
try
{
var publisher = LazyServiceProvider.GetRequiredService<IOrderBatchJobPublisher>();
if (!publisher.IsEnabled)
{
try
{
await ProcessForecastBillingForBatchAsync(order.Id, payloadJson, operatorUserId);
}
catch (Exception ex)
{
logger.LogError(
ex,
"预报计费同步执行失败 OrderId={OrderId} ServiceNo={ServiceNo}",
order.Id,
order.ServiceNo);
await MarkParcelDraftFailedAsync(
order,
TrimPendingReason($"预报计费失败：{ex.Message}"));
}
return;
}
var requestId = $"forecast-billing:{order.Id}:{Guid.NewGuid():N}";
await BatchJobCoordinator.CreateAndEnqueueAsync(
new OrderBatchJobCreateInput
{
Action = OrderBatchActionCodes.ForecastBilling,
OrderIds = [order.Id],
RequestId = requestId,
PayloadJson = payloadJson
},
operatorUserId);
}
catch (Exception ex)
{
logger.LogError(
ex,
"预报计费任务投递失败 OrderId={OrderId} ServiceNo={ServiceNo}",
order.Id,
order.ServiceNo);
try
{
await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason);
}
catch
{
// ignore
}
}
}
private async Task EnqueueInboundBillingAsync(Order order)
{
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>()
@ -2819,7 +3024,7 @@ public partial class OrderService
order.UpdateTime = now;
await OrderRepository.UpdateAsync(order);
EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.OutboundTrackCode);
await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.OutboundTrackCode);
}
private async Task<OrderParcelCreateOutput> BuildParcelCreateOutputAsync(Order order)
@ -2877,46 +3082,56 @@ public partial class OrderService
TrackNo = order.TrackNo,
ConsigneeCountry = order.ConsigneeCountry,
ConsigneeCountryName = order.ConsigneeCountryName,
LabelUrl = labelUrl
LabelUrl = labelUrl,
OrderStatus = order.OrderStatus,
PredeclareFailReason = order.PredeclareFailReason,
ForecastBillingQueued = string.Equals(
order.PredeclareFailReason,
ForecastBillingProcessingReason,
StringComparison.Ordinal)
};
}
private ISmallBagTrackOrderNotifier SmallBagTrackOrderNotifier =>
LazyServiceProvider.GetRequiredService<ISmallBagTrackOrderNotifier>();
private IOrderParcelCreateEventPublisher ParcelCreateEventPublisher =>
LazyServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
private async Task EnqueueGenerateLabelPdfAsync(Order order)
private IOrderSideEffectEventPublisher SideEffectEventPublisher =>
LazyServiceProvider.GetRequiredService<IOrderSideEffectEventPublisher>();
/// <summary>
/// 创单/草稿预报成功后：轨迹订单+A0101 合并事件，系统面单独立事件（Fake.EventBus.RabbitMQ）。
/// </summary>
private async Task PublishParcelCreateTrackAndLabelAsync(
Order order,
Customer? customer,
OrderParcelReceiverInput? receiver,
bool appendInitialTrack,
int? operatorUserId = null)
{
try
var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiver);
request.AppendInitialTrackOnSuccess = appendInitialTrack;
var authorization = LazyServiceProvider
.GetRequiredService<IHttpContextAccessor>()
.HttpContext?
.Request
.Headers
.Authorization
.ToString();
if (!string.IsNullOrWhiteSpace(authorization))
{
var publisher = LazyServiceProvider.GetRequiredService<IOrderBatchJobPublisher>();
if (!publisher.IsEnabled)
{
// 本地未开队列：同步生成，避免 Pending 悬挂
var mapped = (await MapOrderOutputsAsync([order])).First();
await LabelPdfGenerationService.GenerateAndPersistAsync(
order,
mapped,
operatorUserId: CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0);
return;
}
request.Authorization = authorization;
}
await BatchJobCoordinator.CreateAndEnqueueAsync(
new OrderBatchJobCreateInput
{
Action = OrderBatchActionCodes.GenerateLabelPdf,
OrderIds = [order.Id],
RequestId = $"label-pdf:{order.Id}:{Guid.NewGuid():N}"
},
CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0);
}
catch (Exception)
{
// 投递失败不阻断创单；打印时会补生成
}
await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id);
await ParcelCreateEventPublisher.PublishLabelGenerateAsync(
order,
operatorUserId
?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
}
/// <summary>
/// 小包预报成功后异步同步轨迹订单（POST /ProviderService/Order/UpsertSmallBag）。
/// 入库等非创单路径仍可调用；创单路径请走 <see cref="PublishParcelCreateTrackAndLabelAsync"/>。
/// </summary>
private async Task EnqueueSmallBagTrackUpsertAsync(
Order order,
@ -2938,10 +3153,10 @@ public partial class OrderService
request.Authorization = authorization;
}
SmallBagTrackOrderNotifier.Enqueue(request, order.Id, customer?.Id);
await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id);
}
private void EnqueueLifecycleTrackAppend(Order order, string trackCode)
private async Task EnqueueLifecycleTrackAppendAsync(Order order, string trackCode)
{
if (string.IsNullOrWhiteSpace(order.ServiceNo))
{
@ -2956,10 +3171,10 @@ public partial class OrderService
.Authorization
.ToString();
SmallBagTrackOrderNotifier.EnqueueLifecycleTrackAppend(
await SideEffectEventPublisher.PublishLifecycleTrackAppendAsync(
order.Id,
order.ServiceNo,
trackCode,
order.Id,
string.IsNullOrWhiteSpace(authorization) ? null : authorization);
}

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

@ -11,6 +11,7 @@ using Oms.Application.Labeling;
using Oms.Application.Shared;
using Oms.Application.Storage;
using Newtonsoft.Json;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Application.Services.Orders.Batch;
using Oms.Domain.ChannelsAggregate.Repositories;
using Oms.Domain.CustomersAggregate.Repositories;
@ -660,21 +661,24 @@ public partial class OrderService : ApplicationService
return;
}
var publisher = LazyServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
try
{
var pickupService = LazyServiceProvider.GetRequiredService<PickupOrderService>();
var pickup = await pickupService.CreateForParcelInternalAsync(new PickupOrderCreateInput
// 优先入 Fake.EventBus；发布失败则同步回落建揽收并回写单号。
var queued = await publisher.PublishPickupAttachAsync(input, output.OrderId);
if (queued)
{
CustomerId = input.CustomerId,
SenderAddressId = input.SenderAddressId.Value,
Direction = PickupDirectionCodes.NormalizeOrDefault(input.PickupDirection),
WarehouseId = input.WarehouseId.Value,
CarrierCode = input.CarrierCode,
CarrierService = input.CarrierService,
ExpectPickupTime = input.ExpectPickupTime.Value,
OrderIds = [output.OrderId],
Remark = null
});
output.PickupQueued = true;
return;
}
var pickup = await publisher.ExecutePickupAttachAsync(input, output.OrderId);
if (pickup == null)
{
output.PickupCreateError = "揽收单创建失败";
return;
}
output.PickupNo = pickup.PickupNo;
output.PickupOrderId = pickup.Id;
output.CarrierWaybillNo = pickup.CarrierWaybillNo;

2
src/Oms.Domain/OrdersAggregate/Consts/OrderBatchJobStatusCodes.cs
View File

@ -33,6 +33,8 @@ public static class OrderBatchActionCodes
public const string Print = "print";
public const string GenerateLabelPdf = "generate-label-pdf";
public const string InboundBilling = "inbound-billing";
/// <summary>创单/草稿预报：预报应收计费 + 排货应付。</summary>
public const string ForecastBilling = "forecast-billing";
public const string Reallocate = "reallocate";
public const string Export = "export";
public const string ParcelCreate = "parcel-create";