1.03 #5

Merged
zzh merged 7 commits from test into master 2026-10-10 17:06:46 +08:00
52 changed files with 2085 additions and 242 deletions

View File

@ -1,6 +1,6 @@
<Project> <Project>
<PropertyGroup> <PropertyGroup>
<!-- All Fake packages --> <!-- All Fake packages -->
<FakePackageVersion>8.0.5.18</FakePackageVersion> <FakePackageVersion>8.0.5.19</FakePackageVersion>
</PropertyGroup> </PropertyGroup>
</Project> </Project>

View File

@ -17,7 +17,7 @@ SET @sql := IF(
`Id` INT NOT NULL AUTO_INCREMENT COMMENT ''主键'', `Id` INT NOT NULL AUTO_INCREMENT COMMENT ''主键'',
`OrderId` INT NOT NULL COMMENT ''运单Id'', `OrderId` INT NOT NULL COMMENT ''运单Id'',
`ServiceNo` VARCHAR(64) NULL COMMENT ''运单号冗余'', `ServiceNo` VARCHAR(64) NULL COMMENT ''运单号冗余'',
`Action` VARCHAR(32) NOT NULL COMMENT ''动作码 editBasic/editWeight/recalcReceivable/recalcPayable/reallocate'', `Action` VARCHAR(32) NOT NULL COMMENT ''动作码 editBasic/editWeight/recalcReceivable/recalcPayable/reallocate/bindSnapshotVersions/updateAdditionalServices/addFee/pay/preDeclare/downstreamForward/print'',
`ActionName` VARCHAR(64) NOT NULL COMMENT ''动作中文名'', `ActionName` VARCHAR(64) NOT NULL COMMENT ''动作中文名'',
`OperatorId` INT NOT NULL DEFAULT 0 COMMENT ''操作人用户Id'', `OperatorId` INT NOT NULL DEFAULT 0 COMMENT ''操作人用户Id'',
`OperatorName` VARCHAR(64) NULL COMMENT ''操作人展示名'', `OperatorName` VARCHAR(64) NULL COMMENT ''操作人展示名'',

View File

@ -35,6 +35,11 @@ public class OrderOperationLogOutput
public string? Remark { get; set; } public string? Remark { get; set; }
/// <summary>
/// 列表「内容」列:由备注 + 字段变更摘要拼装。
/// </summary>
public string Summary { get; set; } = string.Empty;
public DateTime CreateTime { get; set; } public DateTime CreateTime { get; set; }
} }

View File

@ -41,12 +41,12 @@ public class OrderParcelCreateOutput
public string? ConsigneeCountryName { get; set; } public string? ConsigneeCountryName { get; set; }
/// <summary> /// <summary>
/// 同步创建的揽收单号(上门揽收成功时)。 /// 揽收单号(上门揽收;事件总线消费较快或同步回落时可能已有值)。
/// </summary> /// </summary>
public string? PickupNo { get; set; } public string? PickupNo { get; set; }
/// <summary> /// <summary>
/// 同步创建的揽收单 Id。 /// 揽收单 Id。
/// </summary> /// </summary>
public int? PickupOrderId { get; set; } public int? PickupOrderId { get; set; }
@ -59,4 +59,24 @@ public class OrderParcelCreateOutput
/// 揽收单创建失败原因(运单已成功时仅作提示)。 /// 揽收单创建失败原因(运单已成功时仅作提示)。
/// </summary> /// </summary>
public string? PickupCreateError { get; set; } 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; }
} }

View File

@ -25,6 +25,12 @@ public class OrderUpdateInput
[StringLength(64)] [StringLength(64)]
public string? TrackNo { get; set; } public string? TrackNo { get; set; }
/// <summary>
/// 服务商单号。
/// </summary>
[StringLength(64)]
public string? DownstreamServiceNo { get; set; }
/// <summary> /// <summary>
/// 渠道编码。 /// 渠道编码。
/// </summary> /// </summary>

View File

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

View File

@ -0,0 +1,68 @@
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.UnitOfWork;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Oms.Application.Messaging;
/// <summary>
/// 可靠发布 IntegrationEvent:事务 UoW 内写入 Outbox(由 Fake.EventBus.IntegrationEventLog 投递)。
/// </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))
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Outbox enqueued {EventName} Id={EventId} (ambient TX)",
@event.GetType().Name,
@event.Id);
return;
}
// 无环境事务时开启短事务只写 Outbox,并告警:业务落库与事件非原子。
logger.LogWarning(
"Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX",
@event.GetType().Name,
@event.Id);
using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
// 先打开 Outbox DbContext 以产生可征用的 DbTransaction
var dbProvider = uow.ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
await dbProvider.GetDbContextAsync();
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;
}
}
}

View File

@ -0,0 +1,22 @@
using System.Data;
using Fake.UnitOfWork;
namespace Oms.Application.Messaging;
/// <summary>
/// 集成事件 Outbox 所需的事务性工作单元约定。
/// </summary>
public static class OmsOutboxUnitOfWork
{
/// <summary>
/// 创单/后置事件发布使用的事务 UoW 属性(MySQL ReadCommitted)。
/// </summary>
public static UnitOfWorkAttribute Transactional() => new()
{
IsTransactional = true,
IsolationLevel = IsolationLevel.ReadCommitted
};
public static bool IsAmbientTransactional(IUnitOfWorkManager unitOfWorkManager) =>
unitOfWorkManager.Current?.Context.IsTransactional == true;
}

View File

@ -29,7 +29,8 @@ public sealed class OrderBatchJobEventHandler(
return; return;
} }
using var unitOfWork = unitOfWorkManager.Begin(); // ForecastBilling 成功后继续写轨迹/面单/自动触发 Outbox,须事务性 UoW
using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
await processor.ProcessAsync(@event.ToMessage(), cancellationToken); await processor.ProcessAsync(@event.ToMessage(), cancellationToken);
await unitOfWork.CompleteAsync(cancellationToken); await unitOfWork.CompleteAsync(cancellationToken);
} }

View File

@ -1,14 +1,16 @@
using Fake.EventBus.Distributed; using Fake.EventBus.Distributed;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
namespace Oms.Application.Messaging; namespace Oms.Application.Messaging;
/// <summary> /// <summary>
/// 通过 Fake 分布式事件总线发布订单批量任务。 /// 通过 Fake 分布式事件总线(Outbox)发布订单批量任务。
/// </summary> /// </summary>
public sealed class OrderBatchJobPublisher( public sealed class OrderBatchJobPublisher(
IDistributedEventBus eventBus, IDistributedEventBus eventBus,
IUnitOfWorkManager unitOfWorkManager,
IOptions<OrderBatchRabbitMqOptions> options, IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobPublisher> logger) : IOrderBatchJobPublisher ILogger<OrderBatchJobPublisher> logger) : IOrderBatchJobPublisher
{ {
@ -22,10 +24,11 @@ public sealed class OrderBatchJobPublisher(
} }
var @event = OrderBatchJobIntegrationEvent.FromMessage(message); var @event = OrderBatchJobIntegrationEvent.FromMessage(message);
await eventBus.PublishAsync(@event, cancellationToken); await OmsIntegrationEventPublisher.PublishAsync(
eventBus, unitOfWorkManager, logger, @event, cancellationToken);
logger.LogDebug( logger.LogDebug(
"Published order batch event {EventId} messageId={MessageId} action={Action}", "Outbox enqueued order batch event {EventId} messageId={MessageId} action={Action}",
@event.Id, @event.Id,
message.MessageId, message.MessageId,
message.Action); message.Action);

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

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

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

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

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

View File

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

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

View File

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

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

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

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

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

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

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

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

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

View File

@ -18,6 +18,7 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.EventBus.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectStorage" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.ObjectStorage" Version="$(FakePackageVersion)" />

View File

@ -1,6 +1,8 @@
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
using Oms.Application.Messaging; using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Messaging.ParcelCreate;
namespace Oms.Application; namespace Oms.Application;
@ -11,6 +13,8 @@ public static class OrderBatchMessagingServiceExtensions
services.Configure<OrderBatchRabbitMqOptions>(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName)); services.Configure<OrderBatchRabbitMqOptions>(configuration.GetSection(OrderBatchRabbitMqOptions.SectionName));
services.AddSingleton<IOrderBatchJobPublisher, OrderBatchJobPublisher>(); services.AddSingleton<IOrderBatchJobPublisher, OrderBatchJobPublisher>();
services.AddSingleton<IOrderBatchQueueInspector, OrderBatchQueueInspector>(); services.AddSingleton<IOrderBatchQueueInspector, OrderBatchQueueInspector>();
services.AddSingleton<IOrderParcelCreateEventPublisher, OrderParcelCreateEventPublisher>();
services.AddSingleton<IOrderSideEffectEventPublisher, OrderSideEffectEventPublisher>();
// 消费端由 Fake.EventBus.RabbitMQ(RabbitMqEventBus HostedService)托管 // 消费端由 Fake.EventBus.RabbitMQ(RabbitMqEventBus HostedService)托管
services.AddHostedService<OrderBatchDlxMonitorHostedService>(); services.AddHostedService<OrderBatchDlxMonitorHostedService>();
return services; return services;

View File

@ -187,6 +187,8 @@ public class UserProfileRepository(FakeRbacDbContext dbContext) : IScopedDepende
if (int.TryParse(match.Groups[1].Value, out var number)) if (int.TryParse(match.Groups[1].Value, out var number))
{ {
max = Math.Max(max, number); max = Math.Max(max, number);
} }
} }

View File

