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

订阅轨迹中心轨迹状态，变更运单状态 #15

Merged
zzh merged 4 commits from test into master 2026-10-27 17:50:21 +08:00
Conversation 0 Commits 4 Files Changed 17 +569 -1
11 changed files with 569 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit ff9a3ae8be - Show all commits

5
src/Oms.Application/Integrations/SmallBagTrack/SmallBagTrackServiceExtensions.cs
View File

@ -1,5 +1,6 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Oms.Application.Messaging.TrackChange;
namespace Oms.Application.Integrations.SmallBagTrack;
@ -8,12 +9,16 @@ public static class SmallBagTrackServiceExtensions
public static IServiceCollection AddOmsSmallBagTrack(this IServiceCollection services, IConfiguration configuration)
{
services.Configure<SmallBagTrackOptions>(configuration.GetSection(SmallBagTrackOptions.SectionName));
services.Configure<TrackChangeSubscribeOptions>(
configuration.GetSection(TrackChangeSubscribeOptions.SectionName));
services.AddHttpClient(SmallBagTrackOrderClient.HttpClientName);
services.AddScoped<ISmallBagTrackConfigProvider, SmallBagTrackConfigProvider>();
services.AddScoped<ISmallBagTrackUpsertRequestBuilder, SmallBagTrackUpsertRequestBuilder>();
services.AddScoped<ISmallBagTrackOrderClient, SmallBagTrackOrderClient>();
services.AddScoped<IPickupBigBagTrackSyncer, PickupBigBagTrackSyncer>();
services.AddSingleton<ISmallBagTrackOrderNotifier, SmallBagTrackOrderNotifier>();
services.AddScoped<ITrackChangeOrderStatusApplier, TrackChangeOrderStatusApplier>();
services.AddHostedService<TrackChangeSubscriberHostedService>();
return services;
}
}

12
src/Oms.Application/Messaging/TrackChange/ITrackChangeOrderStatusApplier.cs Normal file
View File

@ -0,0 +1,12 @@
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 根据轨迹变更事件同步 OMS 出库后订单状态。
/// </summary>
public interface ITrackChangeOrderStatusApplier
{
/// <summary>
/// 处理一条 TrackChange 消息；业务跳过不抛异常。
/// </summary>
Task ApplyAsync(TrackChangeEto eto, CancellationToken cancellationToken = default);
}

65
src/Oms.Application/Messaging/TrackChange/TrackChangeEto.cs Normal file
View File

@ -0,0 +1,65 @@
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 轨迹中心轨迹变更广播（RabbitMQ topic：track.on_track_change_ex / TrackChange.{AppKey}）。
/// 字段与 TrackCenter.Application.Events.TrackChangeEto 对齐（Newtonsoft 默认 PascalCase）。
/// </summary>
public sealed class TrackChangeEto
{
/// <summary>轨迹唯一 id；操作失败时可能为 null。</summary>
public string? TrackEsId { get; set; }
/// <summary>操作类型：1 新增，2 删除，3 修改。</summary>
public int Option { get; set; }
/// <summary>操作是否成功。</summary>
public bool IsSuccess { get; set; }
/// <summary>失败原因。</summary>
public string? FailedReason { get; set; }
/// <summary>应用 AppKey。</summary>
public string? AppKey { get; set; }
/// <summary>单号（OMS 侧对应运单号 ServiceNo）。</summary>
public string? OrderNo { get; set; }
/// <summary>客户标识。</summary>
public string? Customer { get; set; }
/// <summary>操作人。</summary>
public string? OperationMan { get; set; }
/// <summary>轨迹内容代码。</summary>
public string? TrackCode { get; set; }
/// <summary>轨迹内容。</summary>
public string? Content { get; set; }
/// <summary>轨迹内容英文。</summary>
public string? EnStatus { get; set; }
/// <summary>轨迹状态代码（如 D03 / D09 / D11）。</summary>
public string? TrackStatusCode { get; set; }
/// <summary>轨迹发生地。</summary>
public string? TrackPlace { get; set; }
/// <summary>轨迹发生时间。</summary>
public DateTime TrackTime { get; set; }
/// <summary>轨迹发生地时区。</summary>
public float TrackTimeZone { get; set; }
/// <summary>是否对外。</summary>
public bool IsExternal { get; set; }
/// <summary>数据来源：1 外部服务商，2 内部系统，3 人工。</summary>
public int Source { get; set; }
/// <summary>轨迹备注。</summary>
public string? Remarks { get; set; }
/// <summary>扩展信息。</summary>
public string? Extra { get; set; }
}

