From 377cc525d64a26d7e709b2ac0483c95feec06980 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Wed, 22 Jul 2026 12:00:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E5=AD=90=E9=A1=B9=E5=8F=8D=E9=A6=88=E4=B8=8E?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E5=8D=95=E5=8F=B7=E6=8C=81=E4=B9=85=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一成功/失败结果快照，补全客户单号与 Excel 行号；系统重试耗尽后标记失败避免子项长期停留在执行中；任务 API 返回 pending/running/cancelled 计数。 --- db.sql | 1 + scripts/order_batch_job_tables.sql | 1 + ...migrate_order_batch_job_item_client_no.sql | 3 + .../Dtos/Orders/OrderBatchJobDtos.cs | 16 + .../Dtos/Orders/OrderBatchPayloadDtos.cs | 2 + .../Messaging/OrderBatchRabbitMqOptions.cs | 5 + .../Orders/Batch/OrderBatchJobCoordinator.cs | 42 +- .../Orders/Batch/OrderBatchJobItemFeedback.cs | 332 ++++++++++++++++ .../Orders/Batch/OrderBatchJobProcessor.cs | 83 ++-- .../Services/Orders/OrderService.Helpers.cs | 1 + .../Services/Orders/OrderService.cs | 1 + .../Entities/OrderBatchJobItem.cs | 3 + src/Oms.Tests/OrderBatchJobProcessorTests.cs | 358 ++++++++++-------- 13 files changed, 651 insertions(+), 197 deletions(-) create mode 100644 sql/migrate_order_batch_job_item_client_no.sql create mode 100644 src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs diff --git a/db.sql b/db.sql index 94e35a5..098d82b 100644 --- a/db.sql +++ b/db.sql @@ -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，用于幂等与去重', diff --git a/scripts/order_batch_job_tables.sql b/scripts/order_batch_job_tables.sql index 2bd2e6a..c3e8577 100644 --- a/scripts/order_batch_job_tables.sql +++ b/scripts/order_batch_job_tables.sql @@ -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，用于幂等与去重', diff --git a/sql/migrate_order_batch_job_item_client_no.sql b/sql/migrate_order_batch_job_item_client_no.sql new file mode 100644 index 0000000..48aa261 --- /dev/null +++ b/sql/migrate_order_batch_job_item_client_no.sql @@ -0,0 +1,3 @@ +-- 批量任务明细：增加客户单号字段（Excel 导入/小包创建反馈用） +ALTER TABLE `t_order_batch_job_item` + ADD COLUMN `ClientNo` VARCHAR(64) NULL COMMENT '客户单号' AFTER `ServiceNo`; diff --git a/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs b/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs index 1036538..1e275d2 100644 --- a/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs @@ -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; } diff --git a/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs b/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs index 0b2d781..b1f6ed5 100644 --- a/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs @@ -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; } diff --git a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs index 7158cf3..4905570 100644 --- a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs +++ b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs @@ -11,4 +11,9 @@ public class OrderBatchRabbitMqOptions /// 是否启用订单批量消息发布与处理。 /// public bool Enabled { get; set; } + + /// + /// 单个子项系统异常的最大重试次数（耗尽后标记失败并 Ack，避免子项长期停留在执行中）。 + /// + public int MaxSystemRetries { get; set; } = 5; } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs index 7de45c3..cee1d9d 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs @@ -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> 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 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, diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs new file mode 100644 index 0000000..34d277c --- /dev/null +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobItemFeedback.cs @@ -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; + +/// +/// 批量任务子项反馈：统一补全标识、写入结果快照、映射 API 输出。 +/// +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(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(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(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(); +} diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index 2167735..691fca8 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -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; /// /// 消费 RabbitMQ 消息后更新任务状态并执行业务。 -/// 正常返回由 Fake EventBus Ack；抛异常则框架重试后进 DLX。 +/// 正常返回由 Fake EventBus Ack；系统失败可重试，耗尽后标记失败避免子项长期停留在执行中。 /// public sealed class OrderBatchJobProcessor( IOrderBatchJobRepository batchJobRepository, IOrderTaskRepository orderTaskRepository, + IOrderRepository orderRepository, IOrderBatchActionExecutor actionExecutor, + IOptions batchOptions, ILogger 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(job.PayloadJson); + var meta = Newtonsoft.Json.JsonConvert.DeserializeObject(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(message.PayloadJson); + var payload = Newtonsoft.Json.JsonConvert.DeserializeObject(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, } diff --git a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs index 1f4c09c..19fc734 100644 --- a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs +++ b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs @@ -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 { diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 8afe2a8..3b5aefa 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -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}", diff --git a/src/Oms.Domain/OrdersAggregate/Entities/OrderBatchJobItem.cs b/src/Oms.Domain/OrdersAggregate/Entities/OrderBatchJobItem.cs index ddfff27..0e0a3eb 100644 --- a/src/Oms.Domain/OrdersAggregate/Entities/OrderBatchJobItem.cs +++ b/src/Oms.Domain/OrdersAggregate/Entities/OrderBatchJobItem.cs @@ -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; diff --git a/src/Oms.Tests/OrderBatchJobProcessorTests.cs b/src/Oms.Tests/OrderBatchJobProcessorTests.cs index 61586e1..c84b9ab 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -1,151 +1,207 @@ -using FluentAssertions; -using Microsoft.Extensions.Logging.Abstractions; -using Moq; -using Oms.Application.Messaging; -using Oms.Application.Services.Orders.Batch; -using Oms.Domain.OrdersAggregate.Consts; -using Oms.Domain.OrdersAggregate.Entities; -using Oms.Domain.OrdersAggregate.Repositories; - -namespace Oms.Tests; - -public class OrderBatchJobProcessorTests -{ - private readonly Mock _batchJobRepository = new(MockBehavior.Strict); - private readonly Mock _orderTaskRepository = new(MockBehavior.Loose); - private readonly Mock _actionExecutor = new(MockBehavior.Strict); - private readonly OrderBatchJobProcessor _processor; - - public OrderBatchJobProcessorTests() - { - _processor = new OrderBatchJobProcessor( - _batchJobRepository.Object, - _orderTaskRepository.Object, - _actionExecutor.Object, - NullLogger.Instance); - } - - [Fact] - public async Task ProcessAsync_BusinessFailure_ShouldMarkFailed() - { - var item = CreateItem(id: 10, status: OrderBatchJobItemStatusCodes.Pending); - var message = CreateMessage(item.Id); - SetupItemLookup(item); - _actionExecutor - .Setup(x => x.ExecuteAsync(message, It.IsAny())) - .ReturnsAsync(OrderBatchActionExecuteResult.BusinessFail("订单不存在")); - SetupJobRefresh(item.JobId, [item]); - - await _processor.ProcessAsync(message); - - item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed); - item.FailReason.Should().Be("订单不存在"); - } - - [Fact] - public async Task ProcessAsync_SystemFailure_ShouldThrow() - { - var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending); - var message = CreateMessage(item.Id); - SetupItemLookup(item); - _actionExecutor - .Setup(x => x.ExecuteAsync(message, It.IsAny())) - .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); - - var act = () => _processor.ProcessAsync(message); - - await act.Should().ThrowAsync() - .WithMessage("*下游超时*"); - } - - [Fact] - public async Task ProcessAsync_CancelledItem_ShouldSkipWithoutExecuting() - { - var item = CreateItem(id: 13, status: OrderBatchJobItemStatusCodes.Cancelled); - var message = CreateMessage(item.Id); - _batchJobRepository - .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny())) - .ReturnsAsync(item); - - await _processor.ProcessAsync(message); - - _actionExecutor.Verify( - x => x.ExecuteAsync(It.IsAny(), It.IsAny()), - Times.Never); - } - - [Fact] - public async Task ProcessAsync_FireAndForgetFailure_ShouldNotThrow() - { - var message = new OrderBatchJobMessage - { - JobId = 0, - JobItemId = 0, - Action = OrderBatchActionCodes.TrackUpsertSmallBag, - OrderId = 99, - MessageId = "track:99:x", - }; - _actionExecutor - .Setup(x => x.ExecuteAsync(message, It.IsAny())) - .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); - - await _processor.ProcessAsync(message); - - _batchJobRepository.Verify( - x => x.GetItemByIdAsync(It.IsAny(), It.IsAny()), - Times.Never); - } - - private void SetupItemLookup(OrderBatchJobItem item) - { - _batchJobRepository - .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny())) - .ReturnsAsync(item); - _batchJobRepository - .Setup(x => x.UpdateItemAsync(item, It.IsAny())) - .Returns(Task.CompletedTask); - } - - private void SetupJobRefresh(int jobId, List items) - { - var job = new OrderBatchJob - { - Id = jobId, - Action = OrderBatchActionCodes.Pay, - Status = OrderBatchJobStatusCodes.Running, - Total = items.Count, - }; - - _batchJobRepository - .Setup(x => x.GetByIdAsync(jobId, It.IsAny())) - .ReturnsAsync(job); - _batchJobRepository - .Setup(x => x.GetItemsByJobIdAsync(jobId, It.IsAny())) - .ReturnsAsync(items); - _batchJobRepository - .Setup(x => x.UpdateJobAsync(job, It.IsAny())) - .Returns(Task.CompletedTask); - } - - private static OrderBatchJobItem CreateItem(int id, string status, int retryCount = 0) => - new() - { - Id = id, - JobId = 100, - OrderId = 2000 + id, - Action = OrderBatchActionCodes.Pay, - Status = status, - RetryCount = retryCount, - }; - - private static OrderBatchJobMessage CreateMessage(int jobItemId) => - new() - { - JobId = 100, - JobItemId = jobItemId, - Action = OrderBatchActionCodes.Pay, - OrderId = 2000 + jobItemId, - OperatorUserId = 1, - MessageId = $"100:{jobItemId}:pay", - }; -} +using FluentAssertions; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Moq; +using Oms.Application.Messaging; +using Oms.Application.Services.Orders.Batch; +using Oms.Domain.OrdersAggregate.Consts; +using Oms.Domain.OrdersAggregate.Entities; +using Oms.Domain.OrdersAggregate.Repositories; + +public class OrderBatchJobProcessorTests +{ + private readonly Mock _batchJobRepository = new(MockBehavior.Strict); + private readonly Mock _orderTaskRepository = new(MockBehavior.Loose); + private readonly Mock _orderRepository = new(MockBehavior.Loose); + private readonly Mock _actionExecutor = new(MockBehavior.Strict); + private readonly OrderBatchJobProcessor _processor; + + public OrderBatchJobProcessorTests() + { + _processor = new OrderBatchJobProcessor( + _batchJobRepository.Object, + _orderTaskRepository.Object, + _orderRepository.Object, + _actionExecutor.Object, + Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 5 }), + NullLogger.Instance); + } + + [Fact] + public async Task ProcessAsync_BusinessFailure_ShouldMarkFailed() + { + var item = CreateItem(id: 10, status: OrderBatchJobItemStatusCodes.Pending); + var message = CreateMessage(item.Id); + SetupItemLookup(item); + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .ReturnsAsync(OrderBatchActionExecuteResult.BusinessFail("订单不存在")); + SetupJobRefresh(item.JobId, [item]); + + await _processor.ProcessAsync(message); + + item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed); + item.FailReason.Should().Be("订单不存在"); + item.ResultJson.Should().Contain("订单不存在"); + } + + [Fact] + public async Task ProcessAsync_SystemFailure_ShouldThrowForRetry() + { + var item = CreateItem(id: 11, status: OrderBatchJobItemStatusCodes.Pending); + var message = CreateMessage(item.Id); + SetupItemLookup(item); + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); + + var act = () => _processor.ProcessAsync(message); + + await act.Should().ThrowAsync() + .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.Instance); + + var item = CreateItem(id: 14, status: OrderBatchJobItemStatusCodes.Pending); + var message = CreateMessage(item.Id); + SetupItemLookup(item); + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .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] + public async Task ProcessAsync_CancelledItem_ShouldSkipWithoutExecuting() + { + var item = CreateItem(id: 13, status: OrderBatchJobItemStatusCodes.Cancelled); + var message = CreateMessage(item.Id); + _batchJobRepository + .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny())) + .ReturnsAsync(item); + + await _processor.ProcessAsync(message); + + _actionExecutor.Verify( + x => x.ExecuteAsync(It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task ProcessAsync_FireAndForgetFailure_ShouldNotThrow() + { + var message = new OrderBatchJobMessage + { + JobId = 0, + JobItemId = 0, + Action = OrderBatchActionCodes.TrackUpsertSmallBag, + OrderId = 99, + MessageId = "track:99:x", + }; + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时")); + + await _processor.ProcessAsync(message); + + _batchJobRepository.Verify( + x => x.GetItemByIdAsync(It.IsAny(), It.IsAny()), + 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())) + .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 + .Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny())) + .ReturnsAsync(item); + _batchJobRepository + .Setup(x => x.UpdateItemAsync(item, It.IsAny())) + .Returns(Task.CompletedTask); + } + + private void SetupJobRefresh(int jobId, List items) + { + var job = new OrderBatchJob + { + Id = jobId, + Action = OrderBatchActionCodes.Pay, + Status = OrderBatchJobStatusCodes.Running, + Total = items.Count, + }; + + _batchJobRepository + .Setup(x => x.GetByIdAsync(jobId, It.IsAny())) + .ReturnsAsync(job); + _batchJobRepository + .Setup(x => x.GetItemsByJobIdAsync(jobId, It.IsAny())) + .ReturnsAsync(items); + _batchJobRepository + .Setup(x => x.UpdateJobAsync(job, It.IsAny())) + .Returns(Task.CompletedTask); + } + + private static OrderBatchJobItem CreateItem(int id, string status, int retryCount = 0) => + new() + { + Id = id, + JobId = 100, + OrderId = 2000 + id, + Action = OrderBatchActionCodes.Pay, + Status = status, + RetryCount = retryCount, + }; + + private static OrderBatchJobMessage CreateMessage(int jobItemId) => + new() + { + JobId = 100, + JobItemId = jobItemId, + Action = OrderBatchActionCodes.Pay, + OrderId = 2000 + jobItemId, + OperatorUserId = 1, + MessageId = $"100:{jobItemId}:pay", + }; +}