@ -1,4 +1,6 @@
using System.Data;
using Fake.Application; using Fake.Application;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Operations;
using Oms.Application.Dtos.Shared; using Oms.Application.Dtos.Shared;
@ -85,6 +87,7 @@ public class InboundRecordService : ApplicationService
/// <param name="input">入参:入库记录创建参数。</param> /// <param name="input">入参:入库记录创建参数。</param>
/// <returns>入库记录结果。</returns> /// <returns>入库记录结果。</returns>
[HttpPost("/operation/inbound/create")] [HttpPost("/operation/inbound/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<InboundRecordOutput> CreateAsync(CreateInboundRecordInput input) public async Task<InboundRecordOutput> CreateAsync(CreateInboundRecordInput input)
{ {
if (string.IsNullOrWhiteSpace(input.WaybillNo)) if (string.IsNullOrWhiteSpace(input.WaybillNo))
@ -141,8 +144,8 @@ public class InboundRecordService : ApplicationService
var savedEntity = await InboundRecordRepository.InsertAsync(entity); var savedEntity = await InboundRecordRepository.InsertAsync(entity);
// 自动触发改异步,避免拉长扫码 RT // 自动触发走 Outbox,避免拉长扫码 RT
AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Inbound, operatorId); await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Inbound, operatorId);
var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim()); var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim());
var output = ObjectMapper.Map<InboundRecord, InboundRecordOutput>(savedEntity); var output = ObjectMapper.Map<InboundRecord, InboundRecordOutput>(savedEntity);

View File

@ -1,4 +1,6 @@
using System.Data;
using Fake.Application; using Fake.Application;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Operations;
using Oms.Application.Dtos.Shared; using Oms.Application.Dtos.Shared;
@ -99,6 +101,7 @@ public class OutboundRecordService : ApplicationService
/// <param name="input">入参:出库记录创建参数。</param> /// <param name="input">入参:出库记录创建参数。</param>
/// <returns>出库记录结果。</returns> /// <returns>出库记录结果。</returns>
[HttpPost("/operation/outbound/create")] [HttpPost("/operation/outbound/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OutboundRecordOutput> CreateAsync(CreateOutboundRecordInput input) public async Task<OutboundRecordOutput> CreateAsync(CreateOutboundRecordInput input)
{ {
if (string.IsNullOrWhiteSpace(input.WaybillNo)) if (string.IsNullOrWhiteSpace(input.WaybillNo))
@ -161,18 +164,8 @@ public class OutboundRecordService : ApplicationService
var savedEntity = await OutboundRecordRepository.InsertAsync(entity); var savedEntity = await OutboundRecordRepository.InsertAsync(entity);
try // 自动触发走 Outbox,避免拉长出库扫码 RT
{ await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Outbound, operatorId);
var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim());
if (refreshed != null)
{
await AutoTriggerRuntimeService.FireAsync(refreshed, AutoTriggerTiming.Outbound, operatorId);
}
}
catch
{
// 自动触发失败不影响出库主流程
}
return ObjectMapper.Map<OutboundRecord, OutboundRecordOutput>(savedEntity); return ObjectMapper.Map<OutboundRecord, OutboundRecordOutput>(savedEntity);
} }

View File