168
src/Oms.Application/Messaging/TrackChange/TrackChangeOrderStatusApplier.cs Normal file
View File

@ -0,0 +1,168 @@
using Fake.DependencyInjection;
using Fake.UnitOfWork;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders;
using Oms.Application.Shared;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 将轨迹状态 D03/D09/D11 同步到运单订单状态（运输中 / 派送失败 / 派送成功）。
/// 规则对齐手动「出库后状态校正」：仅出库后四态可改；派送成功只进不出。
/// </summary>
public sealed class TrackChangeOrderStatusApplier(
IOrderRepository orderRepository,
OrderOperationLogWriter operationLogWriter,
ReferralCommissionManager referralCommissionManager,
MarketingCouponManager marketingCouponManager,
IUnitOfWorkManager unitOfWorkManager,
ILogger<TrackChangeOrderStatusApplier> logger) : ITrackChangeOrderStatusApplier, IScopedDependency
{
private static readonly HashSet<string> PostShipStatuses = new(StringComparer.OrdinalIgnoreCase)
{
OrderStatusCodes.PendingOnline,
OrderStatusCodes.Shipping,
OrderStatusCodes.DeliverySuccess,
OrderStatusCodes.DeliveryFailed
};
public async Task ApplyAsync(TrackChangeEto eto, CancellationToken cancellationToken = default)
{
if (eto == null)
{
return;
}
if (!eto.IsSuccess)
{
logger.LogDebug(
"TrackChange skip failed event OrderNo={OrderNo} Reason={Reason}",
eto.OrderNo,
eto.FailedReason);
return;
}
// 删除轨迹不回退 OMS 状态
if (eto.Option == 2)
{
logger.LogDebug("TrackChange skip delete Option=2 OrderNo={OrderNo}", eto.OrderNo);
return;
}
var targetStatus = TrackStatusOrderStatusMapper.TryMapToOrderStatus(eto.TrackStatusCode);
if (targetStatus == null)
{
logger.LogDebug(
"TrackChange skip unmapped status OrderNo={OrderNo} TrackStatusCode={TrackStatusCode}",
eto.OrderNo,
eto.TrackStatusCode);
return;
}
var orderNo = eto.OrderNo?.Trim();
if (string.IsNullOrWhiteSpace(orderNo))
{
logger.LogWarning("TrackChange skip empty OrderNo TrackStatusCode={TrackStatusCode}", eto.TrackStatusCode);
return;
}
using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional());
var order = await orderRepository.GetByServiceNoAsync(orderNo, cancellationToken);
if (order == null)
{
logger.LogWarning(
"TrackChange order not found ServiceNo={OrderNo} TrackStatusCode={TrackStatusCode}",
orderNo,
eto.TrackStatusCode);
await unitOfWork.CompleteAsync(cancellationToken);
return;
}
var fromStatus = order.OrderStatus?.Trim() ?? string.Empty;
if (!IsPostShipStatus(fromStatus))
{
logger.LogInformation(
"TrackChange skip non post-ship status OrderId={OrderId} ServiceNo={ServiceNo} From={From} TrackStatusCode={TrackStatusCode}",
order.Id,
order.ServiceNo,
fromStatus,
eto.TrackStatusCode);
await unitOfWork.CompleteAsync(cancellationToken);
return;
}
if (string.Equals(fromStatus, OrderStatusCodes.DeliverySuccess, StringComparison.OrdinalIgnoreCase))
{
logger.LogInformation(
"TrackChange skip deliverySuccess terminal OrderId={OrderId} ServiceNo={ServiceNo} TrackStatusCode={TrackStatusCode}",
order.Id,
order.ServiceNo,
eto.TrackStatusCode);
await unitOfWork.CompleteAsync(cancellationToken);
return;
}
if (string.Equals(fromStatus, targetStatus, StringComparison.OrdinalIgnoreCase))
{
logger.LogDebug(
"TrackChange idempotent skip OrderId={OrderId} ServiceNo={ServiceNo} Status={Status}",
order.Id,
order.ServiceNo,
targetStatus);
await unitOfWork.CompleteAsync(cancellationToken);
return;
}
var beforeSnap = OrderOperationLogSnapshot.CaptureOrder(order);
order.OrderStatus = targetStatus;
order.UpdateTime = DateTime.Now;
if (string.Equals(targetStatus, OrderStatusCodes.DeliverySuccess, StringComparison.OrdinalIgnoreCase))
{
order.SignTime = DateTime.Now;
}
await orderRepository.UpdateAsync(order, cancellationToken: cancellationToken);
if (string.Equals(targetStatus, OrderStatusCodes.DeliverySuccess, StringComparison.OrdinalIgnoreCase))
{
await referralCommissionManager.OnOrderDeliverySuccessAsync(order.Id, cancellationToken);
// 重新加载，避免营销侧依赖最新字段
order = await orderRepository.GetByServiceNoAsync(orderNo, cancellationToken) ?? order;
await marketingCouponManager.OnFirstDeliverySuccessAsync(order, cancellationToken);
}
order = await orderRepository.GetByServiceNoAsync(orderNo, cancellationToken) ?? order;
var afterSnap = OrderOperationLogSnapshot.CaptureOrder(order);
var fromName = OmsDocHelper.GetOrderStatusName(fromStatus) ?? fromStatus;
var toName = OmsDocHelper.GetOrderStatusName(targetStatus) ?? targetStatus;
var trackHint = string.IsNullOrWhiteSpace(eto.Content)
? eto.TrackStatusCode
: $"{eto.TrackStatusCode}/{eto.Content}";
await operationLogWriter.TryWriteAsync(
order,
OrderOperationActionCodes.TrackStatusSync,
beforeSnap,
afterSnap,
$"{fromName} → {toName}；轨迹同步 {trackHint}",
operatorId: 0,
operatorName: string.IsNullOrWhiteSpace(eto.OperationMan) ? "轨迹中心" : eto.OperationMan.Trim(),
operatorType: "system",
source: "trackCenter");
await unitOfWork.CompleteAsync(cancellationToken);
logger.LogInformation(
"TrackChange applied OrderId={OrderId} ServiceNo={ServiceNo} {From} → {To} TrackStatusCode={TrackStatusCode}",
order.Id,
order.ServiceNo,
fromStatus,
targetStatus,
eto.TrackStatusCode);
}
private static bool IsPostShipStatus(string? status) =>
!string.IsNullOrWhiteSpace(status) && PostShipStatuses.Contains(status.Trim());
}

