diff --git a/db.sql b/db.sql index 94e35a5..b1963c6 100644 --- a/db.sql +++ b/db.sql @@ -91,8 +91,6 @@ DROP TABLE IF EXISTS `t_order_weight`; DROP TABLE IF EXISTS `t_order_declaration`; DROP TABLE IF EXISTS `t_order_recipient`; DROP TABLE IF EXISTS `t_order_sender`; -DROP TABLE IF EXISTS `t_order_task_fail`; -DROP TABLE IF EXISTS `t_order_task`; DROP TABLE IF EXISTS `t_order_batch_job_item`; DROP TABLE IF EXISTS `t_order_batch_job`; DROP TABLE IF EXISTS `t_order_template`; @@ -224,42 +222,6 @@ CREATE TABLE `t_order_template` ( KEY `idx_ord_tpl_customer_type` (`CustomerId`, `Type`) ) COMMENT='下单模板表'; -CREATE TABLE `t_order_task` ( - `Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1', - `TaskName` VARCHAR(128) NOT NULL COMMENT '任务名称', - `Quantity` INT NULL COMMENT '数量', - `SuccessCount` INT NULL COMMENT '成功数', - `FailCount` INT NULL COMMENT '失败数', - `Status` VARCHAR(16) NOT NULL COMMENT '任务状态：processing/completed/pending', - `TemplateId` INT NULL COMMENT '模板ID', - `CustomerId` INT NULL COMMENT '客户ID', - `IsDeleted` TINYINT NOT NULL DEFAULT 0 COMMENT '软删除：0-否，1-是', - `ConcurrencyStamp` CHAR(36) NOT NULL DEFAULT (UUID()) COMMENT '并发戳', - `CreateUserId` INT NOT NULL DEFAULT 0 COMMENT '创建人', - `CreateTime` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', - `UpdateUserId` INT NULL COMMENT '更新人', - `UpdateTime` DATETIME NULL COMMENT '更新时间', - PRIMARY KEY (`Id`), - KEY `idx_ord_task_customer_status_ctime` (`CustomerId`, `Status`, `CreateTime`), - KEY `idx_ord_task_template` (`TemplateId`) -) COMMENT='下单任务表'; - -CREATE TABLE `t_order_task_fail` ( - `Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1', - `TaskId` INT NOT NULL COMMENT '任务ID', - `RowNumber` INT NOT NULL COMMENT 'Excel行号', - `Reason` VARCHAR(512) NOT NULL COMMENT '失败原因', - `RawDataJson` JSON NULL COMMENT '原始行数据', - `IsDeleted` TINYINT NOT NULL DEFAULT 0 COMMENT '软删除：0-否，1-是', - `CreateUserId` INT NOT NULL DEFAULT 0 COMMENT '创建人', - `CreateTime` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', - `UpdateUserId` INT NULL COMMENT '更新人', - `UpdateTime` DATETIME NULL COMMENT '更新时间', - PRIMARY KEY (`Id`), - KEY `idx_ord_task_fail_task` (`TaskId`), - KEY `idx_ord_task_fail_row` (`TaskId`, `RowNumber`) -) COMMENT='下单任务失败明细表'; - CREATE TABLE `t_order_batch_job` ( `Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键，自增，步长 1', `Action` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '批量操作类型：pay/pre-declare/intercept/parcel-create 等', @@ -288,6 +250,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/design.md b/design.md index add954e..736846a 100644 --- a/design.md +++ b/design.md @@ -65,13 +65,13 @@ - 我的订单：列表、详情、支付、预报、拦截、取消拦截、删除、打印、导出 - 我要查价：按目的国、运输方式、重量、货物类型返回价格方案 -- 我要下单：模板下载、Excel 导入、任务跟踪 +- 我要下单：模板下载、Excel 导入、批量任务跟踪 核心表： - `t_order` - `t_order_template` -- `t_order_task` +- `t_order_batch_job` / `t_order_batch_job_item` - `t_order_sender` - `t_order_recipient` - `t_order_declaration` @@ -403,10 +403,9 @@ OMS 规格文档展示的是 kebab-case 路径示例，例如： 建议至少拆分为以下应用服务： -- `OrderService` +- `OrderService`（含批量任务 `OrderService.BatchJob`） - `OrderPriceCheckService` - `OrderTemplateService` -- `OrderTaskService` - `SalesProductService` - `SalesProductPriceService` - `SalesProductAllocationService` @@ -432,7 +431,7 @@ OMS 规格文档展示的是 kebab-case 路径示例，例如： 以下场景必须显式事务： - 创建订单时同时写主表、地址、申报、件重尺、费用 -- 批量导入任务落库与任务结果回写 +- 批量任务落库与子项结果回写（`t_order_batch_job` / `t_order_batch_job_item`） - 保存价格项时同时覆盖因素与价格明细 - 发布与回滚时版本快照回写 - 充值审核通过并更新余额 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_drop_order_task_merge_batch.sql b/sql/migrate_drop_order_task_merge_batch.sql new file mode 100644 index 0000000..722cd2f --- /dev/null +++ b/sql/migrate_drop_order_task_merge_batch.sql @@ -0,0 +1,6 @@ +-- 合并下单任务表到批量任务：删除 t_order_task / t_order_task_fail +-- 历史导入任务数据不再保留；Excel 导入统一走 t_order_batch_job（action=import-row） +-- 2026-07-22：触发重新发布，确保 API 不再访问已删表 + +DROP TABLE IF EXISTS `t_order_task_fail`; +DROP TABLE IF EXISTS `t_order_task`; 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..4c10a61 100644 --- a/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs @@ -29,6 +29,16 @@ public class OrderBatchJobOutput public int FailCount { get; set; } + public int PendingCount { get; set; } + + public int RunningCount { get; set; } + + public int CancelledCount { get; set; } + + public int? TemplateId { get; set; } + + public string? TaskName { get; set; } + public DateTime? StartedAt { get; set; } public DateTime? FinishedAt { get; set; } @@ -44,12 +54,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..8d3c032 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; } @@ -32,8 +34,6 @@ public class OrderBatchPayloadJobCreateInput /// public class OrderBatchImportRowPayload { - public int TaskId { get; set; } - public int RowNumber { get; set; } public int CustomerId { get; set; } @@ -48,7 +48,7 @@ public class OrderBatchImportRowPayload /// public class OrderBatchImportJobPayload { - public int TaskId { get; set; } + public string TaskName { get; set; } = string.Empty; public int TemplateId { get; set; } diff --git a/src/Oms.Application/Dtos/Orders/OrderTaskDetailOutput.cs b/src/Oms.Application/Dtos/Orders/OrderTaskDetailOutput.cs deleted file mode 100644 index 49d3b70..0000000 --- a/src/Oms.Application/Dtos/Orders/OrderTaskDetailOutput.cs +++ /dev/null @@ -1,57 +0,0 @@ -namespace Oms.Application.Dtos.Orders; - -/// -/// 订单任务详情出参。 -/// -public class OrderTaskDetailOutput -{ - /// - /// 主键。 - /// - public int Id { get; set; } - - /// - /// 任务名称。 - /// - public string TaskName { get; set; } = string.Empty; - - /// - /// 数量。 - /// - public int? Quantity { get; set; } - - /// - /// 成功数量。 - /// - public int? SuccessCount { get; set; } - - /// - /// 失败数量。 - /// - public int? FailCount { get; set; } - - /// - /// 状态。 - /// - public string Status { get; set; } = string.Empty; - - /// - /// 模板主键。 - /// - public int? TemplateId { get; set; } - - /// - /// 创建时间。 - /// - public DateTime CreateTime { get; set; } - - /// - /// 更新时间。 - /// - public DateTime? UpdateTime { get; set; } - - /// - /// 失败明细集合。 - /// - public List FailList { get; set; } = []; -} diff --git a/src/Oms.Application/Dtos/Orders/OrderTaskFailOutput.cs b/src/Oms.Application/Dtos/Orders/OrderTaskFailOutput.cs deleted file mode 100644 index 4b0b7b8..0000000 --- a/src/Oms.Application/Dtos/Orders/OrderTaskFailOutput.cs +++ /dev/null @@ -1,32 +0,0 @@ -namespace Oms.Application.Dtos.Orders; - -/// -/// 订单任务失败项出参。 -/// -public class OrderTaskFailOutput -{ - /// - /// 主键。 - /// - public int Id { get; set; } - - /// - /// 任务主键。 - /// - public int TaskId { get; set; } - - /// - /// 行号。 - /// - public int RowNumber { get; set; } - - /// - /// 失败原因。 - /// - public string Reason { get; set; } = string.Empty; - - /// - /// 原始数据 JSON。 - /// - public object? RawDataJson { get; set; } -} diff --git a/src/Oms.Application/Dtos/Orders/OrderTaskListInput.cs b/src/Oms.Application/Dtos/Orders/OrderTaskListInput.cs deleted file mode 100644 index 4024b9d..0000000 --- a/src/Oms.Application/Dtos/Orders/OrderTaskListInput.cs +++ /dev/null @@ -1,27 +0,0 @@ -namespace Oms.Application.Dtos.Orders; - -/// -/// 订单任务列表查询入参。 -/// -public class OrderTaskListInput -{ - /// - /// 页码，从 1 开始。 - /// - public int Page { get; set; } = 1; - - /// - /// 每页条数。 - /// - public int PageSize { get; set; } = 20; - - /// - /// 状态。 - /// - public string? Status { get; set; } - - /// - /// 关键词。 - /// - public string? Keyword { get; set; } -} diff --git a/src/Oms.Application/Dtos/Orders/OrderTaskOutput.cs b/src/Oms.Application/Dtos/Orders/OrderTaskOutput.cs deleted file mode 100644 index e101001..0000000 --- a/src/Oms.Application/Dtos/Orders/OrderTaskOutput.cs +++ /dev/null @@ -1,42 +0,0 @@ -namespace Oms.Application.Dtos.Orders; - -/// -/// 订单任务出参。 -/// -public class OrderTaskOutput -{ - /// - /// 主键。 - /// - public int Id { get; set; } - - /// - /// 任务名称。 - /// - public string TaskName { get; set; } = string.Empty; - - /// - /// 数量。 - /// - public int? Quantity { get; set; } - - /// - /// 状态。 - /// - public string Status { get; set; } = string.Empty; - - /// - /// 模板主键。 - /// - public int? TemplateId { get; set; } - - /// - /// 创建时间。 - /// - public DateTime CreateTime { get; set; } - - /// - /// 更新时间。 - /// - public DateTime? UpdateTime { get; set; } -} diff --git a/src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs b/src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs index b736848..d5abf12 100644 --- a/src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs +++ b/src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs @@ -1,22 +1,17 @@ namespace Oms.Application.Dtos.Orders; /// -/// 订单结果出参。 +/// 订单 Excel 导入出参（统一返回批量任务 ID）。 /// public class OrderUploadOutput { /// - /// 任务主键。 + /// 批量任务主键。 /// - public int TaskId { get; set; } + public int JobId { get; set; } /// - /// 关联批量 Job（异步导入时返回）。 - /// - public int? JobId { get; set; } - - /// - /// 是否异步处理中。 + /// 是否异步处理中（RabbitMQ 队列）。 /// public bool Async { 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..d0f0c01 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs @@ -1,6 +1,7 @@ using Fake.Application; using Fake.DependencyInjection; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using Newtonsoft.Json; using Oms.Application.Dtos.Orders; using Oms.Application.Dtos.Shared; @@ -23,6 +24,8 @@ public sealed class OrderBatchJobCoordinator( IOrderRepository orderRepository, ICountryRepository countryRepository, IOrderBatchJobPublisher publisher, + OrderBatchJobProcessor batchJobProcessor, + IOptions batchOptions, ILogger logger) : IScopedDependency { private const int DefaultMaxBatchSize = 200; @@ -113,6 +116,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}", @@ -167,6 +171,83 @@ public sealed class OrderBatchJobCoordinator( throw new BusinessException("订单批量队列未启用，请检查 RabbitMQ 配置"); } + var (job, items, itemsInput, created) = await InsertPayloadJobAsync(input, operatorUserId, cancellationToken); + if (!created) + { + return await MapJobOutputAsync(job, cancellationToken); + } + + for (var index = 0; index < items.Count; index++) + { + await publisher.PublishAsync( + BuildPayloadMessage(job, items[index], itemsInput[index], input.CustomerId, operatorUserId), + cancellationToken); + } + + logger.LogInformation( + "Created payload batch job {JobId} action={Action} total={Total}", + job.Id, + job.Action, + job.Total); + + return await MapJobOutputAsync(job, cancellationToken); + } + + /// + /// 同步创建并逐条处理 Payload 批量任务（MQ 未启用时的 Excel 导入等）。 + /// + public async Task CreatePayloadJobAndProcessSyncAsync( + OrderBatchPayloadJobCreateInput input, + int operatorUserId, + CancellationToken cancellationToken = default) + { + var (job, items, itemsInput, created) = await InsertPayloadJobAsync(input, operatorUserId, cancellationToken); + if (!created) + { + return await MapJobOutputAsync(job, cancellationToken); + } + + var maxAttempts = Math.Max(1, batchOptions.Value.MaxSystemRetries); + + for (var index = 0; index < items.Count; index++) + { + var item = items[index]; + var message = BuildPayloadMessage(job, item, itemsInput[index], input.CustomerId, operatorUserId); + + for (var attempt = 0; attempt < maxAttempts; attempt++) + { + try + { + await batchJobProcessor.ProcessAsync(message, cancellationToken); + break; + } + catch (InvalidOperationException) when (attempt < maxAttempts - 1) + { + var refreshed = await batchJobRepository.GetItemByIdAsync(item.Id, cancellationToken); + if (refreshed?.Status is OrderBatchJobItemStatusCodes.Failed or OrderBatchJobItemStatusCodes.Success) + { + break; + } + } + } + } + + logger.LogInformation( + "Processed payload batch job synchronously {JobId} action={Action} total={Total}", + job.Id, + job.Action, + job.Total); + + var refreshedJob = await batchJobRepository.GetByIdAsync(job.Id, cancellationToken) + ?? throw new BusinessException("批量任务不存在"); + return await MapJobOutputAsync(refreshedJob, cancellationToken); + } + + private async Task<(OrderBatchJob Job, List Items, List ItemsInput, bool Created)> InsertPayloadJobAsync( + OrderBatchPayloadJobCreateInput input, + int operatorUserId, + CancellationToken cancellationToken) + { var action = input.Action?.Trim() ?? string.Empty; if (string.IsNullOrWhiteSpace(action)) { @@ -191,7 +272,8 @@ public sealed class OrderBatchJobCoordinator( var existing = await batchJobRepository.GetByRequestIdAsync(input.RequestId.Trim(), cancellationToken); if (existing != null) { - return await MapJobOutputAsync(existing, cancellationToken); + var existingItems = await batchJobRepository.GetItemsByJobIdAsync(existing.Id, cancellationToken); + return (existing, existingItems, itemsInput, false); } } @@ -232,6 +314,7 @@ public sealed class OrderBatchJobCoordinator( JobId = job.Id, OrderId = source.OrderId, ServiceNo = source.ServiceNo, + ClientNo = source.ClientNo, Action = action, Status = OrderBatchJobItemStatusCodes.Pending, MessageId = itemKey, @@ -244,33 +327,43 @@ public sealed class OrderBatchJobCoordinator( } await batchJobRepository.InsertItemsAsync(items, cancellationToken); + return (job, items, itemsInput, true); + } - for (var index = 0; index < items.Count; index++) + private static OrderBatchJobMessage BuildPayloadMessage( + OrderBatchJob job, + OrderBatchJobItem item, + OrderBatchPayloadItemInput source, + int? customerId, + int operatorUserId) => + new() { - var item = items[index]; - var source = itemsInput[index]; - var message = new OrderBatchJobMessage - { - JobId = job.Id, - JobItemId = item.Id, - Action = action, - OrderId = item.OrderId, - CustomerId = input.CustomerId, - PayloadJson = source.PayloadJson, - OperatorUserId = operatorUserId, - MessageId = item.MessageId ?? $"{job.Id}:{index + 1}:{action}", - }; + JobId = job.Id, + JobItemId = item.Id, + Action = job.Action, + OrderId = item.OrderId, + CustomerId = customerId, + PayloadJson = source.PayloadJson, + OperatorUserId = operatorUserId, + MessageId = item.MessageId ?? $"{job.Id}:{item.Id}:{job.Action}", + }; - await publisher.PublishAsync(message, cancellationToken); + private static void ApplyImportMetadata(OrderBatchJobOutput output, OrderBatchJob job) + { + if (!string.Equals(job.Action, OrderBatchActionCodes.ImportRow, StringComparison.OrdinalIgnoreCase) + || string.IsNullOrWhiteSpace(job.PayloadJson)) + { + return; } - logger.LogInformation( - "Created payload batch job {JobId} action={Action} total={Total}", - job.Id, - action, - job.Total); + var meta = JsonConvert.DeserializeObject(job.PayloadJson); + if (meta == null) + { + return; + } - return await MapJobOutputAsync(job, cancellationToken); + output.TemplateId = meta.TemplateId > 0 ? meta.TemplateId : null; + output.TaskName = string.IsNullOrWhiteSpace(meta.TaskName) ? null : meta.TaskName.Trim(); } public async Task PublishTrackUpsertAsync(SmallBagTrackUpsertRequest request, int orderId, int? customerId) @@ -309,18 +402,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 +496,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 +537,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,14 +552,16 @@ 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); } - return new OrderBatchJobOutput + var output = new OrderBatchJobOutput { JobId = job.Id, Action = job.Action, @@ -467,10 +569,15 @@ 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, }; + ApplyImportMetadata(output, job); + return output; } public async Task> GetJobListAsync( 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..e3b09df 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,20 @@ 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 +56,6 @@ public sealed class OrderBatchJobProcessor( if (message.JobItemId <= 0) { - // fire-and-forget（如轨迹同步）：失败只记日志，不抛、不进 DLX if (!result.Success) { logger.LogWarning( @@ -75,25 +76,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,20 +89,36 @@ public sealed class OrderBatchJobProcessor( if (result.IsBusinessFailure) { - item.Status = OrderBatchJobItemStatusCodes.Failed; - item.FailReason = result.FailReason; - await batchJobRepository.UpdateItemAsync(item, cancellationToken); - if (message.Action == OrderBatchActionCodes.ImportRow) + if (result.CreatedOrderId is > 0) { - await RecordImportTaskFailAsync(message, result.FailReason, cancellationToken); + item.OrderId = result.CreatedOrderId.Value; } + await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken); + OrderBatchJobItemFeedback.ApplyBusinessFailure(item, result, action); + await batchJobRepository.UpdateItemAsync(item, cancellationToken); + await RefreshJobSummaryAsync(item.JobId, cancellationToken); 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 +138,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 @@ -145,80 +158,5 @@ public sealed class OrderBatchJobProcessor( } await batchJobRepository.UpdateJobAsync(job, cancellationToken); - await SyncOrderTaskFromJobAsync(job, cancellationToken); - } - - private async Task SyncOrderTaskFromJobAsync(OrderBatchJob job, CancellationToken cancellationToken) - { - if (job.Action != OrderBatchActionCodes.ImportRow || string.IsNullOrWhiteSpace(job.PayloadJson)) - { - return; - } - - var meta = JsonConvert.DeserializeObject(job.PayloadJson); - if (meta == null || meta.TaskId <= 0) - { - return; - } - - var task = await orderTaskRepository.FirstOrDefaultAsync(x => x.Id == meta.TaskId, cancellationToken); - if (task == null) - { - return; - } - - task.SuccessCount = job.SuccessCount; - task.FailCount = job.FailCount; - task.UpdateTime = DateTime.Now; - - if (job.Status is OrderBatchJobStatusCodes.Completed or OrderBatchJobStatusCodes.Failed) - { - task.Status = OrderTaskStatusCodes.Completed; - } - else if (job.Status == OrderBatchJobStatusCodes.Running) - { - task.Status = OrderTaskStatusCodes.Processing; - } - - await orderTaskRepository.UpdateAsync(task, cancellationToken: cancellationToken); - } - - private async Task RecordImportTaskFailAsync( - OrderBatchJobMessage message, - string? reason, - CancellationToken cancellationToken) - { - if (string.IsNullOrWhiteSpace(message.PayloadJson)) - { - return; - } - - var payload = JsonConvert.DeserializeObject(message.PayloadJson); - if (payload == null || payload.TaskId <= 0) - { - return; - } - - object? rawData; - try - { - rawData = JsonConvert.DeserializeObject(payload.RowDataJson) ?? payload.RowDataJson; - } - catch - { - rawData = payload.RowDataJson; - } - - await orderTaskRepository.InsertFailListAsync([ - new OrderTaskFail - { - TaskId = payload.TaskId, - RowNumber = payload.RowNumber, - Reason = reason ?? "导入失败", - RawDataJson = JsonConvert.SerializeObject(rawData), - CreateUserId = message.OperatorUserId > 0 ? message.OperatorUserId : 0, - CreateTime = DateTime.Now, - } - ], cancellationToken); } } diff --git a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs index 1f4c09c..ad93154 100644 --- a/src/Oms.Application/Services/Orders/OrderService.Helpers.cs +++ b/src/Oms.Application/Services/Orders/OrderService.Helpers.cs @@ -344,21 +344,23 @@ public partial class OrderService throw new BusinessException("上传文件不能为空"); } + var batchInput = await BuildImportBatchJobInputAsync(input); + var operatorUserId = CurrentUser.GetId(); + if (BatchJobPublisher.IsEnabled) { - return await ImportOrdersViaQueueAsync(input); + var job = await BatchJobCoordinator.CreatePayloadJobAsync(batchInput, operatorUserId); + return new OrderUploadOutput { JobId = job.JobId, Async = true }; } - return await ImportOrdersSyncAsync(input); + var syncJob = await BatchJobCoordinator.CreatePayloadJobAndProcessSyncAsync(batchInput, operatorUserId); + return new OrderUploadOutput { JobId = syncJob.JobId, Async = false }; } - private async Task ImportOrdersViaQueueAsync(OrderUploadInput input) + private async Task BuildImportBatchJobInputAsync(OrderUploadInput input) { - var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId); - if (template == null) - { - throw new BusinessException("模板不存在"); - } + var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId) + ?? throw new BusinessException("模板不存在"); var effectiveCustomer = await ResolveEffectiveCustomerAsync(); EnsureTemplateAccessible(template, effectiveCustomer.Id); @@ -378,20 +380,12 @@ public partial class OrderService throw new BusinessException($"单次最多导入 {ParcelBatchMaxCount} 行"); } - var task = await OrderTaskRepository.InsertAsync(new OrderTask - { - TaskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}", - Quantity = parsedRows.Count, - SuccessCount = 0, - FailCount = 0, - Status = OrderTaskStatusCodes.Processing, - TemplateId = template.Id, - CustomerId = effectiveCustomer.Id - }); + var taskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}"; + var requestId = $"import-{effectiveCustomer.Id}-{template.Id}-{DateTime.Now:yyyyMMddHHmmssfff}"; var jobPayload = JsonConvert.SerializeObject(new OrderBatchImportJobPayload { - TaskId = task.Id, + TaskName = taskName, TemplateId = template.Id, CustomerId = effectiveCustomer.Id, }); @@ -403,10 +397,10 @@ public partial class OrderService { OrderId = 0, ServiceNo = parcelInput.ServiceNo, - ItemKey = $"import:{task.Id}:{row.RowNumber}", + ClientNo = parcelInput.ClientNo, + ItemKey = $"import:{requestId}:{row.RowNumber}", PayloadJson = JsonConvert.SerializeObject(new OrderBatchImportRowPayload { - TaskId = task.Id, RowNumber = row.RowNumber, CustomerId = effectiveCustomer.Id, TemplateId = template.Id, @@ -415,171 +409,13 @@ public partial class OrderService }; }).ToList(); - var job = await BatchJobCoordinator.CreatePayloadJobAsync( - new OrderBatchPayloadJobCreateInput - { - Action = OrderBatchActionCodes.ImportRow, - CustomerId = effectiveCustomer.Id, - JobPayloadJson = jobPayload, - RequestId = $"import-task-{task.Id}", - Items = payloadItems, - }, - CurrentUser.GetId()); - - return new OrderUploadOutput + return new OrderBatchPayloadJobCreateInput { - TaskId = task.Id, - JobId = job.JobId, - Async = true, - }; - } - - private async Task ImportOrdersSyncAsync(OrderUploadInput input) - { - var template = await OrderTemplateRepository.FirstOrDefaultAsync(x => x.Id == input.TemplateId); - if (template == null) - { - throw new BusinessException("模板不存在"); - } - - var effectiveCustomer = await ResolveEffectiveCustomerAsync(); - EnsureTemplateAccessible(template, effectiveCustomer.Id); - OrderTemplateImportMapping.EnsureTemplateReadyForImport(template); - - var mapping = OrderTemplateImportMapping.Resolve(template); - await using var stream = input.File!.OpenReadStream(); - var parsedRows = await OrderTemplateExcelReader.ReadDataRowsAsync(stream, mapping); - - var task = await OrderTaskRepository.InsertAsync(new OrderTask - { - TaskName = $"批量下单_{DateTime.Now:yyyyMMddHHmmss}", - Quantity = 0, - SuccessCount = 0, - FailCount = 0, - Status = OrderTaskStatusCodes.Processing, - TemplateId = template.Id, - CustomerId = effectiveCustomer.Id - }); - - var quantity = 0; - var successCount = 0; - var failCount = 0; - var failList = new List(); - - foreach (var row in parsedRows) - { - quantity++; - try - { - var parcelInput = OrderTemplateRowMapper.MapToParcelCreateInput( - row.SourceValues, - mapping, - effectiveCustomer.Id); - await CreateParcelCoreAsync(parcelInput); - successCount++; - } - catch (Exception ex) - { - failCount++; - failList.Add(BuildTaskFail(task.Id, row.RowNumber, ex.Message, row.SourceValues)); - } - } - - if (parsedRows.Count == 0) - { - quantity = 1; - failCount = 1; - failList.Add(BuildTaskFail(task.Id, 1, "Excel 中未读取到可用数据", new Dictionary())); - } - - if (failList.Count > 0) - { - await OrderTaskRepository.InsertFailListAsync(failList); - } - - task.Quantity = quantity; - task.SuccessCount = successCount; - task.FailCount = failCount; - task.Status = OrderTaskStatusCodes.Completed; - await OrderTaskRepository.UpdateAsync(task); - return new OrderUploadOutput { TaskId = task.Id }; - } - - private async Task> GetTaskOutputsAsync(OrderTaskListInput input) - { - var page = OmsDocHelper.NormalizePage(input.Page); - var pageSize = OmsDocHelper.NormalizePageSize(input.PageSize); - var keyword = input.Keyword?.Trim(); - var customerId = await OmsSessionHelper.ResolveListCustomerIdAsync( - CustomerRepository, - CurrentUser, - null); - var hasCustomerFilter = customerId.HasValue; - var customerIdValue = customerId.GetValueOrDefault(); - var predicate = (Expression>)(x => - (string.IsNullOrWhiteSpace(input.Status) || x.Status == input.Status) && - (string.IsNullOrWhiteSpace(keyword) || x.TaskName.Contains(keyword)) && - (!hasCustomerFilter || x.CustomerId == customerIdValue)); - var tasks = await OrderTaskRepository.GetPagedListAsync( - predicate, - page, - pageSize, - new Dictionary { [nameof(OrderTask.UpdateTime)] = false, [nameof(OrderTask.Id)] = false }); - var total = await OrderTaskRepository.CountAsync(predicate); - - return new OmsPageResult - { - Items = tasks.Select(x => new OrderTaskOutput - { - Id = x.Id, - TaskName = x.TaskName, - Quantity = x.Quantity, - Status = x.Status, - TemplateId = x.TemplateId, - CreateTime = x.CreateTime, - UpdateTime = OmsDocHelper.NullIfDefault(x.UpdateTime) - }).ToList(), - Total = total - }; - } - - private async Task GetTaskDetailCoreAsync(int taskId) - { - var task = await OrderTaskRepository.FirstOrDefaultAsync(x => x.Id == taskId); - if (task == null) - { - throw new BusinessException("任务不存在"); - } - - var customerId = await OmsSessionHelper.ResolveListCustomerIdAsync( - CustomerRepository, - CurrentUser, - null); - if (customerId.HasValue && task.CustomerId != customerId.Value) - { - throw new BusinessException("任务不存在"); - } - - var failList = await OrderTaskRepository.GetFailListAsync(task.Id); - return new OrderTaskDetailOutput - { - Id = task.Id, - TaskName = task.TaskName, - Quantity = task.Quantity, - SuccessCount = task.SuccessCount, - FailCount = task.FailCount, - Status = task.Status, - TemplateId = task.TemplateId, - CreateTime = task.CreateTime, - UpdateTime = OmsDocHelper.NullIfDefault(task.UpdateTime), - FailList = failList.Select(x => new OrderTaskFailOutput - { - Id = x.Id, - TaskId = x.TaskId, - RowNumber = x.RowNumber, - Reason = x.Reason, - RawDataJson = OmsDocHelper.DeserializeJsonObject(x.RawDataJson) - }).ToList() + Action = OrderBatchActionCodes.ImportRow, + CustomerId = effectiveCustomer.Id, + JobPayloadJson = jobPayload, + RequestId = requestId, + Items = payloadItems, }; } @@ -782,19 +618,6 @@ public partial class OrderService return forecast?.ChargeableWeight; } - private static OrderTaskFail BuildTaskFail(int taskId, int rowNumber, string reason, object rawData) - { - return new OrderTaskFail - { - TaskId = taskId, - RowNumber = rowNumber, - Reason = reason, - RawDataJson = JsonConvert.SerializeObject(rawData), - CreateUserId = 0, - CreateTime = DateTime.Now - }; - } - private static string? GetStringCell(IDictionary row, string key) => row.TryGetValue(key, out var value) ? value?.ToString() : null; private static int? GetintCell(IDictionary row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null; private static int? GetIntCell(IDictionary row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null; diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 8afe2a8..2d46d7e 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -29,7 +29,6 @@ public partial class OrderService : ApplicationService { private IOrderRepository OrderRepository => LazyServiceProvider.GetRequiredService(); private IOrderTemplateRepository OrderTemplateRepository => LazyServiceProvider.GetRequiredService(); - private IOrderTaskRepository OrderTaskRepository => LazyServiceProvider.GetRequiredService(); private IWarehouseAddressRepository WarehouseAddressRepository => LazyServiceProvider.GetRequiredService(); private ISalesProductRepository SalesProductRepository => LazyServiceProvider.GetRequiredService(); private IDictRepository DictRepository => LazyServiceProvider.GetRequiredService(); @@ -335,38 +334,16 @@ public partial class OrderService : ApplicationService } /// - /// 导入订单文件。 + /// 导入订单文件，创建 import-row 批量任务并返回 jobId。 /// /// 入参：订单导入参数。 - /// 订单导入结果。 + /// 批量任务 ID 及是否异步处理。 [HttpPost("/order/upload")] public async Task UploadAsync([FromForm] OrderUploadInput input) { return await ImportOrdersAsync(input); } - /// - /// 查询订单任务列表。 - /// - /// 入参：订单任务列表查询条件。 - /// 订单任务分页结果，包含列表数据和总记录数。 - [HttpGet("/order/task/list")] - public async Task> GetTaskListAsync([FromQuery] OrderTaskListInput input) - { - return await GetTaskOutputsAsync(input); - } - - /// - /// 查询订单任务详情。 - /// - /// 入参：任务主键。 - /// 订单任务详情。 - [HttpGet("/order/task/detail")] - public async Task GetTaskDetailAsync([FromQuery] int taskId) - { - return await GetTaskDetailCoreAsync(taskId); - } - /// /// 创建订单包裹。 /// @@ -412,6 +389,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/Consts/OrderTaskStatusCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderTaskStatusCodes.cs deleted file mode 100644 index 54ffb85..0000000 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderTaskStatusCodes.cs +++ /dev/null @@ -1,26 +0,0 @@ -namespace Oms.Domain.OrdersAggregate.Consts; - -/// -/// 订单导入/批量任务状态（t_order_task.Status，字符串落库；与字典 order_task_status 对齐）。 -/// -public static class OrderTaskStatusCodes -{ - public const string Pending = "pending"; - public const string Processing = "processing"; - public const string Completed = "completed"; - - public static readonly HashSet All = new(StringComparer.OrdinalIgnoreCase) - { - Pending, - Processing, - Completed - }; - - public static string GetDisplayName(string? code) => code?.Trim().ToLowerInvariant() switch - { - Pending => "待处理", - Processing => "处理中", - Completed => "已完成", - _ => code ?? string.Empty - }; -} 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.Domain/OrdersAggregate/Entities/OrderTask.cs b/src/Oms.Domain/OrdersAggregate/Entities/OrderTask.cs deleted file mode 100644 index 9a255e0..0000000 --- a/src/Oms.Domain/OrdersAggregate/Entities/OrderTask.cs +++ /dev/null @@ -1,22 +0,0 @@ -using Oms.Domain.Shared; -using SqlSugar; - -namespace Oms.Domain.OrdersAggregate.Entities; - -[SugarTable("t_order_task")] -public class OrderTask : OmsAuditedAggregateRoot -{ - public string TaskName { get; set; } = string.Empty; - - public int? Quantity { get; set; } - - public int? SuccessCount { get; set; } - - public int? FailCount { get; set; } - - public string Status { get; set; } = string.Empty; - - public int? TemplateId { get; set; } - - public int? CustomerId { get; set; } -} diff --git a/src/Oms.Domain/OrdersAggregate/Entities/OrderTaskFail.cs b/src/Oms.Domain/OrdersAggregate/Entities/OrderTaskFail.cs deleted file mode 100644 index 362816c..0000000 --- a/src/Oms.Domain/OrdersAggregate/Entities/OrderTaskFail.cs +++ /dev/null @@ -1,17 +0,0 @@ -using Oms.Domain.Shared; -using SqlSugar; - -namespace Oms.Domain.OrdersAggregate.Entities; - -[SugarTable("t_order_task_fail")] -public class OrderTaskFail : OmsAuditedEntity -{ - public int TaskId { get; set; } - - public int RowNumber { get; set; } - - public string Reason { get; set; } = string.Empty; - - [SugarColumn(ColumnDataType = "json", IsNullable = true)] - public string? RawDataJson { get; set; } -} diff --git a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderTaskRepository.cs b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderTaskRepository.cs deleted file mode 100644 index f5731ac..0000000 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderTaskRepository.cs +++ /dev/null @@ -1,15 +0,0 @@ -using Fake.Domain.Repositories; -using Oms.Domain.OrdersAggregate.Entities; - -namespace Oms.Domain.OrdersAggregate.Repositories; - -public interface IOrderTaskRepository : IRepository -{ - Task GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default); - - Task ExistsAsync(int id, CancellationToken cancellationToken = default); - - Task> GetFailListAsync(int taskId, CancellationToken cancellationToken = default); - - Task InsertFailListAsync(IEnumerable failList, CancellationToken cancellationToken = default); -} diff --git a/src/Oms.SqlSugarCore/Orders/OrderTaskRepository.cs b/src/Oms.SqlSugarCore/Orders/OrderTaskRepository.cs deleted file mode 100644 index 784accd..0000000 --- a/src/Oms.SqlSugarCore/Orders/OrderTaskRepository.cs +++ /dev/null @@ -1,40 +0,0 @@ -using Fake.DependencyInjection; -using Oms.Domain.OrdersAggregate.Entities; -using Oms.Domain.OrdersAggregate.Repositories; -using Oms.SqlSugarCore.Repositories; - -namespace Oms.SqlSugarCore.Orders; - -public class OrderTaskRepository : OmsSqlSugarRepository, IOrderTaskRepository, IScopedDependency -{ - public Task GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default) - { - return FirstOrDefaultAsync(x => x.CustomerId == customerId && x.Id == id, cancellationToken); - } - - public Task ExistsAsync(int id, CancellationToken cancellationToken = default) - { - return AnyAsync(x => x.Id == id, cancellationToken); - } - - public async Task> GetFailListAsync(int taskId, CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - return await db.Queryable() - .Where(x => x.TaskId == taskId) - .OrderBy(x => x.RowNumber) - .ToListAsync(cancellationToken); - } - - public async Task InsertFailListAsync(IEnumerable failList, CancellationToken cancellationToken = default) - { - var items = failList.ToArray(); - if (items.Length == 0) - { - return; - } - - var db = await GetDbContextAsync(cancellationToken); - await db.Insertable(items).ExecuteCommandAsync(cancellationToken); - } -} diff --git a/src/Oms.Tests/OrderBatchJobProcessorTests.cs b/src/Oms.Tests/OrderBatchJobProcessorTests.cs index 61586e1..0c1f51a 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -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,10 @@ 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 _orderRepository = new(MockBehavior.Loose); private readonly Mock _actionExecutor = new(MockBehavior.Strict); private readonly OrderBatchJobProcessor _processor; @@ -20,8 +19,9 @@ public class OrderBatchJobProcessorTests { _processor = new OrderBatchJobProcessor( _batchJobRepository.Object, - _orderTaskRepository.Object, + _orderRepository.Object, _actionExecutor.Object, + Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 5 }), NullLogger.Instance); } @@ -40,10 +40,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 +57,33 @@ public class OrderBatchJobProcessorTests 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, + _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] @@ -96,6 +124,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())) + .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 diff --git a/系统结构说明.md b/系统结构说明.md index 0b62646..f3862a0 100644 --- a/系统结构说明.md +++ b/系统结构说明.md @@ -179,10 +179,14 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid` | 文件 | 主要 API | 作用 | |------|----------|------| -| `OrderService.cs` | `/order/*`（主文件） | 订单列表/详情/支付/申报/拦截/导出/模板/上传/任务查询 | +| `OrderService.cs` | `/order/*`（主文件） | 订单列表/详情/支付/申报/拦截/导出/模板/上传 | +| `OrderService.BatchJob.cs` | `/order/batch/*` | 批量任务列表/详情/子项/重试/取消 | | `OrderService.ParcelBilling.cs` | `/order/parcel/*` | 小包创建、批量创建、计重算价、费用快照写入 | | `OrderService.ParcelQuote.cs` | `/order/price-check` | 价格试算：匹配产品与分区报价 | -| `OrderService.Helpers.cs` | — | 订单通用辅助（状态校验、映射等） | +| `OrderService.Helpers.cs` | — | 订单通用辅助（导入、状态校验、映射等） | +| `Batch/OrderBatchJobCoordinator.cs` | — | 批量任务创建、幂等、同步/异步入队 | +| `Batch/OrderBatchJobProcessor.cs` | — | 批量子项消费与结果回写 | +| `Batch/OrderBatchJobItemFeedback.cs` | — | 子项成功/失败/重试反馈封装 | | `ParcelBillingCalculator.cs` | — | 静态计重算价引擎：材积重、分区、卖价费用 | | `ParcelQuoteContext.cs` | — | 报价上下文（国家、邮编、尺寸重量等） | | `ParcelQuoteResult.cs` | — | 单产品报价结果模型 | @@ -288,8 +292,8 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid` | `OrderParcelBatchCreateInput.cs` / `OrderParcelBatchCreateOutput.cs` / `OrderParcelBatchCreateItemOutput.cs` | 批量小包创建 | | `OrderParcelItemInput.cs` / `OrderParcelReceiverInput.cs` / `OrderParcelSenderInput.cs` | 小包货品与地址 | | `OrderTemplateListInput.cs` / `OrderTemplateOutput.cs` | 导入模板 | -| `OrderUploadInput.cs` / `OrderUploadOutput.cs` | Excel 批量上传 | -| `OrderTaskListInput.cs` / `OrderTaskOutput.cs` / `OrderTaskDetailOutput.cs` / `OrderTaskFailOutput.cs` | 异步导入任务 | +| `OrderUploadInput.cs` / `OrderUploadOutput.cs` | Excel 批量上传（返回 jobId、async） | +| `OrderBatchJobListInput.cs` / `OrderBatchJobDtos.cs` | 批量任务列表/详情/子项 DTO | | `ProductBillingSnapshotDtos.cs` | 计费快照（产品、价格、分区等）内部 DTO | #### Dtos/Recharge/ @@ -360,15 +364,16 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid` | `Entities/OrderFee.cs` | 订单费用明细 | | `Entities/OrderWeight.cs` | 重量尺寸（预报/复核等） | | `Entities/OrderTemplate.cs` | 导入模板元数据 | -| `Entities/OrderTask.cs` | 批量导入任务 | -| `Entities/OrderTaskFail.cs` | 导入失败行 | +| `Entities/OrderBatchJob.cs` | 批量任务头（导入/支付/打印等） | +| `Entities/OrderBatchJobItem.cs` | 批量任务子项 | +| `Consts/OrderBatchJobStatusCodes.cs` | 批量任务状态常量 | | `Consts/OrderStatusCodes.cs` | 订单状态常量 | | `Consts/OrderFeeTypeCodes.cs` | 费用类型常量 | | `Consts/OrderWeightTypeCodes.cs` | 重量类型常量 | | `Consts/CargoAttributeCodes.cs` | 货物属性常量 | | `Repositories/IOrderRepository.cs` | 订单仓储（含关联加载） | | `Repositories/IOrderTemplateRepository.cs` | 模板仓储 | -| `Repositories/IOrderTaskRepository.cs` | 导入任务仓储 | +| `Repositories/IOrderBatchJobRepository.cs` | 批量任务仓储 | ### ChannelsAggregate/（渠道/产品/计价） @@ -449,7 +454,7 @@ JWT 中业务用户整数 ID 的 Claim 名为 **`mapId`**（对应 `user.userid` | `Recharge/RechargeRecordRepository.cs` | `IRechargeRecordRepository` | 充值记录 | | `Orders/OrderRepository.cs` | `IOrderRepository` | 订单及关联表 | | `Orders/OrderTemplateRepository.cs` | `IOrderTemplateRepository` | 订单模板 | -| `Orders/OrderTaskRepository.cs` | `IOrderTaskRepository` | 导入任务 | +| `Orders/OrderBatchJobRepository.cs` | `IOrderBatchJobRepository` | 批量任务及子项 | | `Channels/SalesProductRepository.cs` | `ISalesProductRepository` | 销售产品 | | `Channels/ServiceChannelRepository.cs` | `IServiceChannelRepository` | 服务渠道 | | `Channels/ServiceChannelGroupRepository.cs` | `IServiceChannelGroupRepository` | 渠道分组 |