@ -26,6 +26,7 @@ public sealed class OrderBatchActionExecutor(
OrderDownstreamCredentialResolver downstreamCredentialResolver, OrderDownstreamCredentialResolver downstreamCredentialResolver,
IItfxForwardClient itfxForwardClient, IItfxForwardClient itfxForwardClient,
OrderFileManager orderFileManager, OrderFileManager orderFileManager,
OrderOperationLogWriter operationLogWriter,
OrderBatchCreateActionHandler createActionHandler, OrderBatchCreateActionHandler createActionHandler,
OrderBatchOrderActionHandler orderActionHandler) : IOrderBatchActionExecutor, IScopedDependency OrderBatchOrderActionHandler orderActionHandler) : IOrderBatchActionExecutor, IScopedDependency
{ {
@ -49,6 +50,7 @@ public sealed class OrderBatchActionExecutor(
OrderBatchActionCodes.Print => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.Print => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.GenerateLabelPdf => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.GenerateLabelPdf => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.InboundBilling => 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.Reallocate => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.MarkProblematic => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.MarkProblematic => await orderActionHandler.ExecuteAsync(message, cancellationToken),
OrderBatchActionCodes.MarkClaimed => await orderActionHandler.ExecuteAsync(message, cancellationToken), OrderBatchActionCodes.MarkClaimed => await orderActionHandler.ExecuteAsync(message, cancellationToken),
@ -224,6 +226,8 @@ public sealed class OrderBatchActionExecutor(
businessMessage)); businessMessage));
} }
var beforeSnap = OrderOperationLogSnapshot.CaptureOrder(order);
if (ItfxCreateOrderOneDataParser.TryParse(forward.ResponseBody, out var data) if (ItfxCreateOrderOneDataParser.TryParse(forward.ResponseBody, out var data)
&& data != null) && data != null)
{ {
@ -242,6 +246,29 @@ public sealed class OrderBatchActionExecutor(
} }
} }
var afterSnap = OrderOperationLogSnapshot.CaptureOrder(order);
var remarkParts = new List<string> { $"供应商 {creds.SupplierCode}" };
if (!string.IsNullOrWhiteSpace(order.DownstreamServiceNo))
{
remarkParts.Add($"服务商单号 {order.DownstreamServiceNo}");
}
if (!string.IsNullOrWhiteSpace(order.TrackNo))
{
remarkParts.Add($"跟踪号 {order.TrackNo}");
}
await operationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.DownstreamForward,
beforeSnap,
afterSnap,
string.Join(";", remarkParts),
message.OperatorUserId > 0 ? message.OperatorUserId : null,
operatorName: null,
operatorType: "staff",
source: "job");
return OrderBatchActionExecuteResult.Ok( return OrderBatchActionExecuteResult.Ok(
serviceNo: order.ServiceNo, serviceNo: order.ServiceNo,
trackNo: order.TrackNo, trackNo: order.TrackNo,

View File

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

View File

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

View File

@ -1,6 +1,6 @@
using Fake.DependencyInjection; using Fake.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Domain.ChannelsAggregate.Consts; using Oms.Domain.ChannelsAggregate.Consts;
using Oms.Domain.ChannelsAggregate.Entities; using Oms.Domain.ChannelsAggregate.Entities;
using Oms.Domain.ChannelsAggregate.Repositories; using Oms.Domain.ChannelsAggregate.Repositories;
@ -16,39 +16,18 @@ public class AutoTriggerRuntimeService(
OrderAutoTriggerService orderAutoTriggerService, OrderAutoTriggerService orderAutoTriggerService,
OrderExceptionWorkflowService exceptionWorkflowService, OrderExceptionWorkflowService exceptionWorkflowService,
OrderIssueNotificationService issueNotificationService, OrderIssueNotificationService issueNotificationService,
IServiceScopeFactory scopeFactory, IOrderSideEffectEventPublisher sideEffectEventPublisher,
ILogger<AutoTriggerRuntimeService> logger) : IScopedDependency ILogger<AutoTriggerRuntimeService> logger) : IScopedDependency
{ {
private const string InterceptOperation = "拦截"; private const string InterceptOperation = "拦截";
/// <summary> /// <summary>
/// 后台异步触发,避免拉长入库/出库扫码请求。 /// 投递自动触发事件(须在事务 UoW 内调用,写入 Outbox)。
/// </summary> /// </summary>
public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId) public Task EnqueueFireAsync(int orderId, AutoTriggerTiming timing, int operatorUserId) =>
{ orderId <= 0
if (orderId <= 0) ? Task.CompletedTask
{ : sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId);
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);
}
});
}
public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId) public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId)
{ {

View File

@ -1,4 +1,6 @@
using System.Globalization; using System.Globalization;
using Newtonsoft.Json;
using Oms.Application.Shared;
using Oms.Domain.OrdersAggregate.Consts; using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities; using Oms.Domain.OrdersAggregate.Entities;
@ -9,6 +11,8 @@ namespace Oms.Application.Services.Orders;
/// </summary> /// </summary>
public static class OrderOperationLogSnapshot public static class OrderOperationLogSnapshot
{ {
public const int SummaryMaxChangeItems = 5;
private static readonly Dictionary<string, string> Labels = new(StringComparer.OrdinalIgnoreCase) private static readonly Dictionary<string, string> Labels = new(StringComparer.OrdinalIgnoreCase)
{ {
["orderStatus"] = "订单状态", ["orderStatus"] = "订单状态",
@ -31,22 +35,39 @@ public static class OrderOperationLogSnapshot
["salesProductRulesVersion"] = "规则版本", ["salesProductRulesVersion"] = "规则版本",
["salesProductPriceVersion"] = "价格版本", ["salesProductPriceVersion"] = "价格版本",
["isPaid"] = "是否已支付", ["isPaid"] = "是否已支付",
["additionalServices"] = "附加服务",
["sender.name"] = "发件人姓名", ["sender.name"] = "发件人姓名",
["sender.company"] = "发件公司",
["sender.phone"] = "发件人电话", ["sender.phone"] = "发件人电话",
["sender.telePhone"] = "发件人固话",
["sender.email"] = "发件人邮箱",
["sender.country"] = "发件国家", ["sender.country"] = "发件国家",
["sender.province"] = "发件省/州", ["sender.province"] = "发件省/州",
["sender.city"] = "发件城市", ["sender.city"] = "发件城市",
["sender.zipCode"] = "发件邮编", ["sender.zipCode"] = "发件邮编",
["sender.address1"] = "发件地址1", ["sender.address1"] = "发件地址1",
["sender.address2"] = "发件地址2", ["sender.address2"] = "发件地址2",
["sender.address3"] = "发件地址3",
["sender.houseNumber"] = "发件门牌号",
["receiver.name"] = "收件人姓名", ["receiver.name"] = "收件人姓名",
["receiver.company"] = "收件公司",
["receiver.phone"] = "收件人电话", ["receiver.phone"] = "收件人电话",
["receiver.telePhone"] = "收件人固话",
["receiver.email"] = "收件人邮箱",
["receiver.country"] = "收件国家", ["receiver.country"] = "收件国家",
["receiver.province"] = "收件省/州", ["receiver.province"] = "收件省/州",
["receiver.city"] = "收件城市", ["receiver.city"] = "收件城市",
["receiver.zipCode"] = "收件邮编", ["receiver.zipCode"] = "收件邮编",
["receiver.streetNumber"] = "收件街道号",
["receiver.address1"] = "收件地址1", ["receiver.address1"] = "收件地址1",
["receiver.address2"] = "收件地址2", ["receiver.address2"] = "收件地址2",
["receiver.address3"] = "收件地址3",
["receiver.taxType"] = "收件税号类型",
["receiver.taxNo"] = "收件税号",
["receiver.certificateType"] = "收件证件类型",
["receiver.certificateNo"] = "收件证件号",
["receiver.eori"] = "收件EORI",
["receiver.vat"] = "收件VAT",
["fee.receivableTotal"] = "应收合计", ["fee.receivableTotal"] = "应收合计",
["fee.payableTotal"] = "应付合计", ["fee.payableTotal"] = "应付合计",
}; };
@ -75,6 +96,7 @@ public static class OrderOperationLogSnapshot
["salesProductRulesVersion"] = Norm(order.SalesProductRulesVersion), ["salesProductRulesVersion"] = Norm(order.SalesProductRulesVersion),
["salesProductPriceVersion"] = Norm(order.SalesProductPriceVersion), ["salesProductPriceVersion"] = Norm(order.SalesProductPriceVersion),
["isPaid"] = FormatByte(order.IsPaid), ["isPaid"] = FormatByte(order.IsPaid),
["additionalServices"] = FormatAdditionalServices(order.AdditionalServices),
}; };
} }
@ -88,13 +110,18 @@ public static class OrderOperationLogSnapshot
return new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase) return new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase)
{ {
["sender.name"] = Norm(sender.SenderName), ["sender.name"] = Norm(sender.SenderName),
["sender.company"] = Norm(sender.Company),
["sender.phone"] = Norm(sender.Phone), ["sender.phone"] = Norm(sender.Phone),
["sender.telePhone"] = Norm(sender.TelePhone),
["sender.email"] = Norm(sender.Email),
["sender.country"] = FormatInt(sender.Country), ["sender.country"] = FormatInt(sender.Country),
["sender.province"] = Norm(sender.Province), ["sender.province"] = Norm(sender.Province),
["sender.city"] = Norm(sender.City), ["sender.city"] = Norm(sender.City),
["sender.zipCode"] = Norm(sender.ZipCode), ["sender.zipCode"] = Norm(sender.ZipCode),
["sender.address1"] = Norm(sender.StreetAddress1), ["sender.address1"] = Norm(sender.StreetAddress1),
["sender.address2"] = Norm(sender.StreetAddress2), ["sender.address2"] = Norm(sender.StreetAddress2),
["sender.address3"] = Norm(sender.StreetAddress3),
["sender.houseNumber"] = Norm(sender.HouseNumber),
}; };
} }
@ -108,13 +135,24 @@ public static class OrderOperationLogSnapshot
return new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase) return new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase)
{ {
["receiver.name"] = Norm(recipient.RecipientsName), ["receiver.name"] = Norm(recipient.RecipientsName),
["receiver.company"] = Norm(recipient.CompanyName),
["receiver.phone"] = Norm(recipient.Phone), ["receiver.phone"] = Norm(recipient.Phone),
["receiver.telePhone"] = Norm(recipient.TelePhone),
["receiver.email"] = Norm(recipient.Email),
["receiver.country"] = FormatInt(recipient.Country), ["receiver.country"] = FormatInt(recipient.Country),
["receiver.province"] = Norm(recipient.Province), ["receiver.province"] = Norm(recipient.Province),
["receiver.city"] = Norm(recipient.City), ["receiver.city"] = Norm(recipient.City),
["receiver.zipCode"] = Norm(recipient.ZipCode), ["receiver.zipCode"] = Norm(recipient.ZipCode),
["receiver.streetNumber"] = Norm(recipient.StreetNumber),
["receiver.address1"] = Norm(recipient.StreetAddress1), ["receiver.address1"] = Norm(recipient.StreetAddress1),
["receiver.address2"] = Norm(recipient.StreetAddress2), ["receiver.address2"] = Norm(recipient.StreetAddress2),
["receiver.address3"] = Norm(recipient.StreetAddress3),
["receiver.taxType"] = FormatInt(recipient.TaxType),
["receiver.taxNo"] = Norm(recipient.TaxNo),
["receiver.certificateType"] = FormatInt(recipient.CertificateType),
["receiver.certificateNo"] = Norm(recipient.CertificateNo),
["receiver.eori"] = Norm(recipient.Eori),
["receiver.vat"] = Norm(recipient.Vat),
}; };
} }
@ -225,6 +263,89 @@ public static class OrderOperationLogSnapshot
return path; return path;
} }
/// <summary>
/// 拼装列表「内容」列:备注优先,再跟最多 N 条字段变更。
/// </summary>
public static string BuildSummary(
IReadOnlyList<OrderOperationLogChangeView> changes,
string? remark,
int maxItems = SummaryMaxChangeItems)
{
var parts = new List<string>();
if (!string.IsNullOrWhiteSpace(remark))
{
parts.Add(remark.Trim());
}
if (changes.Count > 0)
{
var take = Math.Max(0, maxItems);
foreach (var change in changes.Take(take))
{
var label = string.IsNullOrWhiteSpace(change.Label) ? change.Path : change.Label;
parts.Add($"{label}:{DisplayValue(change.Path, change.Before)}→{DisplayValue(change.Path, change.After)}");
}
if (changes.Count > take)
{
parts.Add($"等{changes.Count}项变更");
}
}
return parts.Count == 0 ? "—" : string.Join(";", parts);
}
public readonly record struct OrderOperationLogChangeView(
string Path,
string Label,
string? Before,
string? After);
private static string DisplayValue(string path, string? value)
{
if (string.IsNullOrWhiteSpace(value))
{
return "空";
}
if (string.Equals(path, "isPaid", StringComparison.OrdinalIgnoreCase)
|| string.Equals(path, "isCod", StringComparison.OrdinalIgnoreCase))
{
return value == "1" ? "是" : "否";
}
if (string.Equals(path, "orderStatus", StringComparison.OrdinalIgnoreCase))
{
return OmsDocHelper.GetOrderStatusName(value) ?? value;
}
return value;
}
private static string? FormatAdditionalServices(string? json)
{
if (string.IsNullOrWhiteSpace(json))
{
return null;
}
try
{
var list = JsonConvert.DeserializeObject<List<string>>(json) ?? [];
var codes = list
.Where(x => !string.IsNullOrWhiteSpace(x))
.Select(x => x.Trim())
.Distinct(StringComparer.OrdinalIgnoreCase)
.OrderBy(x => x, StringComparer.OrdinalIgnoreCase)
.ToList();
return codes.Count == 0 ? null : string.Join("、", codes);
}
catch
{
return Norm(json);
}
}
private static string? Norm(string? value) private static string? Norm(string? value)
{ {
if (string.IsNullOrWhiteSpace(value)) if (string.IsNullOrWhiteSpace(value))

View File

@ -33,23 +33,29 @@ public class OrderOperationLogWriter : IScopedDependency
_logger = logger; _logger = logger;
} }
public async Task TryWriteAsync( public Task TryWriteAsync(
Order order, Order order,
string action, string action,
IReadOnlyDictionary<string, string?> before, IReadOnlyDictionary<string, string?> before,
IReadOnlyDictionary<string, string?> after, IReadOnlyDictionary<string, string?> after,
string? remark = null) string? remark = null)
=> TryWriteAsync(order, action, before, after, remark, operatorId: null, operatorName: null);
public async Task TryWriteAsync(
Order order,
string action,
IReadOnlyDictionary<string, string?> before,
IReadOnlyDictionary<string, string?> after,
string? remark,
int? operatorId,
string? operatorName,
string operatorType = "staff",
string source = "pc")
{ {
try try
{ {
var changes = OrderOperationLogSnapshot.Diff(before, after); var changes = OrderOperationLogSnapshot.Diff(before, after);
var operatorId = 0; var (resolvedId, resolvedName) = await ResolveOperatorAsync(operatorId, operatorName);
string? operatorName = null;
if (_currentUser.IsAuthenticated)
{
operatorId = _currentUser.GetId<int>();
operatorName = await _userDisplayNameResolver.ResolveAsync(operatorId);
}
var entity = new OrderOperationLog var entity = new OrderOperationLog
{ {
@ -57,10 +63,10 @@ public class OrderOperationLogWriter : IScopedDependency
ServiceNo = order.ServiceNo, ServiceNo = order.ServiceNo,
Action = action, Action = action,
ActionName = OrderOperationActionCodes.GetActionName(action), ActionName = OrderOperationActionCodes.GetActionName(action),
OperatorId = operatorId, OperatorId = resolvedId,
OperatorName = operatorName, OperatorName = resolvedName,
OperatorType = "staff", OperatorType = string.IsNullOrWhiteSpace(operatorType) ? "staff" : operatorType.Trim(),
Source = "pc", Source = string.IsNullOrWhiteSpace(source) ? "pc" : source.Trim(),
ChangesJson = JsonConvert.SerializeObject( ChangesJson = JsonConvert.SerializeObject(
changes.Select(x => new OrderOperationLogChangeItem changes.Select(x => new OrderOperationLogChangeItem
{ {
@ -70,7 +76,7 @@ public class OrderOperationLogWriter : IScopedDependency
After = x.After After = x.After
}).ToList()), }).ToList()),
Remark = string.IsNullOrWhiteSpace(remark) ? null : remark.Trim(), Remark = string.IsNullOrWhiteSpace(remark) ? null : remark.Trim(),
CreateUserId = operatorId, CreateUserId = resolvedId,
CreateTime = DateTime.Now CreateTime = DateTime.Now
}; };
@ -86,6 +92,28 @@ public class OrderOperationLogWriter : IScopedDependency
} }
} }
/// <summary>
/// 无字段 diff 时写入叙事日志(打印、下游预报成功等)。
/// </summary>
public Task TryWriteSimpleAsync(
Order order,
string action,
string? remark,
int? operatorId = null,
string? operatorName = null,
string operatorType = "staff",
string source = "pc")
=> TryWriteAsync(
order,
action,
new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase),
new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase),
remark,
operatorId,
operatorName,
operatorType,
source);
public async Task<OrderOperationLogListOutput> ListByOrderIdAsync(int orderId) public async Task<OrderOperationLogListOutput> ListByOrderIdAsync(int orderId)
{ {
var rows = await _repository.GetListAsync(x => x.OrderId == orderId); var rows = await _repository.GetListAsync(x => x.OrderId == orderId);
@ -97,6 +125,30 @@ public class OrderOperationLogWriter : IScopedDependency
return new OrderOperationLogListOutput { Items = items }; return new OrderOperationLogListOutput { Items = items };
} }
private async Task<(int OperatorId, string? OperatorName)> ResolveOperatorAsync(
int? operatorId,
string? operatorName)
{
if (operatorId is > 0)
{
var name = string.IsNullOrWhiteSpace(operatorName)
? await _userDisplayNameResolver.ResolveAsync(operatorId.Value)
: operatorName.Trim();
return (operatorId.Value, name);
}
if (_currentUser.IsAuthenticated)
{
var id = _currentUser.GetId<int>();
var name = string.IsNullOrWhiteSpace(operatorName)
? await _userDisplayNameResolver.ResolveAsync(id)
: operatorName.Trim();
return (id, name);
}
return (0, string.IsNullOrWhiteSpace(operatorName) ? null : operatorName.Trim());
}
private static OrderOperationLogOutput MapOutput(OrderOperationLog row) private static OrderOperationLogOutput MapOutput(OrderOperationLog row)
{ {
var changes = new List<OrderOperationLogChangeItem>(); var changes = new List<OrderOperationLogChangeItem>();
@ -113,6 +165,14 @@ public class OrderOperationLogWriter : IScopedDependency
} }
} }
var views = changes
.Select(x => new OrderOperationLogSnapshot.OrderOperationLogChangeView(
x.Path,
x.Label,
x.Before,
x.After))
.ToList();
return new OrderOperationLogOutput return new OrderOperationLogOutput
{ {
Id = row.Id, Id = row.Id,
@ -126,6 +186,7 @@ public class OrderOperationLogWriter : IScopedDependency
Source = row.Source, Source = row.Source,
Changes = changes, Changes = changes,
Remark = row.Remark, Remark = row.Remark,
Summary = OrderOperationLogSnapshot.BuildSummary(views, row.Remark),
CreateTime = row.CreateTime CreateTime = row.CreateTime
}; };
} }