43
src/Oms.Application/Messaging/TrackChange/TrackChangeSubscribeOptions.cs Normal file
View File

@ -0,0 +1,43 @@
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 订阅轨迹中心 TrackChange 广播的配置（Consul / 配置：Oms:TrackChangeSubscribe）。
/// </summary>
public class TrackChangeSubscribeOptions
{
public const string SectionName = "Oms:TrackChangeSubscribe";
/// <summary>是否启用订阅；默认 true。</summary>
public bool Enabled { get; set; } = true;
/// <summary>
/// 轨迹中心广播交换器；须与 TrackCenter GlobalConst.TrackChangeExchange 一致。
/// </summary>
public string Exchange { get; set; } = "track.on_track_change_ex";
/// <summary>
/// 绑定路由键模板，{AppKey} 会替换为轨迹服务 AppKey。
/// </summary>
public string RoutingKeyTemplate { get; set; } = "TrackChange.{AppKey}";
/// <summary>
/// 本地队列名模板；{AppKey} 会替换。为空时使用 oms.track_change.{AppKey}。
/// </summary>
public string? QueueNameTemplate { get; set; } = "oms.track_change.{AppKey}";
/// <summary>
/// Fake.RabbitMQ 连接名；为空则使用 RabbitMqEventBusOptions.ConnectionName。
/// </summary>
public string? ConnectionName { get; set; }
/// <summary>消费预取数量。</summary>
public ushort PrefetchCount { get; set; } = 10;
/// <summary>
/// 覆盖轨迹 AppKey（一般留空，走 Oms:TrackService / 系统其他配置 track_service）。
/// </summary>
public string? AppKey { get; set; }
/// <summary>处理失败时是否重新入队。</summary>
public bool FailedRequeue { get; set; } = true;
}

202
src/Oms.Application/Messaging/TrackChange/TrackChangeSubscriberHostedService.cs Normal file
View File

