feat: 增强批量任务子项反馈与客户单号持久化

统一成功/失败结果快照,补全客户单号与 Excel 行号;系统重试耗尽后标记失败避免子项长期停留在执行中;任务 API 返回 pending/running/cancelled 计数。
This commit is contained in:
zzh 2026-07-22 12:00:53 +08:00
parent fc92e2903b
commit 377cc525d6
13 changed files with 651 additions and 197 deletions

1
db.sql
View File

@ -288,6 +288,7 @@ CREATE TABLE `t_order_batch_job_item` (
`JobId` INT NOT NULL COMMENT '批量任务ID',
`OrderId` INT NOT NULL COMMENT '订单ID',
`ServiceNo` VARCHAR(64) NULL COMMENT '运单号',
`ClientNo` VARCHAR(64) NULL COMMENT '客户单号',
`Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '子项操作类型,通常与任务 Action 一致',
`Status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT '子项状态:pending/running/success/failed/cancelled',
`MessageId` VARCHAR(128) NULL COMMENT 'RabbitMQ 消息ID,用于幂等与去重',

View File

@ -38,6 +38,7 @@ CREATE TABLE `t_order_batch_job_item` (
`JobId` INT NOT NULL COMMENT '批量任务ID',
`OrderId` INT NOT NULL COMMENT '订单ID',
`ServiceNo` VARCHAR(64) NULL COMMENT '运单号',
`ClientNo` VARCHAR(64) NULL COMMENT '客户单号',
`Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '子项操作类型,通常与任务 Action 一致',
`Status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT '子项状态:pending/running/success/failed/cancelled',
`MessageId` VARCHAR(128) NULL COMMENT 'RabbitMQ 消息ID,用于幂等与去重',

View File

@ -0,0 +1,3 @@
-- 批量任务明细:增加客户单号字段(Excel 导入/小包创建反馈用)
ALTER TABLE `t_order_batch_job_item`
ADD COLUMN `ClientNo` VARCHAR(64) NULL COMMENT '客户单号' AFTER `ServiceNo`;

View File

@ -29,6 +29,12 @@ public class OrderBatchJobOutput
public int FailCount { get; set; }
public int PendingCount { get; set; }
public int RunningCount { get; set; }
public int CancelledCount { get; set; }
public DateTime? StartedAt { get; set; }
public DateTime? FinishedAt { get; set; }
@ -44,12 +50,22 @@ public class OrderBatchJobItemOutput
public string? ServiceNo { get; set; }
public string? ClientNo { get; set; }
public int? RowNumber { get; set; }
public string Action { get; set; } = string.Empty;
public string Status { get; set; } = string.Empty;
public string? FailReason { get; set; }
public string? ResultJson { get; set; }
public string? ResultMessage { get; set; }
public int RetryCount { get; set; }
public DateTime? StartedAt { get; set; }
public DateTime? FinishedAt { get; set; }

View File

@ -9,6 +9,8 @@ public class OrderBatchPayloadItemInput
public string? ServiceNo { get; set; }
public string? ClientNo { get; set; }
public string PayloadJson { get; set; } = string.Empty;
public string? ItemKey { get; set; }

View File

@ -11,4 +11,9 @@ public class OrderBatchRabbitMqOptions
/// 是否启用订单批量消息发布与处理。
/// </summary>
public bool Enabled { get; set; }
/// <summary>
/// 单个子项系统异常的最大重试次数(耗尽后标记失败并 Ack,避免子项长期停留在执行中)。
/// </summary>
public int MaxSystemRetries { get; set; } = 5;
}

View File

@ -113,6 +113,7 @@ public sealed class OrderBatchJobCoordinator(
JobId = job.Id,
OrderId = orderId,
ServiceNo = order?.ServiceNo,
ClientNo = order?.ClientNo,
Action = action,
Status = OrderBatchJobItemStatusCodes.Pending,
MessageId = $"{job.Id}:{orderId}:{action}",
@ -232,6 +233,7 @@ public sealed class OrderBatchJobCoordinator(
JobId = job.Id,
OrderId = source.OrderId,
ServiceNo = source.ServiceNo,
ClientNo = source.ClientNo,
Action = action,
Status = OrderBatchJobItemStatusCodes.Pending,
MessageId = itemKey,
@ -309,18 +311,14 @@ public sealed class OrderBatchJobCoordinator(
public async Task<List<OrderBatchJobItemOutput>> GetJobItemsAsync(int jobId, CancellationToken cancellationToken = default)
{
var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken);
var jobAction = job?.Action ?? string.Empty;
var items = await batchJobRepository.GetItemsByJobIdAsync(jobId, cancellationToken);
return items.Select(x => new OrderBatchJobItemOutput
{
Id = x.Id,
OrderId = x.OrderId,
ServiceNo = x.ServiceNo,
Status = x.Status,
FailReason = x.FailReason,
ResultJson = x.ResultJson,
StartedAt = x.StartedAt,
FinishedAt = x.FinishedAt,
}).ToList();
return items
.Select(x => OrderBatchJobItemFeedback.MapToOutput(x, jobAction))
.OrderBy(x => x.RowNumber ?? int.MaxValue)
.ThenBy(x => x.Id)
.ToList();
}
public async Task<OrderBatchJobManageOutput> RetryFailedAsync(
@ -407,6 +405,14 @@ public sealed class OrderBatchJobCoordinator(
item.FinishedAt = now;
item.UpdateTime = now;
item.UpdateUserId = operatorUserId;
item.ResultJson = Newtonsoft.Json.JsonConvert.SerializeObject(new
{
success = false,
action = item.Action,
reason = item.FailReason,
clientNo = item.ClientNo,
orderId = item.OrderId > 0 ? item.OrderId : (int?)null,
});
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
}
@ -440,6 +446,9 @@ public sealed class OrderBatchJobCoordinator(
var items = await batchJobRepository.GetItemsByJobIdAsync(job.Id, cancellationToken);
var successCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Success);
var failCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Failed);
var pendingCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Pending);
var runningCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Running);
var cancelledCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Cancelled);
if (job.SuccessCount != successCount || job.FailCount != failCount)
{
@ -452,9 +461,11 @@ public sealed class OrderBatchJobCoordinator(
or OrderBatchJobItemStatusCodes.Failed
or OrderBatchJobItemStatusCodes.Cancelled))
{
job.Status = failCount > 0 && successCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
job.Status = cancelledCount == items.Count
? OrderBatchJobStatusCodes.Cancelled
: failCount > 0 && successCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
job.FinishedAt ??= DateTime.Now;
await batchJobRepository.UpdateJobAsync(job, cancellationToken);
}
@ -467,6 +478,9 @@ public sealed class OrderBatchJobCoordinator(
Total = job.Total,
SuccessCount = job.SuccessCount,
FailCount = job.FailCount,
PendingCount = pendingCount,
RunningCount = runningCount,
CancelledCount = cancelledCount,
StartedAt = job.StartedAt,
FinishedAt = job.FinishedAt,
CreateTime = job.CreateTime,

View File

@ -0,0 +1,332 @@
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Oms.Application.Dtos.Orders;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Application.Services.Orders.Batch;
/// <summary>
/// 批量任务子项反馈:统一补全标识、写入结果快照、映射 API 输出。
/// </summary>
public static class OrderBatchJobItemFeedback
{
private static readonly string[] ClientNoRowKeys =
[
"clientNo", "ClientNo", "customerOrderNo", "CustomerOrderNo",
"客户单号", "客户订单号", "外部单号",
];
public static async Task EnrichIdentifiersAsync(
OrderBatchJobItem item,
string action,
IOrderRepository orderRepository,
CancellationToken cancellationToken = default)
{
EnrichFromPayload(item, action);
if (item.OrderId <= 0)
{
return;
}
if (!string.IsNullOrWhiteSpace(item.ClientNo) && !string.IsNullOrWhiteSpace(item.ServiceNo))
{
return;
}
var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == item.OrderId, cancellationToken);
if (order == null)
{
return;
}
item.ClientNo ??= NullIfWhiteSpace(order.ClientNo);
item.ServiceNo ??= NullIfWhiteSpace(order.ServiceNo);
}
public static void EnrichFromPayload(OrderBatchJobItem item, string action)
{
if (string.IsNullOrWhiteSpace(item.PayloadJson))
{
return;
}
if (string.Equals(action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase))
{
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(item.PayloadJson);
if (payload == null)
{
return;
}
if (string.IsNullOrWhiteSpace(item.ClientNo))
{
item.ClientNo = TryExtractClientNoFromRowJson(payload.RowDataJson);
}
return;
}
if (string.Equals(action, OrderBatchActionCodes.ParcelCreate, StringComparison.OrdinalIgnoreCase))
{
var input = JsonConvert.DeserializeObject<OrderParcelCreateInput>(item.PayloadJson);
if (input == null)
{
return;
}
item.ClientNo ??= NullIfWhiteSpace(input.ClientNo);
}
}
public static int? TryGetImportRowNumber(OrderBatchJobItem item, string action)
{
if (!string.Equals(action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase)
|| string.IsNullOrWhiteSpace(item.PayloadJson))
{
return null;
}
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(item.PayloadJson);
return payload is { RowNumber: > 0 } ? payload.RowNumber : null;
}
public static void ApplySuccess(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
item.Status = OrderBatchJobItemStatusCodes.Success;
item.FailReason = null;
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
if (!string.IsNullOrWhiteSpace(result.ClientNo))
{
item.ClientNo = result.ClientNo.Trim();
}
if (!string.IsNullOrWhiteSpace(result.ServiceNo))
{
item.ServiceNo = result.ServiceNo.Trim();
}
item.ResultJson = BuildSuccessResultJson(item, result, action);
}
public static void ApplyBusinessFailure(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = string.IsNullOrWhiteSpace(result.FailReason) ? "操作失败" : result.FailReason.Trim();
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
item.ResultJson = BuildFailureResultJson(item, action, item.FailReason, retryCount: item.RetryCount);
}
public static void ApplySystemRetry(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action,
int maxRetries)
{
item.RetryCount += 1;
var reason = string.IsNullOrWhiteSpace(result.FailReason) ? "系统异常" : result.FailReason.Trim();
item.FailReason = $"系统异常(第 {item.RetryCount}/{maxRetries} 次重试):{reason}";
if (item.RetryCount >= maxRetries)
{
ApplySystemFailure(item, action, $"系统重试已达上限:{reason}");
return;
}
item.Status = OrderBatchJobItemStatusCodes.Pending;
item.ResultJson = BuildFailureResultJson(item, action, item.FailReason, retryCount: item.RetryCount, retrying: true);
}
public static void ApplySystemFailure(OrderBatchJobItem item, string action, string reason)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = reason;
item.ResultJson = BuildFailureResultJson(item, action, reason, retryCount: item.RetryCount);
}
public static OrderBatchJobItemOutput MapToOutput(OrderBatchJobItem item, string jobAction)
{
EnrichFromPayload(item, jobAction);
return new OrderBatchJobItemOutput
{
Id = item.Id,
OrderId = item.OrderId,
ServiceNo = item.ServiceNo,
ClientNo = item.ClientNo,
RowNumber = TryGetImportRowNumber(item, jobAction),
Action = string.IsNullOrWhiteSpace(item.Action) ? jobAction : item.Action,
Status = item.Status,
FailReason = item.FailReason,
ResultJson = item.ResultJson,
ResultMessage = BuildResultMessage(item),
RetryCount = item.RetryCount,
StartedAt = item.StartedAt,
FinishedAt = item.FinishedAt,
};
}
public static string? BuildResultMessage(OrderBatchJobItem item)
{
if (item.Status == OrderBatchJobItemStatusCodes.Success)
{
return TryReadResultField(item.ResultJson, "message") ?? "操作成功";
}
if (item.Status == OrderBatchJobItemStatusCodes.Failed)
{
return item.FailReason;
}
if (item.Status == OrderBatchJobItemStatusCodes.Cancelled)
{
return item.FailReason ?? "已取消";
}
if (item.Status == OrderBatchJobItemStatusCodes.Running)
{
return "处理中";
}
return null;
}
private static string BuildSuccessResultJson(
OrderBatchJobItem item,
OrderBatchActionExecuteResult result,
string action)
{
if (!string.IsNullOrWhiteSpace(result.ResultJson))
{
return result.ResultJson;
}
return JsonConvert.SerializeObject(new
{
success = true,
action,
message = ResolveSuccessMessage(action),
clientNo = item.ClientNo ?? result.ClientNo,
serviceNo = item.ServiceNo ?? result.ServiceNo,
orderId = item.OrderId > 0 ? item.OrderId : result.CreatedOrderId,
trackNo = result.TrackNo,
labelUrl = result.LabelUrl,
rowNumber = TryGetImportRowNumber(item, action),
});
}
private static string BuildFailureResultJson(
OrderBatchJobItem item,
string action,
string reason,
int retryCount,
bool retrying = false)
{
return JsonConvert.SerializeObject(new
{
success = false,
action,
retrying,
retryCount,
reason,
clientNo = item.ClientNo,
serviceNo = item.ServiceNo,
orderId = item.OrderId > 0 ? item.OrderId : (int?)null,
rowNumber = TryGetImportRowNumber(item, action),
});
}
private static string ResolveSuccessMessage(string action) =>
action switch
{
OrderBatchActionCodes.ImportRow => "导入成功",
OrderBatchActionCodes.ParcelCreate => "创建成功",
OrderBatchActionCodes.Pay => "支付成功",
OrderBatchActionCodes.PreDeclare => "预报成功",
OrderBatchActionCodes.Intercept => "拦截成功",
OrderBatchActionCodes.CancelIntercept => "取消拦截成功",
OrderBatchActionCodes.Delete => "删除成功",
OrderBatchActionCodes.Print => "打印成功",
OrderBatchActionCodes.GenerateLabelPdf => "面单生成成功",
OrderBatchActionCodes.MarkProblematic => "问题件创建成功",
OrderBatchActionCodes.MarkClaimed => "理赔件创建成功",
OrderBatchActionCodes.HandleStatusUpdate => "处理状态更新成功",
OrderBatchActionCodes.DownstreamForwardCreateOne => "下游预报成功",
_ => "操作成功",
};
private static string? TryExtractClientNoFromRowJson(string? rowDataJson)
{
if (string.IsNullOrWhiteSpace(rowDataJson))
{
return null;
}
try
{
if (JsonConvert.DeserializeObject(rowDataJson) is not JObject obj)
{
return null;
}
foreach (var key in ClientNoRowKeys)
{
if (!obj.TryGetValue(key, StringComparison.OrdinalIgnoreCase, out var token))
{
continue;
}
var value = token?.ToString()?.Trim();
if (!string.IsNullOrWhiteSpace(value))
{
return value;
}
}
}
catch
{
// ignore malformed row snapshot
}
return null;
}
private static string? TryReadResultField(string? resultJson, string field)
{
if (string.IsNullOrWhiteSpace(resultJson))
{
return null;
}
try
{
var obj = JObject.Parse(resultJson);
return obj[field]?.ToString()?.Trim();
}
catch
{
return null;
}
}
private static string? NullIfWhiteSpace(string? value) =>
string.IsNullOrWhiteSpace(value) ? null : value.Trim();
}