View File

@ -1,3 +1,4 @@
using System.Globalization;
using Fake.Application; using Fake.Application;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using Newtonsoft.Json; using Newtonsoft.Json;
@ -52,6 +53,13 @@ public partial class OrderService
order.TrackNo = string.IsNullOrWhiteSpace(input.TrackNo) ? null : input.TrackNo.Trim(); order.TrackNo = string.IsNullOrWhiteSpace(input.TrackNo) ? null : input.TrackNo.Trim();
} }
if (input.DownstreamServiceNo != null)
{
order.DownstreamServiceNo = string.IsNullOrWhiteSpace(input.DownstreamServiceNo)
? null
: input.DownstreamServiceNo.Trim();
}
if (input.ChannelCode != null) if (input.ChannelCode != null)
{ {
order.ChannelCode = string.IsNullOrWhiteSpace(input.ChannelCode) ? null : input.ChannelCode.Trim(); order.ChannelCode = string.IsNullOrWhiteSpace(input.ChannelCode) ? null : input.ChannelCode.Trim();
@ -137,7 +145,9 @@ public partial class OrderService
order = await GetOrderOrThrowAsync(order.Id); order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeParties: true); var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeParties: true);
var remark = input.Items != null ? "申报信息已更新" : null; // 前端保存基本信息时总会带 items;仅在无字段级 diff 时用备注兜底,避免掩盖真实字段变更
var fieldChanges = OrderOperationLogSnapshot.Diff(beforeSnap, afterSnap);
var remark = input.Items != null && fieldChanges.Count == 0 ? "申报信息已更新" : null;
await OperationLogWriter.TryWriteAsync( await OperationLogWriter.TryWriteAsync(
order, order,
OrderOperationActionCodes.EditBasic, OrderOperationActionCodes.EditBasic,
@ -323,7 +333,7 @@ public partial class OrderService
"规则"); "规则");
if (!string.Equals(order.SalesProductRulesVersion, resolved, StringComparison.Ordinal)) if (!string.Equals(order.SalesProductRulesVersion, resolved, StringComparison.Ordinal))
{ {
remarkParts.Add($"rules:{order.SalesProductRulesVersion ?? "-"}→{resolved}"); remarkParts.Add($"规则版本 {order.SalesProductRulesVersion ?? "-"}→{resolved}");
order.SalesProductRulesVersion = resolved; order.SalesProductRulesVersion = resolved;
} }
} }
@ -337,7 +347,7 @@ public partial class OrderService
"价格"); "价格");
if (!string.Equals(order.SalesProductPriceVersion, resolved, StringComparison.Ordinal)) if (!string.Equals(order.SalesProductPriceVersion, resolved, StringComparison.Ordinal))
{ {
remarkParts.Add($"price:{order.SalesProductPriceVersion ?? "-"}→{resolved}"); remarkParts.Add($"价格版本 {order.SalesProductPriceVersion ?? "-"}→{resolved}");
order.SalesProductPriceVersion = resolved; order.SalesProductPriceVersion = resolved;
} }
} }
@ -354,11 +364,13 @@ public partial class OrderService
{ {
EnsureReceivableMutable(order); EnsureReceivableMutable(order);
await RecalculateReceivableCoreAsync(order); await RecalculateReceivableCoreAsync(order);
remarkParts.Add("并重算应收");
} }
if (input.RecalculatePayable) if (input.RecalculatePayable)
{ {
await RecalculatePayableCoreAsync(order); await RecalculatePayableCoreAsync(order);
remarkParts.Add("并重算应付");
} }
order = await GetOrderOrThrowAsync(order.Id); order = await GetOrderOrThrowAsync(order.Id);
@ -368,7 +380,7 @@ public partial class OrderService
OrderOperationActionCodes.BindSnapshotVersions, OrderOperationActionCodes.BindSnapshotVersions,
beforeSnap, beforeSnap,
afterSnap, afterSnap,
remarkParts.Count == 0 ? null : string.Join(';', remarkParts)); remarkParts.Count == 0 ? null : string.Join(';', remarkParts));
return (await MapOrderOutputsAsync([order])).First(); return (await MapOrderOutputsAsync([order])).First();
} }
@ -785,6 +797,7 @@ public partial class OrderService
{ {
var order = await GetOrderOrThrowAsync(input.Id); var order = await GetOrderOrThrowAsync(input.Id);
EnsureReceivableMutable(order); EnsureReceivableMutable(order);
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true);
if (string.IsNullOrWhiteSpace(order.ProductCode)) if (string.IsNullOrWhiteSpace(order.ProductCode))
{ {
@ -803,6 +816,15 @@ public partial class OrderService
order.UpdateTime = DateTime.Now; order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.UpdateAdditionalServices,
beforeSnap,
afterSnap);
return (await MapOrderOutputsAsync([order])).First(); return (await MapOrderOutputsAsync([order])).First();
} }
@ -892,6 +914,8 @@ public partial class OrderService
throw new BusinessException("计费阶段无效"); throw new BusinessException("计费阶段无效");
} }
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true);
var fee = new OrderFee var fee = new OrderFee
{ {
Type = type, Type = type,
@ -912,6 +936,17 @@ public partial class OrderService
order.UpdateTime = DateTime.Now; order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true);
var typeName = type == OrderFeeTypeCodes.Receivable ? "应收" : "应付";
var amountText = input.Amount.ToString("0.####", CultureInfo.InvariantCulture);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.AddFee,
beforeSnap,
afterSnap,
$"新增{typeName}「{costItem}」{currency} {amountText}");
return (await MapOrderOutputsAsync([order])).First(); return (await MapOrderOutputsAsync([order])).First();
} }

View File

