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

refactor: 合并 OrderTask 至批量任务，移除 t_order_task 依赖

Browse Source 
Excel 导入与任务进度统一走 t_order_batch_job；删除旧实体、仓储与 /order/task/* 接口，上传返回 jobId。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zzh 2026-07-22 12:36:57 +08:00
parent 1d4b1c9665
commit bb09a51f4a
21 changed files with 371 additions and 873 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
db.sql
View File

@ -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 等',

9
design.md
View File

@ -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`）
- 保存价格项时同时覆盖因素与价格明细
- 发布与回滚时版本快照回写
- 充值审核通过并更新余额

5
sql/migrate_drop_order_task_merge_batch.sql Normal file
View File

@ -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`;

4
src/Oms.Application/Dtos/Orders/OrderBatchJobDtos.cs
View File

@ -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; }

4
src/Oms.Application/Dtos/Orders/OrderBatchPayloadDtos.cs
View File

@ -34,8 +34,6 @@ public class OrderBatchPayloadJobCreateInput
/// </summary>
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
/// </summary>
public class OrderBatchImportJobPayload
{
public int TaskId { get; set; }
public string TaskName { get; set; } = string.Empty;
public int TemplateId { get; set; }

57
src/Oms.Application/Dtos/Orders/OrderTaskDetailOutput.cs
View File

@ -1,57 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务详情出参。
/// </summary>
public class OrderTaskDetailOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务名称。
/// </summary>
public string TaskName { get; set; } = string.Empty;
/// <summary>
/// 数量。
/// </summary>
public int? Quantity { get; set; }
/// <summary>
/// 成功数量。
/// </summary>
public int? SuccessCount { get; set; }
/// <summary>
/// 失败数量。
/// </summary>
public int? FailCount { get; set; }
/// <summary>
/// 状态。
/// </summary>
public string Status { get; set; } = string.Empty;
/// <summary>
/// 模板主键。
/// </summary>
public int? TemplateId { get; set; }
/// <summary>
/// 创建时间。
/// </summary>
public DateTime CreateTime { get; set; }
/// <summary>
/// 更新时间。
/// </summary>
public DateTime? UpdateTime { get; set; }
/// <summary>
/// 失败明细集合。
/// </summary>
public List<OrderTaskFailOutput> FailList { get; set; } = [];
}

32
src/Oms.Application/Dtos/Orders/OrderTaskFailOutput.cs
View File

@ -1,32 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务失败项出参。
/// </summary>
public class OrderTaskFailOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务主键。
/// </summary>
public int TaskId { get; set; }
/// <summary>
/// 行号。
/// </summary>
public int RowNumber { get; set; }
/// <summary>
/// 失败原因。
/// </summary>
public string Reason { get; set; } = string.Empty;
/// <summary>
/// 原始数据 JSON。
/// </summary>
public object? RawDataJson { get; set; }
}

27
src/Oms.Application/Dtos/Orders/OrderTaskListInput.cs
View File

@ -1,27 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务列表查询入参。
/// </summary>
public class OrderTaskListInput
{
/// <summary>
/// 页码，从 1 开始。
/// </summary>
public int Page { get; set; } = 1;
/// <summary>
/// 每页条数。
/// </summary>
public int PageSize { get; set; } = 20;
/// <summary>
/// 状态。
/// </summary>
public string? Status { get; set; }
/// <summary>
/// 关键词。
/// </summary>
public string? Keyword { get; set; }
}

42
src/Oms.Application/Dtos/Orders/OrderTaskOutput.cs
View File

@ -1,42 +0,0 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单任务出参。
/// </summary>
public class OrderTaskOutput
{
/// <summary>
/// 主键。
/// </summary>
public int Id { get; set; }
/// <summary>
/// 任务名称。
/// </summary>
public string TaskName { get; set; } = string.Empty;
/// <summary>
/// 数量。
/// </summary>
public int? Quantity { get; set; }
/// <summary>
/// 状态。
/// </summary>
public string Status { get; set; } = string.Empty;
/// <summary>
/// 模板主键。
/// </summary>
public int? TemplateId { get; set; }
/// <summary>
/// 创建时间。
/// </summary>
public DateTime CreateTime { get; set; }
/// <summary>
/// 更新时间。
/// </summary>
public DateTime? UpdateTime { get; set; }
}

13
src/Oms.Application/Dtos/Orders/OrderUploadOutput.cs
View File

@ -1,22 +1,17 @@
namespace Oms.Application.Dtos.Orders;
/// <summary>
/// 订单结果出参。
/// 订单 Excel 导入出参（统一返回批量任务 ID）。
/// </summary>
public class OrderUploadOutput
{
/// <summary>
/// 任务主键。
/// 批量任务主键。
/// </summary>
public int TaskId { get; set; }
public int JobId { get; set; }
/// <summary>
/// 关联批量 Job（异步导入时返回）。
/// </summary>
public int? JobId { get; set; }
/// <summary>
/// 是否异步处理中。
/// 是否异步处理中（RabbitMQ 队列）。
/// </summary>
public bool Async { get; set; }
}

139
src/Oms.Application/Services/Orders/Batch/OrderBatchJobCoordinator.cs
View File

@ -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<OrderBatchRabbitMqOptions> batchOptions,
ILogger<OrderBatchJobCoordinator> 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);
}
/// <summary>
/// 同步创建并逐条处理 Payload 批量任务（MQ 未启用时的 Excel 导入等）。
/// </summary>
public async Task<OrderBatchJobOutput> 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<OrderBatchJobItem> Items, List<OrderBatchPayloadItemInput> 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<OrderBatchImportJobPayload>(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<OmsPageResult<OrderBatchJobOutput>> GetJobListAsync(

81
src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs
View File

@ -15,7 +15,6 @@ namespace Oms.Application.Services.Orders.Batch;
/// </summary>
public sealed class OrderBatchJobProcessor(
IOrderBatchJobRepository batchJobRepository,
IOrderTaskRepository orderTaskRepository,
IOrderRepository orderRepository,
IOrderBatchActionExecutor actionExecutor,
IOptions<OrderBatchRabbitMqOptions> 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<OrderBatchImportJobPayload>(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<OrderBatchImportRowPayload>(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);
}
}

218
src/Oms.Application/Services/Orders/OrderService.Helpers.cs
View File

@ -344,21 +344,23 @@ public partial class OrderService
throw new BusinessException("上传文件不能为空");
}
var batchInput = await BuildImportBatchJobInputAsync(input);
var operatorUserId = CurrentUser.GetId<int>();
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<OrderUploadOutput> ImportOrdersViaQueueAsync(OrderUploadInput input)
private async Task<OrderBatchPayloadJobCreateInput> 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<int>());
return new OrderUploadOutput
return new OrderBatchPayloadJobCreateInput
{
TaskId = task.Id,
JobId = job.JobId,
Async = true,
};
}
private async Task<OrderUploadOutput> 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<OrderTaskFail>();
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<string, object>()));
}
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<OmsPageResult<OrderTaskOutput>> 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<Func<OrderTask, bool>>)(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<string, bool> { [nameof(OrderTask.UpdateTime)] = false, [nameof(OrderTask.Id)] = false });
var total = await OrderTaskRepository.CountAsync(predicate);
return new OmsPageResult<OrderTaskOutput>
{
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<OrderTaskDetailOutput> 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<string, object> row, string key) => row.TryGetValue(key, out var value) ? value?.ToString() : null;
private static int? GetintCell(IDictionary<string, object> row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null;
private static int? GetIntCell(IDictionary<string, object> row, string key) => int.TryParse(GetStringCell(row, key), out var value) ? value : null;

23
src/Oms.Application/Services/Orders/OrderService.cs
View File

@ -29,7 +29,6 @@ public partial class OrderService : ApplicationService
{
private IOrderRepository OrderRepository => LazyServiceProvider.GetRequiredService<IOrderRepository>();
private IOrderTemplateRepository OrderTemplateRepository => LazyServiceProvider.GetRequiredService<IOrderTemplateRepository>();
private IOrderTaskRepository OrderTaskRepository => LazyServiceProvider.GetRequiredService<IOrderTaskRepository>();
private IWarehouseAddressRepository WarehouseAddressRepository => LazyServiceProvider.GetRequiredService<IWarehouseAddressRepository>();
private ISalesProductRepository SalesProductRepository => LazyServiceProvider.GetRequiredService<ISalesProductRepository>();
private IDictRepository DictRepository => LazyServiceProvider.GetRequiredService<IDictRepository>();
@ -345,28 +344,6 @@ public partial class OrderService : ApplicationService
return await ImportOrdersAsync(input);
}
/// <summary>
/// 查询订单任务列表。
/// </summary>
/// <param name="input">入参：订单任务列表查询条件。</param>
/// <returns>订单任务分页结果，包含列表数据和总记录数。</returns>
[HttpGet("/order/task/list")]
public async Task<OmsPageResult<OrderTaskOutput>> GetTaskListAsync([FromQuery] OrderTaskListInput input)
{
return await GetTaskOutputsAsync(input);
}
/// <summary>
/// 查询订单任务详情。
/// </summary>
/// <param name="taskId">入参：任务主键。</param>
/// <returns>订单任务详情。</returns>
[HttpGet("/order/task/detail")]
public async Task<OrderTaskDetailOutput> GetTaskDetailAsync([FromQuery] int taskId)
{
return await GetTaskDetailCoreAsync(taskId);
}
/// <summary>
/// 创建订单包裹。
/// </summary>

26
src/Oms.Domain/OrdersAggregate/Consts/OrderTaskStatusCodes.cs
View File

@ -1,26 +0,0 @@
namespace Oms.Domain.OrdersAggregate.Consts;
/// <summary>
/// 订单导入/批量任务状态（t_order_task.Status，字符串落库；与字典 order_task_status 对齐）。
/// </summary>
public static class OrderTaskStatusCodes
{
public const string Pending = "pending";
public const string Processing = "processing";
public const string Completed = "completed";
public static readonly HashSet<string> All = new(StringComparer.OrdinalIgnoreCase)
{
Pending,
Processing,
Completed
};
public static string GetDisplayName(string? code) => code?.Trim().ToLowerInvariant() switch
{
Pending => "待处理",
Processing => "处理中",
Completed => "已完成",
_ => code ?? string.Empty
};
}

22
src/Oms.Domain/OrdersAggregate/Entities/OrderTask.cs
View File

@ -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; }
}

17
src/Oms.Domain/OrdersAggregate/Entities/OrderTaskFail.cs
View File

@ -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; }
}

15
src/Oms.Domain/OrdersAggregate/Repositories/IOrderTaskRepository.cs
View File

@ -1,15 +0,0 @@
using Fake.Domain.Repositories;
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Domain.OrdersAggregate.Repositories;
public interface IOrderTaskRepository : IRepository<OrderTask>
{
Task<OrderTask?> GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default);
Task<bool> ExistsAsync(int id, CancellationToken cancellationToken = default);
Task<List<OrderTaskFail>> GetFailListAsync(int taskId, CancellationToken cancellationToken = default);
Task InsertFailListAsync(IEnumerable<OrderTaskFail> failList, CancellationToken cancellationToken = default);
}

40
src/Oms.SqlSugarCore/Orders/OrderTaskRepository.cs
View File

@ -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<OrderTask>, IOrderTaskRepository, IScopedDependency
{
public Task<OrderTask?> GetByCustomerAndIdAsync(int customerId, int id, CancellationToken cancellationToken = default)
{
return FirstOrDefaultAsync(x => x.CustomerId == customerId && x.Id == id, cancellationToken);
}
public Task<bool> ExistsAsync(int id, CancellationToken cancellationToken = default)
{
return AnyAsync(x => x.Id == id, cancellationToken);
}
public async Task<List<OrderTaskFail>> GetFailListAsync(int taskId, CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
return await db.Queryable<OrderTaskFail>()
.Where(x => x.TaskId == taskId)
.OrderBy(x => x.RowNumber)
.ToListAsync(cancellationToken);
}
public async Task InsertFailListAsync(IEnumerable<OrderTaskFail> failList, CancellationToken cancellationToken = default)
{
var items = failList.ToArray();
if (items.Length == 0)
{
return;
}
var db = await GetDbContextAsync(cancellationToken);
await db.Insertable(items).ExecuteCommandAsync(cancellationToken);
}
}

411
src/Oms.Tests/OrderBatchJobProcessorTests.cs
View File

@ -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<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
private readonly Mock<IOrderTaskRepository> _orderTaskRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchActionExecutor> _actionExecutor = new(MockBehavior.Strict);
private readonly OrderBatchJobProcessor _processor;
public OrderBatchJobProcessorTests()
{
_processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderTaskRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 5 }),
NullLogger<OrderBatchJobProcessor>.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<CancellationToken>()))
.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<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
var act = () => _processor.ProcessAsync(message);
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*下游超时*");
item.RetryCount.Should().Be(1);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
}
[Fact]
public async Task ProcessAsync_SystemFailureExhausted_ShouldMarkFailedWithoutThrow()
{
var processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderTaskRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 1 }),
NullLogger<OrderBatchJobProcessor>.Instance);
var item = CreateItem(id: 14, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
SetupJobRefresh(item.JobId, [item]);
await processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.RetryCount.Should().Be(1);
item.FailReason.Should().Contain("系统重试已达上限");
}
[Fact]
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<CancellationToken>()))
.ReturnsAsync(item);
await _processor.ProcessAsync(message);
_actionExecutor.Verify(
x => x.ExecuteAsync(It.IsAny<OrderBatchJobMessage>(), It.IsAny<CancellationToken>()),
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<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
await _processor.ProcessAsync(message);
_batchJobRepository.Verify(
x => x.GetItemByIdAsync(It.IsAny<int>(), It.IsAny<CancellationToken>()),
Times.Never);
}
[Fact]
public async Task ProcessAsync_Success_ShouldPersistClientNo()
{
var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending);
item.Action = OrderBatchActionCodes.ImportRow;
var message = CreateMessage(item.Id);
message.Action = OrderBatchActionCodes.ImportRow;
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.Ok(
createdOrderId: 9001,
serviceNo: "ID20260722001",
clientNo: "CUST-001"));
SetupJobRefresh(item.JobId, [item]);
await _processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Success);
item.OrderId.Should().Be(9001);
item.ClientNo.Should().Be("CUST-001");
item.ServiceNo.Should().Be("ID20260722001");
item.ResultJson.Should().Contain("CUST-001");
}
private void SetupItemLookup(OrderBatchJobItem item)
{
_batchJobRepository
.Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny<CancellationToken>()))
.ReturnsAsync(item);
_batchJobRepository
.Setup(x => x.UpdateItemAsync(item, It.IsAny<CancellationToken>()))
.Returns(Task.CompletedTask);
}
private void SetupJobRefresh(int jobId, List<OrderBatchJobItem> items)
{
var job = new OrderBatchJob
{
Id = jobId,
Action = OrderBatchActionCodes.Pay,
Status = OrderBatchJobStatusCodes.Running,
Total = items.Count,
};
_batchJobRepository
.Setup(x => x.GetByIdAsync(jobId, It.IsAny<CancellationToken>()))
.ReturnsAsync(job);
_batchJobRepository
.Setup(x => x.GetItemsByJobIdAsync(jobId, It.IsAny<CancellationToken>()))
.ReturnsAsync(items);
_batchJobRepository
.Setup(x => x.UpdateJobAsync(job, It.IsAny<CancellationToken>()))
.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<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
private readonly Mock<IOrderBatchActionExecutor> _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<OrderBatchJobProcessor>.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<CancellationToken>()))
.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<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
var act = () => _processor.ProcessAsync(message);
await act.Should().ThrowAsync<InvalidOperationException>()
.WithMessage("*下游超时*");
item.RetryCount.Should().Be(1);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
}
[Fact]
public async Task ProcessAsync_SystemFailureExhausted_ShouldMarkFailedWithoutThrow()
{
var processor = new OrderBatchJobProcessor(
_batchJobRepository.Object,
_orderRepository.Object,
_actionExecutor.Object,
Options.Create(new OrderBatchRabbitMqOptions { MaxSystemRetries = 1 }),
NullLogger<OrderBatchJobProcessor>.Instance);
var item = CreateItem(id: 14, status: OrderBatchJobItemStatusCodes.Pending);
var message = CreateMessage(item.Id);
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
SetupJobRefresh(item.JobId, [item]);
await processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Failed);
item.RetryCount.Should().Be(1);
item.FailReason.Should().Contain("系统重试已达上限");
}
[Fact]
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<CancellationToken>()))
.ReturnsAsync(item);
await _processor.ProcessAsync(message);
_actionExecutor.Verify(
x => x.ExecuteAsync(It.IsAny<OrderBatchJobMessage>(), It.IsAny<CancellationToken>()),
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<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.SystemFail("下游超时"));
await _processor.ProcessAsync(message);
_batchJobRepository.Verify(
x => x.GetItemByIdAsync(It.IsAny<int>(), It.IsAny<CancellationToken>()),
Times.Never);
}
[Fact]
public async Task ProcessAsync_Success_ShouldPersistClientNo()
{
var item = CreateItem(id: 12, status: OrderBatchJobItemStatusCodes.Pending);
item.Action = OrderBatchActionCodes.ImportRow;
var message = CreateMessage(item.Id);
message.Action = OrderBatchActionCodes.ImportRow;
SetupItemLookup(item);
_actionExecutor
.Setup(x => x.ExecuteAsync(message, It.IsAny<CancellationToken>()))
.ReturnsAsync(OrderBatchActionExecuteResult.Ok(
createdOrderId: 9001,
serviceNo: "ID20260722001",
clientNo: "CUST-001"));
SetupJobRefresh(item.JobId, [item]);
await _processor.ProcessAsync(message);
item.Status.Should().Be(OrderBatchJobItemStatusCodes.Success);
item.OrderId.Should().Be(9001);
item.ClientNo.Should().Be("CUST-001");
item.ServiceNo.Should().Be("ID20260722001");
item.ResultJson.Should().Contain("CUST-001");
}
private void SetupItemLookup(OrderBatchJobItem item)
{
_batchJobRepository
.Setup(x => x.GetItemByIdAsync(item.Id, It.IsAny<CancellationToken>()))
.ReturnsAsync(item);
_batchJobRepository
.Setup(x => x.UpdateItemAsync(item, It.IsAny<CancellationToken>()))
.Returns(Task.CompletedTask);
}
private void SetupJobRefresh(int jobId, List<OrderBatchJobItem> items)
{
var job = new OrderBatchJob
{
Id = jobId,
Action = OrderBatchActionCodes.Pay,
Status = OrderBatchJobStatusCodes.Running,
Total = items.Count,
};
_batchJobRepository
.Setup(x => x.GetByIdAsync(jobId, It.IsAny<CancellationToken>()))
.ReturnsAsync(job);
_batchJobRepository
.Setup(x => x.GetItemsByJobIdAsync(jobId, It.IsAny<CancellationToken>()))
.ReturnsAsync(items);
_batchJobRepository
.Setup(x => x.UpdateJobAsync(job, It.IsAny<CancellationToken>()))
.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",
};
}

21
系统结构说明.md
View File

@ -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` | 渠道分组 |