View File

@ -1,6 +1,6 @@
using Fake.DependencyInjection;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Microsoft.Extensions.Options;
using Oms.Application.Dtos.Orders;
using Oms.Application.Messaging;
using Oms.Domain.OrdersAggregate.Consts;
@ -11,18 +11,21 @@ namespace Oms.Application.Services.Orders.Batch;
/// <summary>
/// 消费 RabbitMQ 消息后更新任务状态并执行业务。
/// 正常返回由 Fake EventBus Ack;抛异常则框架重试后进 DLX。
/// 正常返回由 Fake EventBus Ack;系统失败可重试,耗尽后标记失败避免子项长期停留在执行中。
/// </summary>
public sealed class OrderBatchJobProcessor(
IOrderBatchJobRepository batchJobRepository,
IOrderTaskRepository orderTaskRepository,
IOrderRepository orderRepository,
IOrderBatchActionExecutor actionExecutor,
IOptions<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobProcessor> logger) : IScopedDependency
{
public async Task ProcessAsync(
OrderBatchJobMessage message,
CancellationToken cancellationToken = default)
{
var maxSystemRetries = Math.Max(1, batchOptions.Value.MaxSystemRetries);
OrderBatchJobItem? item = null;
if (message.JobItemId > 0)
{
@ -54,7 +57,6 @@ public sealed class OrderBatchJobProcessor(
if (message.JobItemId <= 0)
{
// fire-and-forget(如轨迹同步):失败只记日志,不抛、不进 DLX
if (!result.Success)
{
logger.LogWarning(
@ -75,25 +77,12 @@ public sealed class OrderBatchJobProcessor(
item.FinishedAt = DateTime.Now;
item.UpdateTime = DateTime.Now;
var action = string.IsNullOrWhiteSpace(item.Action) ? message.Action : item.Action;
if (result.Success)
{
item.Status = OrderBatchJobItemStatusCodes.Success;
item.FailReason = null;
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
if (!string.IsNullOrWhiteSpace(result.ServiceNo))
{
item.ServiceNo = result.ServiceNo;
}
if (!string.IsNullOrWhiteSpace(result.ResultJson))
{
item.ResultJson = result.ResultJson;
}
OrderBatchJobItemFeedback.ApplySuccess(item, result, action);
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
await RefreshJobSummaryAsync(item.JobId, cancellationToken);
return;
@ -101,9 +90,15 @@ public sealed class OrderBatchJobProcessor(
if (result.IsBusinessFailure)
{
item.Status = OrderBatchJobItemStatusCodes.Failed;
item.FailReason = result.FailReason;
if (result.CreatedOrderId is > 0)
{
item.OrderId = result.CreatedOrderId.Value;
}
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
OrderBatchJobItemFeedback.ApplyBusinessFailure(item, result, action);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
if (message.Action == OrderBatchActionCodes.ImportRow)
{
await RecordImportTaskFailAsync(message, result.FailReason, cancellationToken);
@ -113,8 +108,23 @@ public sealed class OrderBatchJobProcessor(
return;
}
// 系统失败:抛出让 EventBus 重试,耗尽后进 DLX
throw new InvalidOperationException(result.FailReason ?? "批量任务系统失败");
await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
OrderBatchJobItemFeedback.ApplySystemRetry(item, result, action, maxSystemRetries);
await batchJobRepository.UpdateItemAsync(item, cancellationToken);
if (item.Status == OrderBatchJobItemStatusCodes.Failed)
{
logger.LogError(
"Batch job item {JobItemId} action={Action} failed after {RetryCount} system retries: {Reason}",
item.Id,
action,
item.RetryCount,
item.FailReason);
await RefreshJobSummaryAsync(item.JobId, cancellationToken);
return;
}
throw new InvalidOperationException(item.FailReason ?? result.FailReason ?? "批量任务系统失败");
}
private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken)
@ -134,9 +144,18 @@ public sealed class OrderBatchJobProcessor(
or OrderBatchJobItemStatusCodes.Failed
or OrderBatchJobItemStatusCodes.Cancelled))
{
job.Status = job.FailCount > 0 && job.SuccessCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
var cancelledCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Cancelled);
if (cancelledCount == items.Count)
{
job.Status = OrderBatchJobStatusCodes.Cancelled;
}
else
{
job.Status = job.FailCount > 0 && job.SuccessCount == 0
? OrderBatchJobStatusCodes.Failed
: OrderBatchJobStatusCodes.Completed;
}
job.FinishedAt = DateTime.Now;
}
else
@ -155,7 +174,7 @@ public sealed class OrderBatchJobProcessor(
return;
}
var meta = JsonConvert.DeserializeObject<OrderBatchImportJobPayload>(job.PayloadJson);
var meta = Newtonsoft.Json.JsonConvert.DeserializeObject<OrderBatchImportJobPayload>(job.PayloadJson);
if (meta == null || meta.TaskId <= 0)
{
return;
@ -171,7 +190,7 @@ public sealed class OrderBatchJobProcessor(
task.FailCount = job.FailCount;
task.UpdateTime = DateTime.Now;
if (job.Status is OrderBatchJobStatusCodes.Completed or OrderBatchJobStatusCodes.Failed)
if (job.Status is OrderBatchJobStatusCodes.Completed or OrderBatchJobStatusCodes.Failed or OrderBatchJobStatusCodes.Cancelled)
{
task.Status = OrderTaskStatusCodes.Completed;
}
@ -193,7 +212,7 @@ public sealed class OrderBatchJobProcessor(
return;
}
var payload = JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(message.PayloadJson);
var payload = Newtonsoft.Json.JsonConvert.DeserializeObject<OrderBatchImportRowPayload>(message.PayloadJson);
if (payload == null || payload.TaskId <= 0)
{
return;
@ -202,7 +221,7 @@ public sealed class OrderBatchJobProcessor(
object? rawData;
try
{
rawData = JsonConvert.DeserializeObject(payload.RowDataJson) ?? payload.RowDataJson;
rawData = Newtonsoft.Json.JsonConvert.DeserializeObject(payload.RowDataJson) ?? payload.RowDataJson;
}
catch
{
@ -215,7 +234,7 @@ public sealed class OrderBatchJobProcessor(
TaskId = payload.TaskId,
RowNumber = payload.RowNumber,
Reason = reason ?? "导入失败",
RawDataJson = JsonConvert.SerializeObject(rawData),
RawDataJson = Newtonsoft.Json.JsonConvert.SerializeObject(rawData),
CreateUserId = message.OperatorUserId > 0 ? message.OperatorUserId : 0,
CreateTime = DateTime.Now,
}

View File

@ -403,6 +403,7 @@ public partial class OrderService
{
OrderId = 0,
ServiceNo = parcelInput.ServiceNo,
ClientNo = parcelInput.ClientNo,
ItemKey = $"import:{task.Id}:{row.RowNumber}",
PayloadJson = JsonConvert.SerializeObject(new OrderBatchImportRowPayload
{

View File

@ -412,6 +412,7 @@ public partial class OrderService : ApplicationService
{
OrderId = 0,
ServiceNo = request.ServiceNo,
ClientNo = request.ClientNo,
ItemKey = !string.IsNullOrWhiteSpace(request.RequestId)
? $"parcel:{request.RequestId}"
: $"parcel:{batchNo}:{index + 1}",

View File

@ -13,6 +13,9 @@ public class OrderBatchJobItem : OmsAuditedEntity
[SugarColumn(Length = 64, IsNullable = true)]
public string? ServiceNo { get; set; }
[SugarColumn(Length = 64, IsNullable = true)]
public string? ClientNo { get; set; }
public string Action { get; set; } = string.Empty;
public string Status { get; set; } = string.Empty;

View File

@ -1,5 +1,6 @@
using FluentAssertions;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Moq;
using Oms.Application.Messaging;
using Oms.Application.Services.Orders.Batch;
@ -7,12 +8,11 @@ using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;
namespace Oms.Tests;
public class OrderBatchJobProcessorTests
{
private readonly Mock<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
private readonly Mock<IOrderTaskRepository> _orderTaskRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchActionExecutor> _actionExecutor = new(MockBehavior.Strict);
private readonly OrderBatchJobProcessor _processor;
@ -21,7 +21,9 @@ public class OrderBatchJobProcessorTests
_processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderTaskRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 5 }),
NullLogger<OrderBatchJobProcessor>.Instance);
}
@ -40,10 +42,11 @@ public class OrderBatchJobProcessorTests
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.FailReason.Should().Be("订单不存在");
item.ResultJson.Should().Contain("订单不存在");
}
[Fact]
public async Task ProcessAsync_SystemFailure_ShouldThrow()
public async Task ProcessAsync_SystemFailure_ShouldThrowForRetry()
{
var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
@ -56,6 +59,34 @@ public class OrderBatchJobProcessorTests
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*下游超时*");
item.RetryCount.Should().Be(1);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
}
[Fact]
public async Task ProcessAsync_SystemFailureExhausted_ShouldMarkFailedWithoutThrow()
{
var processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderTaskRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 1 }),
NullLogger<OrderBatchJobProcessor>.Instance);
var item = CreateItem(id: 14, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
SetupJobRefresh(item.JobId, [item]);
await processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.RetryCount.Should().Be(1);
item.FailReason.Should().Contain("系统重试已达上限");
}
[Fact]
@ -96,6 +127,31 @@ public class OrderBatchJobProcessorTests
Times.Never);
}
[Fact]
public async Task ProcessAsync_Success_ShouldPersistClientNo()
{
var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending);
item.Action = OrderBatchActionCodes.ImportRow;
var message = CreateMessage(item.Id);
message.Action = OrderBatchActionCodes.ImportRow;
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.Ok(
createdOrderId: 9001,
serviceNo: "ID20260722001",
clientNo: "CUST-001"));
SetupJobRefresh(item.JobId, [item]);
await _processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Success);
item.OrderId.Should().Be(9001);
item.ClientNo.Should().Be("CUST-001");
item.ServiceNo.Should().Be("ID20260722001");
item.ResultJson.Should().Contain("CUST-001");
}
private void SetupItemLookup(OrderBatchJobItem item)
{
_batchJobRepository