@ -9,6 +9,8 @@ using Oms.Application.Dtos.Orders;
using Oms.Application.Integrations.SmallBagTrack; using Oms.Application.Integrations.SmallBagTrack;
using Oms.Application.Labeling; using Oms.Application.Labeling;
using Oms.Application.Messaging; using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Application.Shared; using Oms.Application.Shared;
using Oms.Application.Storage; using Oms.Application.Storage;
using Oms.Application.Services.Orders.Batch; using Oms.Application.Services.Orders.Batch;
@ -36,6 +38,8 @@ public partial class OrderService
private const int ParcelBatchMaxCount = 2000; private const int ParcelBatchMaxCount = 2000;
private const string InboundBillingProcessingReason = "入库后计费/排货处理中"; private const string InboundBillingProcessingReason = "入库后计费/排货处理中";
private const string InboundBillingEnqueueFailReason = "入库计费任务投递失败,请重跑入库计费"; private const string InboundBillingEnqueueFailReason = "入库计费任务投递失败,请重跑入库计费";
private const string ForecastBillingProcessingReason = "预报计费/排货处理中";
private const string ForecastBillingEnqueueFailReason = "预报计费任务投递失败,请重新预报";
private static readonly ConcurrentDictionary<int, byte> InboundBillingInFlight = new(); private static readonly ConcurrentDictionary<int, byte> InboundBillingInFlight = new();
@ -143,10 +147,11 @@ public partial class OrderService
// 客户主动「先保存为草稿」:落库草稿后直接返回,不跑预报计费。 // 客户主动「先保存为草稿」:落库草稿后直接返回,不跑预报计费。
if (input.SaveAsDraft) if (input.SaveAsDraft)
{ {
await TryUpsertCustoms9610Async(order, input.PlatformCode); await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode);
return await BuildParcelCreateOutputAsync(order); return await BuildParcelCreateOutputAsync(order);
} }
// 偏远标记轻量同步;计费重/应收/排货走 ForecastBilling MQ(失败不回滚建单)。
order.IsRemote = await IsRemoteHitAsync( order.IsRemote = await IsRemoteHitAsync(
product, product,
snapshotBinding.Rules, snapshotBinding.Rules,
@ -154,62 +159,24 @@ public partial class OrderService
input.Receiver?.Province, input.Receiver?.Province,
input.Receiver?.City, input.Receiver?.City,
input.Receiver?.ZipCode) ? (byte)1 : (byte)0; input.Receiver?.ZipCode) ? (byte)1 : (byte)0;
order.OrderStatus = OrderStatusCodes.Draft;
var forecast = weights.First(x => x.Type == OrderWeightTypeCodes.Forecast); order.PredeclareFailReason = ForecastBillingProcessingReason;
await ApplyChargeableWeightAsync(product, snapshotBinding.Rules, forecast); order.AllocationPendingReason = string.Empty;
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.UpdateTime = DateTime.Now; order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
await EnqueueSmallBagTrackUpsertAsync(order, customer, input.Receiver, appendInitialTrack: true); await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, input.PlatformCode);
await EnqueueGenerateLabelPdfAsync(order); await EnqueueForecastBillingAsync(
order,
input.CouponInstanceId,
operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
try var output = await BuildParcelCreateOutputAsync(order);
{ output.ForecastBillingQueued = string.Equals(
var triggerUserId = operatorUserId ?? CurrentUser.GetId<int>(); order.PredeclareFailReason,
await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId); ForecastBillingProcessingReason,
} StringComparison.Ordinal);
catch return output;
{
// 自动触发失败不影响预报主流程
}
return await BuildParcelCreateOutputAsync(order);
} }
catch (Exception ex) catch (Exception ex)
{ {
@ -2287,8 +2254,7 @@ public partial class OrderService
} }
/// <summary> /// <summary>
/// 草稿重新预报:按订单当前收件/件重尺/快照重跑黑名单、计重、分区与预报应收。 /// 草稿重新预报:同步做黑名单等硬校验,计重/应收/排货投递 ForecastBilling MQ。
/// 失败时保持草稿并写入 PredeclareFailReason。
/// </summary> /// </summary>
internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null) internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null)
{ {
@ -2319,63 +2285,52 @@ public partial class OrderService
throw new BusinessException("收件地址命中黑名单区域"); throw new BusinessException("收件地址命中黑名单区域");
} }
order.IsRemote = await IsRemoteHitAsync( var weights = await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]);
product, if (!weights.Any(x =>
snapshotBinding.Rules, string.Equals(x.Type, OrderWeightTypeCodes.Forecast, StringComparison.OrdinalIgnoreCase)))
address.CountryCode, {
address.Province, throw new BusinessException("订单缺少预报件重尺");
address.City, }
address.ZipCode) ? (byte)1 : (byte)0;
var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) order.OrderStatus = OrderStatusCodes.Draft;
.OrderBy(x => x.Id) order.PredeclareFailReason = ForecastBillingProcessingReason;
.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.AllocationPendingReason = string.Empty; order.AllocationPendingReason = string.Empty;
order.UpdateTime = DateTime.Now; order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
var customer = order.CustomerId.HasValue await EnqueueForecastBillingAsync(
? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) order,
: null; order.CouponInstanceId,
await EnqueueSmallBagTrackUpsertAsync(order, customer, receiver: null, appendInitialTrack: true); operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
await EnqueueGenerateLabelPdfAsync(order);
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>(); throw CreateDraftSavedBusinessException(
await AutoTriggerRuntimeService.FireAsync(order, AutoTriggerTiming.OrderCreate, triggerUserId); refreshed,
} StripDraftSavedMarker(refreshed.PredeclareFailReason));
catch
{
// 自动触发失败不影响预报主流程
} }
} }
catch (BusinessException ex) catch (BusinessException ex)
{ {
if (string.Equals(ex.ErrorCode, OrderParcelDraftFailData.ErrorCodeValue, StringComparison.Ordinal))
{
throw;
}
var cleanReason = StripDraftSavedMarker(ex.Message); var cleanReason = StripDraftSavedMarker(ex.Message);
await MarkParcelDraftFailedAsync(order, cleanReason); if (!string.Equals(cleanReason, ForecastBillingProcessingReason, StringComparison.Ordinal))
{
await MarkParcelDraftFailedAsync(order, cleanReason);
}
throw CreateDraftSavedBusinessException(order, cleanReason); throw CreateDraftSavedBusinessException(order, cleanReason);
} }
catch (Exception ex) catch (Exception ex)
@ -2427,6 +2382,7 @@ public partial class OrderService
try try
{ {
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true);
var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]))
.OrderBy(x => x.Id) .OrderBy(x => x.Id)
.ToList(); .ToList();
@ -2464,7 +2420,17 @@ public partial class OrderService
order.UpdateTime = DateTime.Now; order.UpdateTime = DateTime.Now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.InboundTrackCode); // 操作日志同步落库(件重尺已写完);计费/排货仍走 MQ,不把审计塞进队列
order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.Inbound,
beforeSnap,
afterSnap,
FormatMeasuredDimensionsRemark("收货", length!.Value, width!.Value, height!.Value, actualWeight!.Value));
await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.InboundTrackCode);
await EnqueueInboundBillingAsync(order); await EnqueueInboundBillingAsync(order);
} }
finally finally
@ -2646,6 +2612,270 @@ public partial class OrderService
return text[..512]; 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;
}
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true);
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 afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.PreDeclare,
beforeSnap,
afterSnap,
"预报计费/排货完成",
operatorUserId > 0 ? operatorUserId : null,
operatorName: null,
operatorType: "staff",
source: "job");
var customer = order.CustomerId.HasValue
? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value)
: null;
await PublishParcelCreateTrackAndLabelAsync(
order,
customer,
receiver: null,
appendInitialTrack: true,
operatorUserId);
await AutoTriggerRuntimeService.EnqueueFireAsync(
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) private async Task EnqueueInboundBillingAsync(Order order)
{ {
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>() var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>()
@ -2797,6 +3027,7 @@ public partial class OrderService
EnsureBillingProductActive(product); EnsureBillingProductActive(product);
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true);
var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id]))
.OrderBy(x => x.Id) .OrderBy(x => x.Id)
.ToList(); .ToList();
@ -2819,7 +3050,26 @@ public partial class OrderService
order.UpdateTime = now; order.UpdateTime = now;
await OrderRepository.UpdateAsync(order); await OrderRepository.UpdateAsync(order);
EnqueueLifecycleTrackAppend(order, SmallBagTrackLifecycleEvents.OutboundTrackCode); order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.Outbound,
beforeSnap,
afterSnap,
FormatMeasuredDimensionsRemark("发货", length!.Value, width!.Value, height!.Value, actualWeight!.Value));
await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.OutboundTrackCode);
}
private static string FormatMeasuredDimensionsRemark(
string stage,
decimal length,
decimal width,
decimal height,
decimal actualWeight)
{
return $"{stage} 长×宽×高×重:{length:0.####}×{width:0.####}×{height:0.####}×{actualWeight:0.####}";
} }
private async Task<OrderParcelCreateOutput> BuildParcelCreateOutputAsync(Order order) private async Task<OrderParcelCreateOutput> BuildParcelCreateOutputAsync(Order order)
@ -2877,46 +3127,56 @@ public partial class OrderService
TrackNo = order.TrackNo, TrackNo = order.TrackNo,
ConsigneeCountry = order.ConsigneeCountry, ConsigneeCountry = order.ConsigneeCountry,
ConsigneeCountryName = order.ConsigneeCountryName, ConsigneeCountryName = order.ConsigneeCountryName,
LabelUrl = labelUrl LabelUrl = labelUrl,
OrderStatus = order.OrderStatus,
PredeclareFailReason = order.PredeclareFailReason,
ForecastBillingQueued = string.Equals(
order.PredeclareFailReason,
ForecastBillingProcessingReason,
StringComparison.Ordinal)
}; };
} }
private ISmallBagTrackOrderNotifier SmallBagTrackOrderNotifier => private IOrderParcelCreateEventPublisher ParcelCreateEventPublisher =>
LazyServiceProvider.GetRequiredService<ISmallBagTrackOrderNotifier>(); 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>(); request.Authorization = authorization;
if (!publisher.IsEnabled) }
{
// 本地未开队列:同步生成,避免 Pending 悬挂
var mapped = (await MapOrderOutputsAsync([order])).First();
await LabelPdfGenerationService.GenerateAndPersistAsync(
order,
mapped,
operatorUserId: CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0);
return;
}
await BatchJobCoordinator.CreateAndEnqueueAsync( await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id);
new OrderBatchJobCreateInput await ParcelCreateEventPublisher.PublishLabelGenerateAsync(
{ order,
Action = OrderBatchActionCodes.GenerateLabelPdf, operatorUserId
OrderIds = [order.Id], ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0));
RequestId = $"label-pdf:{order.Id}:{Guid.NewGuid():N}"
},
CurrentUser.IsAuthenticated ? CurrentUser.GetId<int>() : 0);
}
catch (Exception)
{
// 投递失败不阻断创单;打印时会补生成
}
} }
/// <summary> /// <summary>
/// 小包预报成功后异步同步轨迹订单(POST /ProviderService/Order/UpsertSmallBag)。 /// 小包预报成功后异步同步轨迹订单(POST /ProviderService/Order/UpsertSmallBag)。
/// 入库等非创单路径仍可调用;创单路径请走 <see cref="PublishParcelCreateTrackAndLabelAsync"/>。
/// </summary> /// </summary>
private async Task EnqueueSmallBagTrackUpsertAsync( private async Task EnqueueSmallBagTrackUpsertAsync(
Order order, Order order,
@ -2938,10 +3198,10 @@ public partial class OrderService
request.Authorization = authorization; 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)) if (string.IsNullOrWhiteSpace(order.ServiceNo))
{ {
@ -2956,10 +3216,10 @@ public partial class OrderService
.Authorization .Authorization
.ToString(); .ToString();
SmallBagTrackOrderNotifier.EnqueueLifecycleTrackAppend( await SideEffectEventPublisher.PublishLifecycleTrackAppendAsync(
order.Id,
order.ServiceNo, order.ServiceNo,
trackCode, trackCode,
order.Id,
string.IsNullOrWhiteSpace(authorization) ? null : authorization); string.IsNullOrWhiteSpace(authorization) ? null : authorization);
} }