@ -0,0 +1,202 @@
using System.Text;
using Fake.EventBus.RabbitMQ;
using Fake.RabbitMQ;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using Oms.Application.Integrations.SmallBagTrack;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 订阅轨迹中心 TrackChange 广播，驱动 OMS 出库后订单状态变更。
/// </summary>
public sealed class TrackChangeSubscriberHostedService : BackgroundService
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly IRabbitMqConnectionPool _connectionPool;
private readonly IOptions<TrackChangeSubscribeOptions> _subscribeOptions;
private readonly IOptions<RabbitMqEventBusOptions> _eventBusOptions;
private readonly ILogger<TrackChangeSubscriberHostedService> _logger;
private IModel? _channel;
public TrackChangeSubscriberHostedService(
IServiceScopeFactory scopeFactory,
IRabbitMqConnectionPool connectionPool,
IOptions<TrackChangeSubscribeOptions> subscribeOptions,
IOptions<RabbitMqEventBusOptions> eventBusOptions,
ILogger<TrackChangeSubscriberHostedService> logger)
{
_scopeFactory = scopeFactory;
_connectionPool = connectionPool;
_subscribeOptions = subscribeOptions;
_eventBusOptions = eventBusOptions;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var options = _subscribeOptions.Value;
if (!options.Enabled)
{
_logger.LogInformation("TrackChange subscribe disabled");
return;
}
string? appKey;
try
{
appKey = await ResolveAppKeyAsync(stoppingToken);
}
catch (Exception ex)
{
_logger.LogError(ex, "TrackChange subscribe failed to resolve AppKey; consumer not started");
return;
}
if (string.IsNullOrWhiteSpace(appKey))
{
_logger.LogWarning("TrackChange subscribe skipped: AppKey empty (configure Oms:TrackService or Oms:TrackChangeSubscribe:AppKey)");
return;
}
var exchange = string.IsNullOrWhiteSpace(options.Exchange)
? "track.on_track_change_ex"
: options.Exchange.Trim();
var routingKey = (options.RoutingKeyTemplate ?? "TrackChange.{AppKey}")
.Replace("{AppKey}", appKey, StringComparison.OrdinalIgnoreCase);
var queueName = string.IsNullOrWhiteSpace(options.QueueNameTemplate)
? $"oms.track_change.{appKey}"
: options.QueueNameTemplate.Replace("{AppKey}", appKey, StringComparison.OrdinalIgnoreCase);
var connectionName = string.IsNullOrWhiteSpace(options.ConnectionName)
? _eventBusOptions.Value.ConnectionName
: options.ConnectionName;
try
{
// Connection 由 Fake 连接池托管，禁止 Dispose
var connection = _connectionPool.Get(connectionName);
_channel = connection.CreateModel();
_channel.ExchangeDeclare(exchange, ExchangeType.Topic, durable: true, autoDelete: false);
_channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false);
_channel.QueueBind(queueName, exchange, routingKey);
if (options.PrefetchCount > 0)
{
_channel.BasicQos(0, options.PrefetchCount, false);
}
var consumer = new AsyncEventingBasicConsumer(_channel);
consumer.Received += OnReceivedAsync;
_channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
_logger.LogInformation(
"TrackChange subscribe started Exchange={Exchange} RoutingKey={RoutingKey} Queue={Queue}",
exchange,
routingKey,
queueName);
}
catch (Exception ex)
{
_logger.LogError(ex, "TrackChange subscribe failed to start");
DisposeChannel();
return;
}
try
{
await Task.Delay(Timeout.Infinite, stoppingToken);
}
catch (OperationCanceledException)
{
// shutdown
}
finally
{
DisposeChannel();
_logger.LogInformation("TrackChange subscribe stopped");
}
}
private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs ea)
{
var channel = _channel;
if (channel == null || !channel.IsOpen)
{
return;
}
var options = _subscribeOptions.Value;
try
{
var json = Encoding.UTF8.GetString(ea.Body.Span);
var eto = JsonConvert.DeserializeObject<TrackChangeEto>(json);
if (eto == null)
{
_logger.LogWarning("TrackChange deserialize null body");
channel.BasicAck(ea.DeliveryTag, false);
return;
}
await using var scope = _scopeFactory.CreateAsyncScope();
var applier = scope.ServiceProvider.GetRequiredService<ITrackChangeOrderStatusApplier>();
await applier.ApplyAsync(eto);
channel.BasicAck(ea.DeliveryTag, false);
}
catch (Exception ex)
{
_logger.LogError(ex, "TrackChange handle failed DeliveryTag={DeliveryTag}", ea.DeliveryTag);
try
{
channel.BasicNack(ea.DeliveryTag, false, options.FailedRequeue);
}
catch (Exception nackEx)
{
_logger.LogError(nackEx, "TrackChange Nack failed DeliveryTag={DeliveryTag}", ea.DeliveryTag);
}
}
}
private async Task<string?> ResolveAppKeyAsync(CancellationToken cancellationToken)
{
var overrideKey = _subscribeOptions.Value.AppKey?.Trim();
if (!string.IsNullOrWhiteSpace(overrideKey))
{
return overrideKey;
}
await using var scope = _scopeFactory.CreateAsyncScope();
var trackConfig = scope.ServiceProvider.GetRequiredService<ISmallBagTrackConfigProvider>();
var settings = await trackConfig.GetAsync(cancellationToken);
return string.IsNullOrWhiteSpace(settings.AppKey) ? null : settings.AppKey.Trim();
}
private void DisposeChannel()
{
try
{
_channel?.Close();
_channel?.Dispose();
}
catch (Exception ex)
{
_logger.LogDebug(ex, "TrackChange channel dispose ignored");
}
finally
{
_channel = null;
}
}
public override void Dispose()
{
DisposeChannel();
base.Dispose();
}
}

