diff --git a/db.sql b/db.sql index 098d82b..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 等', 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/sql/migrate_drop_order_task_merge_batch.sql b/sql/migrate_drop_order_task_merge_batch.sql new file mode 100644 index 0000000..4a8f194 --- /dev/null +++ b/sql/migrate_drop_order_task_merge_batch.sql @@ -0,0 +1,5 @@ +-- 合并下单任务表到批量任务：删除 t_order_task / t_order_task_fail +-- 历史导入任务数据不再保留；Excel 导入统一走 t_order_batch_job（action=import-row） + +DROP TABLE IF EXISTS `t_order_task_fail`; +DROP TABLE IF EXISTS `t_order_task`; diff --git a/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs b/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs index 1e275d2..4c10a61 100644 --- a/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs @@ -35,6 +35,10 @@ public class OrderBatchJobOutput 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; } diff --git a/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs b/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs index b1f6ed5..8d3c032 100644 --- a/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs +++ b/src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs @@ -34,8 +34,6 @@ public class OrderBatchPayloadJobCreateInput /// public class OrderBatchImportRowPayload { - public int TaskId { get; set; } - public int RowNumber { get; set; } public int CustomerId { get; set; } @@ -50,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/Services/Orders/Batch/OrderBatchJobCoordinator.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs index cee1d9d..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; @@ -168,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)) { @@ -192,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); } } @@ -246,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) @@ -470,7 +561,7 @@ public sealed class OrderBatchJobCoordinator( await batchJobRepository.UpdateJobAsync(job, cancellationToken); } - return new OrderBatchJobOutput + var output = new OrderBatchJobOutput { JobId = job.Id, Action = job.Action, @@ -485,6 +576,8 @@ public sealed class OrderBatchJobCoordinator( FinishedAt = job.FinishedAt, CreateTime = job.CreateTime, }; + ApplyImportMetadata(output, job); + return output; } public async Task> GetJobListAsync( diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index 691fca8..e3b09df 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -15,7 +15,6 @@ namespace Oms.Application.Services.Orders.Batch; /// public sealed class OrderBatchJobProcessor( IOrderBatchJobRepository batchJobRepository, - IOrderTaskRepository orderTaskRepository, IOrderRepository orderRepository, IOrderBatchActionExecutor actionExecutor, IOptions batchOptions, @@ -99,11 +98,6 @@ public sealed class OrderBatchJobProcessor( OrderBatchJobItemFeedback.ApplyBusinessFailure(item, result, action); await batchJobRepository.UpdateItemAsync(item, cancellationToken); - if (message.Action == OrderBatchActionCodes.ImportRow) - { - await RecordImportTaskFailAsync(message, result.FailReason, cancellationToken); - } - await RefreshJobSummaryAsync(item.JobId, cancellationToken); return; } @@ -164,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 = Newtonsoft.Json.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 or OrderBatchJobStatusCodes.Cancelled) - { - 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 = Newtonsoft.Json.JsonConvert.DeserializeObject(message.PayloadJson); - if (payload == null || payload.TaskId <= 0) - { - return; - } - - object? rawData; - try - { - rawData = Newtonsoft.Json.JsonConvert.DeserializeObject(payload.RowDataJson) ?? payload.RowDataJson; - } - catch - { - rawData = payload.RowDataJson; - } - - await orderTaskRepository.InsertFailListAsync([ - new OrderTaskFail - { - TaskId = payload.TaskId, - RowNumber = payload.RowNumber, - Reason = reason ?? "导入失败", - RawDataJson = Newtonsoft.Json.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 19fc734..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, }); @@ -404,10 +398,9 @@ public partial class OrderService OrderId = 0, ServiceNo = parcelInput.ServiceNo, ClientNo = parcelInput.ClientNo, - ItemKey = $"import:{task.Id}:{row.RowNumber}", + ItemKey = $"import:{requestId}:{row.RowNumber}", PayloadJson = JsonConvert.SerializeObject(new OrderBatchImportRowPayload { - TaskId = task.Id, RowNumber = row.RowNumber, CustomerId = effectiveCustomer.Id, TemplateId = template.Id, @@ -416,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, }; } @@ -783,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 3b5aefa..f153507 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(); @@ -345,28 +344,6 @@ public partial class OrderService : ApplicationService 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); - } - /// /// 创建订单包裹。 /// 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/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 c84b9ab..0c1f51a 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -1,207 +1,204 @@ -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", - }; -} +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 _orderRepository = new(MockBehavior.Loose); + private readonly Mock _actionExecutor = new(MockBehavior.Strict); + private readonly OrderBatchJobProcessor _processor; + + public OrderBatchJobProcessorTests() + { + _processor = new OrderBatchJobProcessor( + _batchJobRepository.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, + _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", + }; +} 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` | 渠道分组 |