View File

@ -1,6 +1,9 @@
using System.Data;
using System.Globalization;
using Fake.Application; using Fake.Application;
using Fake.Domain.Exceptions; using Fake.Domain.Exceptions;
using Fake.ObjectStorage; using Fake.ObjectStorage;
using Fake.UnitOfWork;
using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Mvc; using Microsoft.AspNetCore.Mvc;
using MiniExcelLibs; using MiniExcelLibs;
@ -11,6 +14,7 @@ using Oms.Application.Labeling;
using Oms.Application.Shared; using Oms.Application.Shared;
using Oms.Application.Storage; using Oms.Application.Storage;
using Newtonsoft.Json; using Newtonsoft.Json;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Application.Services.Orders.Batch; using Oms.Application.Services.Orders.Batch;
using Oms.Domain.ChannelsAggregate.Repositories; using Oms.Domain.ChannelsAggregate.Repositories;
using Oms.Domain.CustomersAggregate.Repositories; using Oms.Domain.CustomersAggregate.Repositories;
@ -146,6 +150,7 @@ public partial class OrderService : ApplicationService
throw new BusinessException("订单未关联客户,无法支付"); throw new BusinessException("订单未关联客户,无法支付");
} }
var beforeSnap = await CaptureOrderAuditSnapshotAsync(order);
var fees = await OrderRepository.GetFeesByOrderIdsAsync([order.Id]); var fees = await OrderRepository.GetFeesByOrderIdsAsync([order.Id]);
var payAmount = order.PayableAmount ?? fees var payAmount = order.PayableAmount ?? fees
.Where(x => x.Type == OrderFeeTypeCodes.Receivable) .Where(x => x.Type == OrderFeeTypeCodes.Receivable)
@ -188,6 +193,16 @@ public partial class OrderService : ApplicationService
order.IsPaid = 1; order.IsPaid = 1;
order.OrderStatus = OrderStatusCodes.PendingShip; order.OrderStatus = OrderStatusCodes.PendingShip;
await UpdateOrderAsync(order); await UpdateOrderAsync(order);
order = await GetOrderOrThrowAsync(order.Id);
var afterSnap = await CaptureOrderAuditSnapshotAsync(order);
var amountText = payAmount.ToString("0.####", CultureInfo.InvariantCulture);
await OperationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.Pay,
beforeSnap,
afterSnap,
$"支付金额 {amountText}");
} }
/// <summary> /// <summary>
@ -252,6 +267,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参:订单操作参数。</param> /// <param name="input">入参:订单操作参数。</param>
/// <returns>无返回体,仅完成订单预报处理。</returns> /// <returns>无返回体,仅完成订单预报处理。</returns>
[HttpPost("/order/pre-declare")] [HttpPost("/order/pre-declare")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task PreDeclareAsync(OrderActionInput input) public async Task PreDeclareAsync(OrderActionInput input)
{ {
var order = await GetOrderOrThrowAsync(input.Id); var order = await GetOrderOrThrowAsync(input.Id);
@ -345,6 +361,10 @@ public partial class OrderService : ApplicationService
var existingUrl = await OrderFileManager.GetSystemLabelUrlAsync(order.Id); var existingUrl = await OrderFileManager.GetSystemLabelUrlAsync(order.Id);
if (!string.IsNullOrWhiteSpace(existingUrl)) if (!string.IsNullOrWhiteSpace(existingUrl))
{ {
await OperationLogWriter.TryWriteSimpleAsync(
order,
OrderOperationActionCodes.Print,
"打印面单(使用已有文件)");
return new OrderPrintOutput return new OrderPrintOutput
{ {
PrintUrl = existingUrl!, PrintUrl = existingUrl!,
@ -364,6 +384,10 @@ public partial class OrderService : ApplicationService
{ {
// 兼容回落:仍无模板时给 HTML stub,避免彻底不可打 // 兼容回落:仍无模板时给 HTML stub,避免彻底不可打
var fallbackUrl = await CreateOrderPrintFileAsync(mapped); var fallbackUrl = await CreateOrderPrintFileAsync(mapped);
await OperationLogWriter.TryWriteSimpleAsync(
order,
OrderOperationActionCodes.Print,
"打印面单(回落 HTML)");
return new OrderPrintOutput return new OrderPrintOutput
{ {
PrintUrl = fallbackUrl, PrintUrl = fallbackUrl,
@ -372,6 +396,10 @@ public partial class OrderService : ApplicationService
}; };
} }
await OperationLogWriter.TryWriteSimpleAsync(
order,
OrderOperationActionCodes.Print,
"打印面单");
return new OrderPrintOutput return new OrderPrintOutput
{ {
PrintUrl = result.LabelUrl!, PrintUrl = result.LabelUrl!,
@ -419,6 +447,21 @@ public partial class OrderService : ApplicationService
PrintUrl = url, PrintUrl = url,
LabelGenerateStatus = LabelGenerateStatusCodes.Ready LabelGenerateStatus = LabelGenerateStatusCodes.Ready
}); });
try
{
var existingOrder = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId);
if (existingOrder != null)
{
await OperationLogWriter.TryWriteSimpleAsync(
existingOrder,
OrderOperationActionCodes.Print,
"批量打印(使用已有文件)");
}
}
catch
{
// 日志失败不影响打印
}
} }
else else
{ {
@ -565,6 +608,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参:订单包裹创建参数。</param> /// <param name="input">入参:订单包裹创建参数。</param>
/// <returns>订单包裹创建结果。</returns> /// <returns>订单包裹创建结果。</returns>
[HttpPost("/order/parcel/create")] [HttpPost("/order/parcel/create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelCreateOutput> CreateParcelAsync(OrderParcelCreateInput input) public async Task<OrderParcelCreateOutput> CreateParcelAsync(OrderParcelCreateInput input)
{ {
var inbound = string.IsNullOrWhiteSpace(input.InboundMethod) var inbound = string.IsNullOrWhiteSpace(input.InboundMethod)
@ -596,6 +640,7 @@ public partial class OrderService : ApplicationService
/// <summary> /// <summary>
/// 客户 OpenAPI 创单:同一客户相同 clientNo 视为幂等,返回已存在订单,不新建。 /// 客户 OpenAPI 创单:同一客户相同 clientNo 视为幂等,返回已存在订单,不新建。
/// </summary> /// </summary>
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelCreateOutput> CreateParcelForCustomerOpenApiAsync(OrderParcelCreateInput input) public async Task<OrderParcelCreateOutput> CreateParcelForCustomerOpenApiAsync(OrderParcelCreateInput input)
{ {
NormalizeParcelCreateInput(input); NormalizeParcelCreateInput(input);
@ -660,21 +705,24 @@ public partial class OrderService : ApplicationService
return; return;
} }
var publisher = LazyServiceProvider.GetRequiredService<IOrderParcelCreateEventPublisher>();
try try
{ {
var pickupService = LazyServiceProvider.GetRequiredService<PickupOrderService>(); // 优先入 Fake.EventBus;发布失败则同步回落建揽收并回写单号。
var pickup = await pickupService.CreateForParcelInternalAsync(new PickupOrderCreateInput var queued = await publisher.PublishPickupAttachAsync(input, output.OrderId);
if (queued)
{ {
CustomerId = input.CustomerId, output.PickupQueued = true;
SenderAddressId = input.SenderAddressId.Value, return;
Direction = PickupDirectionCodes.NormalizeOrDefault(input.PickupDirection), }
WarehouseId = input.WarehouseId.Value,
CarrierCode = input.CarrierCode, var pickup = await publisher.ExecutePickupAttachAsync(input, output.OrderId);
CarrierService = input.CarrierService, if (pickup == null)
ExpectPickupTime = input.ExpectPickupTime.Value, {
OrderIds = [output.OrderId], output.PickupCreateError = "揽收单创建失败";
Remark = null return;
}); }
output.PickupNo = pickup.PickupNo; output.PickupNo = pickup.PickupNo;
output.PickupOrderId = pickup.Id; output.PickupOrderId = pickup.Id;
output.CarrierWaybillNo = pickup.CarrierWaybillNo; output.CarrierWaybillNo = pickup.CarrierWaybillNo;
@ -691,6 +739,7 @@ public partial class OrderService : ApplicationService
/// <param name="input">入参:批量创建小包请求集合,每条记录沿用单票创建结构。</param> /// <param name="input">入参:批量创建小包请求集合,每条记录沿用单票创建结构。</param>
/// <returns>批量处理结果,包含批次号、成功失败数量与逐条执行明细。</returns> /// <returns>批量处理结果,包含批次号、成功失败数量与逐条执行明细。</returns>
[HttpPost("/order/parcel/batch-create")] [HttpPost("/order/parcel/batch-create")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task<OrderParcelBatchCreateOutput> BatchCreateParcelAsync(OrderParcelBatchCreateInput input) public async Task<OrderParcelBatchCreateOutput> BatchCreateParcelAsync(OrderParcelBatchCreateInput input)
{ {
if (input == null || input.Count == 0) if (input == null || input.Count == 0)

View File

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

View File

@ -8,6 +8,14 @@ public static class OrderOperationActionCodes
public const string RecalcPayable = "recalcPayable"; public const string RecalcPayable = "recalcPayable";
public const string Reallocate = "reallocate"; public const string Reallocate = "reallocate";
public const string BindSnapshotVersions = "bindSnapshotVersions"; public const string BindSnapshotVersions = "bindSnapshotVersions";
public const string UpdateAdditionalServices = "updateAdditionalServices";
public const string AddFee = "addFee";
public const string Pay = "pay";
public const string PreDeclare = "preDeclare";
public const string DownstreamForward = "downstreamForward";
public const string Print = "print";
public const string Inbound = "inbound";
public const string Outbound = "outbound";
public static string GetActionName(string action) => action switch public static string GetActionName(string action) => action switch
{ {
@ -17,6 +25,14 @@ public static class OrderOperationActionCodes
RecalcPayable => "重算应付", RecalcPayable => "重算应付",
Reallocate => "重新排货", Reallocate => "重新排货",
BindSnapshotVersions => "切换价格/规则版本", BindSnapshotVersions => "切换价格/规则版本",
UpdateAdditionalServices => "更新附加服务",
AddFee => "添加费用",
Pay => "支付",
PreDeclare => "重新预报",
DownstreamForward => "预报下游",
Print => "打印面单",
Inbound => "入库",
Outbound => "出库",
_ => action _ => action
}; };
} }

View File

@ -0,0 +1,23 @@
using System.Collections.Concurrent;
using Fake.EventBus;
using Fake.EventBus.Distributed;
namespace Oms.Tests.Messaging.Fakes;
public sealed class RecordingPhysicalDistributedEventBus : IPhysicalDistributedEventBus
{
private readonly ConcurrentQueue<IntegrationEvent> _published = new();
public IReadOnlyList<IntegrationEvent> Published => _published.ToArray();
public void Clear() => _published.Clear();
public Task PublishAsync(IntegrationEvent @event, CancellationToken cancellationToken = default)
{
_published.Enqueue(@event);
return Task.CompletedTask;
}
public Task PublishAsync(Event @event, CancellationToken cancellationToken = default)
=> PublishAsync((IntegrationEvent)@event, cancellationToken);
}

View File

@ -0,0 +1,51 @@
using Fake.Autofac;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Modularity;
using Fake.SyncEx;
using Microsoft.Data.Sqlite;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
[DependsOn(typeof(FakeAutofacModule))]
[DependsOn(typeof(FakeEntityFrameworkCoreModule))]
[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))]
public class OmsOutboxTestModule : FakeModule
{
public override void ConfigureServices(ServiceConfigurationContext context)
{
context.Services.AddSingleton<RecordingPhysicalDistributedEventBus>();
context.Services.AddSingleton<IPhysicalDistributedEventBus>(sp =>
sp.GetRequiredService<RecordingPhysicalDistributedEventBus>());
context.Services.Configure<OutboxPublisherOptions>(o =>
{
o.StartupDelay = TimeSpan.FromHours(1);
o.ScanInterval = TimeSpan.FromHours(1);
});
context.Services.Configure<InboxCleanupOptions>(o =>
{
o.StartupDelay = TimeSpan.FromHours(1);
o.CleanupInterval = TimeSpan.FromHours(1);
});
var connection = new SqliteConnection("Filename=:memory:");
connection.Open();
context.Services.AddDbContext<IntegrationEventLogContext>(builder =>
{
builder.UseSqlite(connection);
}, ServiceLifetime.Scoped, ServiceLifetime.Singleton);
}
public override void PreConfigureApplication(ApplicationConfigureContext context)
{
using var scope = context.ServiceProvider.CreateScope();
var ctx = scope.ServiceProvider.GetRequiredService<IntegrationEventLogContext>();
SyncContext.Run(async () => await ctx.Database.EnsureCreatedAsync());
}
}

View File

@ -0,0 +1,90 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Oms.Application.Messaging;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderBatchJobPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderBatchJobOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_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]
public async Task ForecastBilling任务消息_事务Commit后投递()
{
_physicalBus.Clear();
var message = new OrderBatchJobMessage
{
JobId = 1,
JobItemId = 2,
Action = OrderBatchActionCodes.ForecastBilling,
OrderId = 55,
MessageId = "job:1:55:forecast-billing",
OperatorUserId = 9,
PayloadJson = "{\"CouponInstanceId\":1}"
};
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAsync(message);
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
var evt = _physicalBus.Published.Should().ContainSingle()
.Which.Should().BeOfType<OrderBatchJobIntegrationEvent>().Subject;
evt.Action.Should().Be(OrderBatchActionCodes.ForecastBilling);
evt.OrderId.Should().Be(55);
evt.MessageId.Should().Be(message.MessageId);
}
[Fact]
public async Task 队列未启用时_Publish应抛错()
{
var disabled = new OrderBatchJobPublisher(
ServiceProvider.GetRequiredService<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*");
}
}

View File

@ -0,0 +1,131 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Oms.Application.Integrations.SmallBagTrack;
using Oms.Application.Messaging;
using Oms.Application.Messaging.ParcelCreate;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly IDistributedEventBus _distributedEventBus;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IOutboxEventLogService _outbox;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderParcelCreateEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderParcelCreateOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_distributedEventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_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]
public void IDistributedEventBus_应为Outbox装饰器()
{
_distributedEventBus.Should().BeOfType<OutboxDistributedEventBus>();
}
[Fact]
public async Task 事务Uow内发布9610_应在Complete后投递物理总线()
{
_physicalBus.Clear();
var order = new Order { Id = 101, ServiceNo = "SN101" };
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order, "lazada");
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().ContainSingle();
var evt = _physicalBus.Published[0].Should().BeOfType<OrderParcelCustoms9610UpsertIntegrationEvent>().Subject;
evt.OrderId.Should().Be(101);
evt.PlatformCode.Should().Be("lazada");
var pending = (await _outbox.RetrieveEventLogsPendingToPublishAsync(50)).ToList();
pending.Should().NotContain(e => e.EventId == evt.Id);
}
[Fact]
public async Task 事务Uow内发布轨迹与面单_Commit后应各投递一次()
{
_physicalBus.Clear();
var order = new Order { Id = 202, ServiceNo = "SN202" };
var request = new SmallBagTrackUpsertRequest
{
OrderNo = "SN202",
AppendInitialTrackOnSuccess = true
};
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishTrackUpsertAsync(order, request, customerId: 9);
await _publisher.PublishLabelGenerateAsync(order, operatorUserId: 7);
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().HaveCount(2);
_physicalBus.Published.OfType<OrderParcelTrackUpsertIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 202 && e.CustomerId == 9 && e.AppendInitialTrackOnSuccess);
_physicalBus.Published.OfType<OrderParcelLabelGenerateIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 202 && e.OperatorUserId == 7);
}
[Fact]
public async Task 回滚后不应投递()
{
_physicalBus.Clear();
var order = new Order { Id = 303, ServiceNo = "SN303" };
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishCustoms9610UpsertAsync(order);
await uow.RollbackAsync();
}
_physicalBus.Published.Should().BeEmpty();
}
[Fact]
public async Task 无环境事务时_短事务仍能写入并投递()
{
_physicalBus.Clear();
_uowManager.Current.Should().BeNull();
var order = new Order { Id = 404, ServiceNo = "SN404" };
await _publisher.PublishCustoms9610UpsertAsync(order, "shopee");
_physicalBus.Published.Should().ContainSingle()
.Which.Should().BeOfType<OrderParcelCustoms9610UpsertIntegrationEvent>()
.Which.OrderId.Should().Be(404);
}
}

View File

@ -0,0 +1,62 @@
using Fake;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.Testing;
using Fake.UnitOfWork;
using FluentAssertions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Oms.Application.Messaging;
using Oms.Application.Messaging.OrderSideEffects;
using Oms.Application.Services.Orders.Issues;
using Oms.Tests.Messaging.Fakes;
namespace Oms.Tests.Messaging;
public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools<OmsOutboxTestModule>
{
private readonly IUnitOfWorkManager _uowManager;
private readonly IDistributedEventBus _distributedEventBus;
private readonly RecordingPhysicalDistributedEventBus _physicalBus;
private readonly IEfDbContextProvider<IntegrationEventLogContext> _dbProvider;
private readonly OrderSideEffectEventPublisher _publisher;
protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options)
{
options.UseAutofac();
}
public OrderSideEffectOutboxPublisherTests()
{
_uowManager = ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
_distributedEventBus = ServiceProvider.GetRequiredService<IDistributedEventBus>();
_physicalBus = ServiceProvider.GetRequiredService<RecordingPhysicalDistributedEventBus>();
_dbProvider = ServiceProvider.GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
_publisher = new OrderSideEffectEventPublisher(
_distributedEventBus,
_uowManager,
NullLogger<OrderSideEffectEventPublisher>.Instance);
}
[Fact]
public async Task 事务内发布自动触发与生命周期轨迹_Commit后投递()
{
_physicalBus.Clear();
using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional()))
{
await _dbProvider.GetDbContextAsync();
await _publisher.PublishAutoTriggerFireAsync(11, AutoTriggerTiming.OrderCreate, 3);
await _publisher.PublishLifecycleTrackAppendAsync(11, "SN11", "A0301");
_physicalBus.Published.Should().BeEmpty();
await uow.CompleteAsync();
}
_physicalBus.Published.Should().HaveCount(2);
_physicalBus.Published.OfType<OrderAutoTriggerFireIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 11 && e.Timing == AutoTriggerTiming.OrderCreate.ToString() && e.OperatorUserId == 3);
_physicalBus.Published.OfType<OrderLifecycleTrackAppendIntegrationEvent>().Should().ContainSingle(e =>
e.OrderId == 11 && e.OrderNo == "SN11");
}
}