38
src/Oms.Application/Messaging/TrackChange/TrackStatusOrderStatusMapper.cs Normal file
View File

@ -0,0 +1,38 @@
using Oms.Domain.OrdersAggregate.Consts;
namespace Oms.Application.Messaging.TrackChange;
/// <summary>
/// 轨迹状态码 → OMS 出库后订单状态。
/// </summary>
public static class TrackStatusOrderStatusMapper
{
/// <summary>运输中。</summary>
public const string InTransit = "D03";
/// <summary>派送失败。</summary>
public const string DeliveryFailed = "D09";
/// <summary>派送成功。</summary>
public const string DeliverySuccess = "D11";
/// <summary>
/// 将轨迹状态码映射为 OMS OrderStatus；未映射返回 null。
/// </summary>
public static string? TryMapToOrderStatus(string? trackStatusCode)
{
var code = trackStatusCode?.Trim();
if (string.IsNullOrWhiteSpace(code))
{
return null;
}
return code.ToUpperInvariant() switch
{
InTransit => OrderStatusCodes.Shipping,
DeliveryFailed => OrderStatusCodes.DeliveryFailed,
DeliverySuccess => OrderStatusCodes.DeliverySuccess,
_ => null
};
}
}

4
src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs
View File

@ -19,6 +19,9 @@ public static class OrderOperationActionCodes
public const string Outbound = "outbound";
public const string ChangePostShipStatus = "changePostShipStatus";
/// <summary>轨迹中心状态同步（TrackChange D03/D09/D11）。</summary>
public const string TrackStatusSync = "trackStatusSync";
public static string GetActionName(string action) => action switch
{
EditBasic => "编辑订单详情",
@ -37,6 +40,7 @@ public static class OrderOperationActionCodes
Inbound => "入库",
Outbound => "出库",
ChangePostShipStatus => "修改出库后状态",
TrackStatusSync => "轨迹状态同步",
_ => action
};
}

1
src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs
View File

@ -52,6 +52,7 @@ internal sealed class OrderE2eTestHost : IAsyncDisposable
["Redis:Host"] = "127.0.0.1:6379",
["Redis:Database"] = "0",
["OrderBatch:RabbitMq:Enabled"] = "false",
["Oms:TrackChangeSubscribe:Enabled"] = "false",
["Oms:TrackService:Enabled"] = "false",
["Oms:TrackService:BaseUrl"] = "http://127.0.0.1:9090",
["Oms:TrackService:AppKey"] = "E2E_KEY",

29
src/Oms.Tests/Messaging/TrackStatusOrderStatusMapperTests.cs Normal file
View File

@ -0,0 +1,29 @@
using FluentAssertions;
using Oms.Application.Messaging.TrackChange;
using Oms.Domain.OrdersAggregate.Consts;
namespace Oms.Tests.Messaging;
public class TrackStatusOrderStatusMapperTests
{
[Theory]
[InlineData("D03", OrderStatusCodes.Shipping)]
[InlineData("d03", OrderStatusCodes.Shipping)]
[InlineData("D09", OrderStatusCodes.DeliveryFailed)]
[InlineData("D11", OrderStatusCodes.DeliverySuccess)]
public void TryMapToOrderStatus_KnownCodes_Maps(string code, string expected)
{
TrackStatusOrderStatusMapper.TryMapToOrderStatus(code).Should().Be(expected);
}
[Theory]
[InlineData(null)]
[InlineData("")]
[InlineData(" ")]
[InlineData("A0701")]
[InlineData("D01")]
public void TryMapToOrderStatus_Unknown_ReturnsNull(string? code)
{
TrackStatusOrderStatusMapper.TryMapToOrderStatus(code).Should().BeNull();
}
}

3
src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs
View File

@ -76,7 +76,8 @@ internal sealed class SimpleAdminApiTestHost : IAsyncDisposable
["FeiShuNotice:EnableFeiShuExceptionSubscribe"] = "false",
["Redis:Host"] = "127.0.0.1:6379",
["Redis:Database"] = "0",
["OrderBatch:RabbitMq:Enabled"] = "false"
["OrderBatch:RabbitMq:Enabled"] = "false",
["Oms:TrackChangeSubscribe:Enabled"] = "false"
});
}, environmentName: "Testing");