diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index e3b09df..9b04635 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -25,16 +25,9 @@ public sealed class OrderBatchJobProcessor( CancellationToken cancellationToken = default) { var maxSystemRetries = Math.Max(1, batchOptions.Value.MaxSystemRetries); - OrderBatchJobItem? item = null; - if (message.JobItemId > 0) + var item = await ResolveJobItemAsync(message, cancellationToken); + if (item != null) { - item = await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken); - if (item == null) - { - logger.LogWarning("Job item {JobItemId} not found, ack message {MessageId}", message.JobItemId, message.MessageId); - return; - } - if (item.Status == OrderBatchJobItemStatusCodes.Success) { return; @@ -54,7 +47,7 @@ public sealed class OrderBatchJobProcessor( var result = await actionExecutor.ExecuteAsync(message, cancellationToken); - if (message.JobItemId <= 0) + if (item == null) { if (!result.Success) { @@ -68,9 +61,13 @@ public sealed class OrderBatchJobProcessor( return; } - item = await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken); + item = await batchJobRepository.GetItemByIdAsync(item.Id, cancellationToken); if (item == null) { + logger.LogWarning( + "Job item {JobItemId} not found after execute, ack message {MessageId}", + message.JobItemId, + message.MessageId); return; } @@ -121,6 +118,36 @@ public sealed class OrderBatchJobProcessor( throw new InvalidOperationException(item.FailReason ?? result.FailReason ?? "批量任务系统失败"); } + private async Task ResolveJobItemAsync( + OrderBatchJobMessage message, + CancellationToken cancellationToken) + { + if (message.JobItemId > 0) + { + return await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken); + } + + if (message.JobId > 0 && !string.IsNullOrWhiteSpace(message.MessageId)) + { + var resolved = await batchJobRepository.GetItemByJobAndMessageIdAsync( + (int)message.JobId, + message.MessageId.Trim(), + cancellationToken); + if (resolved != null) + { + logger.LogWarning( + "Resolved job item by messageId jobId={JobId} messageId={MessageId} itemId={JobItemId}", + message.JobId, + message.MessageId, + resolved.Id); + } + + return resolved; + } + + return null; + } + private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken) { var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken); diff --git a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderBatchJobRepository.cs b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderBatchJobRepository.cs index cbb8aac..6b83ba7 100644 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderBatchJobRepository.cs +++ b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderBatchJobRepository.cs @@ -13,6 +13,11 @@ public interface IOrderBatchJobRepository : IRepository Task GetItemByIdAsync(int itemId, CancellationToken cancellationToken = default); + Task GetItemByJobAndMessageIdAsync( + int jobId, + string messageId, + CancellationToken cancellationToken = default); + Task InsertItemsAsync(IEnumerable items, CancellationToken cancellationToken = default); Task UpdateItemAsync(OrderBatchJobItem item, CancellationToken cancellationToken = default); diff --git a/src/Oms.SqlSugarCore/Orders/OrderBatchJobRepository.cs b/src/Oms.SqlSugarCore/Orders/OrderBatchJobRepository.cs index fee46fb..5cd3914 100644 --- a/src/Oms.SqlSugarCore/Orders/OrderBatchJobRepository.cs +++ b/src/Oms.SqlSugarCore/Orders/OrderBatchJobRepository.cs @@ -29,8 +29,29 @@ public class OrderBatchJobRepository : OmsSqlSugarRepository, IOr public async Task GetItemByIdAsync(int itemId, CancellationToken cancellationToken = default) { var db = await GetDbContextAsync(cancellationToken); - return await db.Queryable() - .FirstAsync(it => it.Id == itemId, cancellationToken); + var list = await db.Queryable() + .Where(it => it.Id == itemId) + .Take(1) + .ToListAsync(cancellationToken); + return list.FirstOrDefault(); + } + + public async Task GetItemByJobAndMessageIdAsync( + int jobId, + string messageId, + CancellationToken cancellationToken = default) + { + if (jobId <= 0 || string.IsNullOrWhiteSpace(messageId)) + { + return null; + } + + var db = await GetDbContextAsync(cancellationToken); + var list = await db.Queryable() + .Where(it => it.JobId == jobId && it.MessageId == messageId) + .Take(1) + .ToListAsync(cancellationToken); + return list.FirstOrDefault(); } public async Task InsertItemsAsync(IEnumerable items, CancellationToken cancellationToken = default) @@ -43,6 +64,41 @@ public class OrderBatchJobRepository : OmsSqlSugarRepository, IOr var db = await GetDbContextAsync(cancellationToken); await db.Insertable(list).ExecuteCommandAsync(cancellationToken); + + // 批量 Insert 不会回填自增 Id,需重新加载以便 MQ 消息携带正确的 JobItemId。 + var jobId = list[0].JobId; + var persisted = await db.Queryable() + .Where(x => x.JobId == jobId) + .OrderBy(x => x.Id) + .ToListAsync(cancellationToken); + + var byMessageId = persisted + .Where(x => !string.IsNullOrWhiteSpace(x.MessageId)) + .ToDictionary(x => x.MessageId!, x => x.Id); + + foreach (var item in list) + { + if (string.IsNullOrWhiteSpace(item.MessageId)) + { + continue; + } + + if (byMessageId.TryGetValue(item.MessageId, out var persistedId)) + { + item.Id = persistedId; + } + } + + var unmatched = list.Where(x => x.Id == 0).ToList(); + if (unmatched.Count > 0) + { + var assignedIds = list.Where(x => x.Id > 0).Select(x => x.Id).ToHashSet(); + var spare = persisted.Where(x => !assignedIds.Contains(x.Id)).OrderBy(x => x.Id).ToList(); + for (var i = 0; i < unmatched.Count && i < spare.Count; i++) + { + unmatched[i].Id = spare[i].Id; + } + } } public async Task UpdateItemAsync(OrderBatchJobItem item, CancellationToken cancellationToken = default) diff --git a/src/Oms.Tests/OrderBatchJobProcessorTests.cs b/src/Oms.Tests/OrderBatchJobProcessorTests.cs index 0c1f51a..77d8466 100644 --- a/src/Oms.Tests/OrderBatchJobProcessorTests.cs +++ b/src/Oms.Tests/OrderBatchJobProcessorTests.cs @@ -149,6 +149,36 @@ public class OrderBatchJobProcessorTests item.ResultJson.Should().Contain("CUST-001"); } + [Fact] + public async Task ProcessAsync_ZeroJobItemId_ShouldResolveByMessageIdAndMarkSuccess() + { + var item = CreateItem(id: 16, status: OrderBatchJobItemStatusCodes.Pending); + item.Action = OrderBatchActionCodes.ImportRow; + item.MessageId = "import:req:2"; + var message = new OrderBatchJobMessage + { + JobId = item.JobId, + JobItemId = 0, + Action = OrderBatchActionCodes.ImportRow, + MessageId = item.MessageId, + OperatorUserId = 1, + }; + + _batchJobRepository + .Setup(x => x.GetItemByJobAndMessageIdAsync(item.JobId, item.MessageId, It.IsAny())) + .ReturnsAsync(item); + SetupItemLookup(item); + _actionExecutor + .Setup(x => x.ExecuteAsync(message, It.IsAny())) + .ReturnsAsync(OrderBatchActionExecuteResult.Ok(createdOrderId: 9002, clientNo: "CUST-002")); + SetupJobRefresh(item.JobId, [item]); + + await _processor.ProcessAsync(message); + + item.Status.Should().Be(OrderBatchJobItemStatusCodes.Success); + item.OrderId.Should().Be(9002); + } + private void SetupItemLookup(OrderBatchJobItem item) { _batchJobRepository