View File

@ -13,9 +13,14 @@
<PrivateAssets>all</PrivateAssets> <PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets> <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference> </PackageReference>
<PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Testing" Version="$(FakePackageVersion)" />
<PackageReference Include="FluentAssertions" Version="8.7.1" /> <PackageReference Include="FluentAssertions" Version="8.7.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="8.0.11" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" /> <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
<PackageReference Include="Moq" Version="4.20.72" /> <PackageReference Include="Moq" Version="4.20.72" />
<PackageReference Include="Shouldly" Version="4.3.0" />
<PackageReference Include="xunit" Version="2.9.2" /> <PackageReference Include="xunit" Version="2.9.2" />
<PackageReference Include="xunit.runner.visualstudio" Version="2.8.2"> <PackageReference Include="xunit.runner.visualstudio" Version="2.8.2">
<PrivateAssets>all</PrivateAssets> <PrivateAssets>all</PrivateAssets>

View File

@ -15,12 +15,13 @@
<ItemGroup> <ItemGroup>
<PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.Autofac" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Consul" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.Consul" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.EventBus.IntegrationEventLog" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.FileManagement.Application" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.FileManagement.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.FileManagement.Infrastructure" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.FileManagement.Infrastructure" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.ObjectStorage.TencentCos" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.ObjectStorage.TencentCos" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Serilog" Version="$(FakePackageVersion)" /> <PackageReference Include="Fake.Serilog" Version="$(FakePackageVersion)" />
<PackageReference Include="Serilog.AspNetCore" Version="8.0.3"/> <PackageReference Include="Serilog.AspNetCore" Version="8.0.3" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>

View File

@ -7,6 +7,7 @@ using Fake.AspNetCore.Auditing;
using Fake.AspNetCore.ExceptionHandling; using Fake.AspNetCore.ExceptionHandling;
using Fake.AspNetCore.Newtonsoft; using Fake.AspNetCore.Newtonsoft;
using Fake.AspNetCore.Swagger; using Fake.AspNetCore.Swagger;
using Fake.EventBus.IntegrationEventLog;
using Fake.FeiShu; using Fake.FeiShu;
using Fake.FileManagement.Application; using Fake.FileManagement.Application;
using Fake.FileManagement.Infrastructure; using Fake.FileManagement.Infrastructure;
@ -37,6 +38,8 @@ namespace SimpleAdmin.Api;
typeof(FakeFileManagementApplicationModule), typeof(FakeFileManagementApplicationModule),
typeof(FakeFileManagementInfrastructureModule), typeof(FakeFileManagementInfrastructureModule),
typeof(FakeObjectStorageTencentCosModule), typeof(FakeObjectStorageTencentCosModule),
// 在 RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox
typeof(FakeEventBusIntegrationEventLogModule),
typeof(FakeAspNetCoreModule), typeof(FakeAspNetCoreModule),
typeof(FakeSerilogModule) typeof(FakeSerilogModule)
)] )]
@ -80,6 +83,20 @@ public class SimpleAdminApiModule : FakeModule
services.AddOmsWeChat(configuration); services.AddOmsWeChat(configuration);
services.AddSingleton<IRbacAuthCache, RbacAuthCache>(); services.AddSingleton<IRbacAuthCache, RbacAuthCache>();
// Outbox / Inbox 与业务同库(EF),通过 DbTransaction 征用 SqlSugar 事务
services.AddDbContext<IntegrationEventLogContext>(options =>
{
var connectionString = configuration.GetConnectionString("IntegrationEventLog")
?? configuration.GetConnectionString("Default")
?? configuration["ConnectionStrings:Default"]
?? string.Empty;
var serverVersion = ServerVersion.AutoDetect(connectionString);
options.UseMySql(connectionString, serverVersion);
#if DEBUG
options.EnableSensitiveDataLogging();
#endif
});
//Json格式化全局处理 //Json格式化全局处理
context.Services.AddFakeNewtonsoft(); context.Services.AddFakeNewtonsoft();
@ -188,11 +205,14 @@ public class SimpleAdminApiModule : FakeModule
// 端点映射 // 端点映射
app.MapControllers(); app.MapControllers();
// FileManagement 与 RBAC 同库并行(EF);应用 Pending migrations 建 stored_file // FileManagement / Outbox 与业务同库并行(EF)
using (var scope = app.Services.CreateScope()) using (var scope = app.Services.CreateScope())
{ {
var fileDb = scope.ServiceProvider.GetRequiredService<FileManagementDbContext>(); var fileDb = scope.ServiceProvider.GetRequiredService<FileManagementDbContext>();
fileDb.Database.Migrate(); fileDb.Database.Migrate();
var outboxDb = scope.ServiceProvider.GetRequiredService<IntegrationEventLogContext>();
IntegrationEventLogSchema.EnsureTablesAsync(outboxDb).GetAwaiter().GetResult();
} }
app.Services.GetRequiredService<IFeiShuNotificationService>().Send("Application started"); app.Services.GetRequiredService<IFeiShuNotificationService>().Send("Application started");

View File

@ -0,0 +1,48 @@
using Fake.EventBus.IntegrationEventLog;
using Microsoft.EntityFrameworkCore;
namespace SimpleAdmin.Api.Storage;
/// <summary>
/// Outbox/Inbox 与业务同库时 EnsureCreated 无效(库已存在),用 IF NOT EXISTS 建表。
/// </summary>
internal static class IntegrationEventLogSchema
{
public static async Task EnsureTablesAsync(
IntegrationEventLogContext db,
CancellationToken cancellationToken = default)
{
await db.Database.ExecuteSqlRawAsync(
"""
CREATE TABLE IF NOT EXISTS `OutboxEventLog` (
`EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL,
`EventTypeName` varchar(512) NOT NULL,
`State` int NOT NULL,
`TimesSent` int NOT NULL,
`CreationTime` datetime(6) NOT NULL,
`Content` longtext NOT NULL,
`TransactionId` varchar(64) NOT NULL,
`LockExpiresAt` datetime(6) NULL,
PRIMARY KEY (`EventId`),
KEY `IX_OutboxEventLog_State_CreationTime` (`State`, `CreationTime`),
KEY `IX_OutboxEventLog_TransactionId` (`TransactionId`)
) CHARACTER SET utf8mb4;
""",
cancellationToken);
await db.Database.ExecuteSqlRawAsync(
"""
CREATE TABLE IF NOT EXISTS `InboxEventLog` (
`EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL,
`EventTypeName` varchar(512) NOT NULL,
`State` int NOT NULL,
`ProcessedTime` datetime(6) NOT NULL,
`Content` longtext NOT NULL,
`ErrorMessage` longtext NULL,
PRIMARY KEY (`EventId`),
KEY `IX_InboxEventLog_ProcessedTime` (`ProcessedTime`)
) CHARACTER SET utf8mb4;
""",
cancellationToken);
}
}

View File

@ -146,5 +146,13 @@
"RabbitMq": { "RabbitMq": {
"Enabled": false "Enabled": false
} }
},
"IntegrationEventLog": {
"OutboxPublisher": {
"StartupDelay": "00:00:05",
"ScanInterval": "00:00:10",
"BatchSize": 100,
"MaxDegreeOfParallelism": 8
}
} }
} }