From 5fd25d479c10ff06535947505d26c4b13d2ccd87 Mon Sep 17 00:00:00 2001 From: zzh <1368735060@qq.com> Date: Thu, 15 Oct 2026 09:05:26 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8E=BB=E6=8E=89=E5=88=9B=E5=8D=95?= =?UTF-8?q?=E5=90=8E=E7=BD=AE=E6=AD=A5=E9=AA=A4=E8=A1=A8=EF=BC=8C=E8=BD=A8?= =?UTF-8?q?=E8=BF=B9=E8=AE=A2=E5=8D=95=E6=94=B9=E5=9C=A8=E6=8E=92=E8=B4=A7?= =?UTF-8?q?=E6=88=90=E5=8A=9F=E5=90=8E=E5=88=9B=E5=BB=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 回到创单提交后各自投递；并在下单接口注释同步与 MQ 步骤。 Co-authored-by: Cursor --- db.sql | 21 -- sql/drop_order_side_effect.sql | 20 ++ sql/migrate_order_side_effect.sql | 39 --- .../Messaging/OmsAfterCommit.cs | 59 ++-- .../Messaging/OrderSideEffectExecutor.cs | 202 ----------- .../Messaging/OrderSideEffectPayload.cs | 34 -- .../Messaging/OrderSideEffectQueue.cs | 83 ----- src/Oms.Application/OmsApplicationModule.cs | 1 - .../Orders/Batch/OrderBatchJobProcessor.cs | 5 +- .../Issues/AutoTriggerRuntimeService.cs | 94 +----- .../Issues/OrderExceptionWorkflowService.cs | 52 ++- .../Orders/OrderService.ParcelBilling.cs | 195 +++++++++-- .../Orders/OrderService.SideEffects.cs | 317 ------------------ .../Services/Orders/OrderService.cs | 34 +- .../Workers/OrderSideEffectWorker.cs | 41 --- .../Consts/OrderSideEffectCodes.cs | 44 --- .../Entities/OrderSideEffect.cs | 30 -- .../Repositories/IOrderRepository.cs | 19 -- .../IOrderSideEffectRepository.cs | 55 --- .../Orders/OrderRepository.cs | 81 ----- .../Orders/OrderSideEffectRepository.cs | 153 --------- 21 files changed, 268 insertions(+), 1311 deletions(-) create mode 100644 sql/drop_order_side_effect.sql delete mode 100644 sql/migrate_order_side_effect.sql delete mode 100644 src/Oms.Application/Messaging/OrderSideEffectExecutor.cs delete mode 100644 src/Oms.Application/Messaging/OrderSideEffectPayload.cs delete mode 100644 src/Oms.Application/Messaging/OrderSideEffectQueue.cs delete mode 100644 src/Oms.Application/Services/Orders/OrderService.SideEffects.cs delete mode 100644 src/Oms.Application/Workers/OrderSideEffectWorker.cs delete mode 100644 src/Oms.Domain/OrdersAggregate/Consts/OrderSideEffectCodes.cs delete mode 100644 src/Oms.Domain/OrdersAggregate/Entities/OrderSideEffect.cs delete mode 100644 src/Oms.Domain/OrdersAggregate/Repositories/IOrderSideEffectRepository.cs delete mode 100644 src/Oms.SqlSugarCore/Orders/OrderSideEffectRepository.cs diff --git a/db.sql b/db.sql index 199ec6e..3f1111e 100644 --- a/db.sql +++ b/db.sql @@ -101,7 +101,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_side_effect`; DROP TABLE IF EXISTS `t_order_batch_job_item`; DROP TABLE IF EXISTS `t_order_batch_job`; DROP TABLE IF EXISTS `t_order_template`; @@ -307,26 +306,6 @@ CREATE TABLE `t_order_template` ( KEY `idx_ord_tpl_customer_type` (`CustomerId`, `Type`) ) COMMENT='下单模板表'; -CREATE TABLE `t_order_side_effect` ( - `Id` INT NOT NULL AUTO_INCREMENT COMMENT '主键', - `OrderId` INT NOT NULL COMMENT '订单 Id', - `Step` VARCHAR(64) NOT NULL COMMENT '步骤：customs-9610/forecast-billing/track-upsert 等', - `Status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT 'pending/running/done/failed', - `PayloadJson` TEXT NULL COMMENT '步骤入参', - `AttemptCount` INT NOT NULL DEFAULT 0 COMMENT '已尝试次数', - `NextRunTime` DATETIME NULL COMMENT '下次可执行时间', - `LastError` VARCHAR(512) NULL COMMENT '最近一次失败原因', - `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_order_side_effect_due` (`Status`, `NextRunTime`), - KEY `idx_order_side_effect_order` (`OrderId`, `Step`) -) 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/sql/drop_order_side_effect.sql b/sql/drop_order_side_effect.sql new file mode 100644 index 0000000..04869f6 --- /dev/null +++ b/sql/drop_order_side_effect.sql @@ -0,0 +1,20 @@ +-- 删除创单后置步骤表（已不再使用）。可重复执行。 + +SET @db := DATABASE(); + +SET @exists := ( + SELECT COUNT(1) + FROM information_schema.TABLES + WHERE TABLE_SCHEMA = @db + AND TABLE_NAME = 't_order_side_effect' +); + +SET @sql := IF( + @exists > 0, + 'DROP TABLE `t_order_side_effect`', + 'SELECT ''t_order_side_effect already dropped'' AS info' +); + +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/sql/migrate_order_side_effect.sql b/sql/migrate_order_side_effect.sql deleted file mode 100644 index 20fbf7a..0000000 --- a/sql/migrate_order_side_effect.sql +++ /dev/null @@ -1,39 +0,0 @@ --- 创单后置步骤表：与订单同一事务写入，提交后由后台执行。 --- 可重复执行。不建唯一索引。 - -SET @db := DATABASE(); - -SET @exists := ( - SELECT COUNT(1) - FROM information_schema.TABLES - WHERE TABLE_SCHEMA = @db - AND TABLE_NAME = 't_order_side_effect' -); - -SET @sql := IF( - @exists = 0, - 'CREATE TABLE `t_order_side_effect` ( - `Id` INT NOT NULL AUTO_INCREMENT COMMENT ''主键'', - `OrderId` INT NOT NULL COMMENT ''订单 Id'', - `Step` VARCHAR(64) NOT NULL COMMENT ''步骤：customs-9610/forecast-billing/track-upsert 等'', - `Status` VARCHAR(32) NOT NULL DEFAULT ''pending'' COMMENT ''pending/running/done/failed'', - `PayloadJson` TEXT NULL COMMENT ''步骤入参，含轨迹授权头，日志中不要打印全文'', - `AttemptCount` INT NOT NULL DEFAULT 0 COMMENT ''已尝试次数'', - `NextRunTime` DATETIME NULL COMMENT ''下次可执行时间'', - `LastError` VARCHAR(512) NULL COMMENT ''最近一次失败原因'', - `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_order_side_effect_due` (`Status`, `NextRunTime`), - KEY `idx_order_side_effect_order` (`OrderId`, `Step`) - ) COMMENT=''创单后置步骤''', - 'SELECT ''t_order_side_effect already exists'' AS info' -); - -PREPARE stmt FROM @sql; -EXECUTE stmt; -DEALLOCATE PREPARE stmt; diff --git a/src/Oms.Application/Messaging/OmsAfterCommit.cs b/src/Oms.Application/Messaging/OmsAfterCommit.cs index deda445..9b31ee7 100644 --- a/src/Oms.Application/Messaging/OmsAfterCommit.cs +++ b/src/Oms.Application/Messaging/OmsAfterCommit.cs @@ -1,9 +1,11 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Fake.UnitOfWork; namespace Oms.Application.Messaging; /// -/// 当前事务提交并释放后再执行。回滚不会执行。 +/// 把创单/预报之后的副作用延后到当前事务提交成功再执行，每一步单独提交。 /// internal static class OmsAfterCommit { @@ -12,36 +14,13 @@ internal static class OmsAfterCommit var current = unitOfWorkManager.Current; if (current != null && !current.IsCompleted) { - // OnCompleted 仍处在业务事务里，此时写 Outbox 征用不到连接。 - // 等工作单元成功释放后再开独立事务。回滚时 IsCompleted 为 false，不跑。 + // OnCompleted 仍处在创单事务里，此时再写 Outbox 征用不到连接，事件会丢。 + // 等工作单元释放后再开独立事务。 EventHandler handler = null!; handler = (_, _) => { current.Disposed -= handler; - if (!current.IsCompleted) - { - return; - } - - var suppressed = ExecutionContext.SuppressFlow(); - try - { - _ = Task.Run(async () => - { - try - { - await action(); - } - catch - { - // 具体步骤在执行器里记日志。这里避免未观察的任务异常。 - } - }); - } - finally - { - suppressed.Dispose(); - } + _ = Task.Run(action); }; current.Disposed += handler; return Task.CompletedTask; @@ -49,4 +28,30 @@ internal static class OmsAfterCommit return action(); } + + /// + /// 新开独立事务执行一步。失败只记日志，不抛回调用方。 + /// + public static async Task RunStepAsync( + IServiceScopeFactory scopeFactory, + ILogger logger, + string step, + Func work) + { + try + { + using var scope = scopeFactory.CreateScope(); + var unitOfWorkManager = scope.ServiceProvider.GetRequiredService(); + using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); + await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager); + await work(scope.ServiceProvider); + await unitOfWork.CompleteAsync(); + return true; + } + catch (Exception ex) + { + logger.LogError(ex, "订单后置步骤失败 Step={Step}", step); + return false; + } + } } diff --git a/src/Oms.Application/Messaging/OrderSideEffectExecutor.cs b/src/Oms.Application/Messaging/OrderSideEffectExecutor.cs deleted file mode 100644 index bf46bc5..0000000 --- a/src/Oms.Application/Messaging/OrderSideEffectExecutor.cs +++ /dev/null @@ -1,202 +0,0 @@ -using Fake.Application; -using Fake.DependencyInjection; -using Fake.UnitOfWork; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Oms.Application.Services.Orders; -using Oms.Domain.OrdersAggregate.Consts; -using Oms.Domain.OrdersAggregate.Repositories; - -namespace Oms.Application.Messaging; - -/// -/// 每一步后置任务单独开事务执行。失败按次数重试，超过次数记失败。 -/// -public sealed class OrderSideEffectExecutor( - IOrderSideEffectRepository repository, - IUnitOfWorkManager unitOfWorkManager, - OrderService orderService, - ILogger logger) : IScopedDependency -{ - private static readonly TimeSpan StaleRunning = TimeSpan.FromMinutes(2); - - /// - /// 当前事务提交成功后再执行到期步骤。回滚不会触发。 - /// - public static Task ScheduleAfterCommitAsync( - IUnitOfWorkManager unitOfWorkManager, - IServiceScopeFactory scopeFactory, - ILogger logger) - { - return OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => ProcessDueAsync(scopeFactory, logger)); - } - - /// - /// 捞出到期步骤并逐条执行。可重复调用，抢占失败的步骤会被跳过。 - /// - public static async Task ProcessDueAsync( - IServiceScopeFactory scopeFactory, - ILogger logger, - CancellationToken cancellationToken = default) - { - List ids; - try - { - using var scope = scopeFactory.CreateScope(); - var repository = scope.ServiceProvider.GetRequiredService(); - var now = DateTime.Now; - await repository.ResetStaleRunningAsync(now.Subtract(StaleRunning), now, cancellationToken); - ids = await repository.ListDueIdsAsync(30, now, cancellationToken); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - logger.LogError(ex, "捞取创单后置步骤失败"); - return; - } - - foreach (var id in ids) - { - if (cancellationToken.IsCancellationRequested) - { - break; - } - - try - { - using var scope = scopeFactory.CreateScope(); - var executor = scope.ServiceProvider.GetRequiredService(); - await executor.ExecuteAsync(id, cancellationToken); - } - catch (Exception ex) - { - logger.LogError(ex, "订单后置步骤执行失败 SideEffectId={SideEffectId}", id); - } - } - } - - /// - /// 仍停在「预报计费/排货处理中」且没有未完成预报步骤的订单，补一条预报计费步骤。 - /// - public static async Task RecoverStuckForecastAsync( - IServiceScopeFactory scopeFactory, - ILogger logger, - CancellationToken cancellationToken = default) - { - try - { - using var scope = scopeFactory.CreateScope(); - var unitOfWorkManager = scope.ServiceProvider.GetRequiredService(); - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - var repository = scope.ServiceProvider.GetRequiredService(); - var queue = scope.ServiceProvider.GetRequiredService(); - var olderThan = DateTime.Now.AddSeconds(-90); - var orderIds = await repository.ListStuckForecastOrderIdsAsync( - OrderService.ForecastBillingProcessingReason, - olderThan, - 20, - cancellationToken); - foreach (var orderId in orderIds) - { - await queue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.ForecastBilling, - new OrderSideEffectPayload(), - cancellationToken); - } - - await unitOfWork.CompleteAsync(cancellationToken); - } - catch (Exception ex) - { - logger.LogError(ex, "补投递卡住的预报计费失败"); - } - } - - private async Task ExecuteAsync(int id, CancellationToken cancellationToken) - { - var claimed = await ClaimAsync(id, cancellationToken); - if (!claimed) - { - return; - } - - var row = await repository.FirstOrDefaultAsync(x => x.Id == id, cancellationToken); - if (row == null) - { - return; - } - - try - { - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - await OmsIntegrationEventPublisher.EnsureOutboxDbContextAsync(unitOfWorkManager); - await orderService.ExecuteSideEffectAsync(row, cancellationToken); - await repository.MarkDoneAsync(row.Id, DateTime.Now, cancellationToken); - await unitOfWork.CompleteAsync(cancellationToken); - } - catch (Exception ex) - { - logger.LogError(ex, "订单后置步骤失败 Step={Step} OrderId={OrderId}", row.Step, row.OrderId); - var exhausted = await MarkRetryAsync(row, ex, cancellationToken); - if (exhausted - && string.Equals(row.Step, OrderSideEffectSteps.ForecastBilling, StringComparison.Ordinal)) - { - await StampForecastEnqueueFailureAsync(row.OrderId, cancellationToken); - } - } - } - - private async Task ClaimAsync(int id, CancellationToken cancellationToken) - { - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - var claimed = await repository.TryClaimAsync(id, DateTime.Now, cancellationToken); - await unitOfWork.CompleteAsync(cancellationToken); - return claimed; - } - - private async Task MarkRetryAsync( - Domain.OrdersAggregate.Entities.OrderSideEffect row, - Exception ex, - CancellationToken cancellationToken) - { - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - var latest = await repository.FirstOrDefaultAsync(x => x.Id == row.Id, cancellationToken); - var attempts = latest?.AttemptCount ?? row.AttemptCount; - var exhausted = attempts >= OrderSideEffectQueue.MaxAttempts; - var message = ex is BusinessException business ? business.Message : ex.Message; - if (string.IsNullOrWhiteSpace(message)) - { - message = ex.GetType().Name; - } - - if (message.Length > 512) - { - message = message[..512]; - } - - var now = DateTime.Now; - await repository.MarkRetryAsync( - row.Id, - exhausted ? OrderSideEffectStatusCodes.Failed : OrderSideEffectStatusCodes.Pending, - message, - exhausted ? null : now.AddSeconds(15 * Math.Max(1, attempts)), - now, - cancellationToken); - await unitOfWork.CompleteAsync(cancellationToken); - return exhausted; - } - - private async Task StampForecastEnqueueFailureAsync(int orderId, CancellationToken cancellationToken) - { - try - { - using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent()); - await orderService.MarkForecastEnqueueFailedIfStillProcessingAsync(orderId); - await unitOfWork.CompleteAsync(cancellationToken); - } - catch (Exception ex) - { - logger.LogError(ex, "回写预报计费投递失败原因失败 OrderId={OrderId}", orderId); - } - } -} diff --git a/src/Oms.Application/Messaging/OrderSideEffectPayload.cs b/src/Oms.Application/Messaging/OrderSideEffectPayload.cs deleted file mode 100644 index 8f629c5..0000000 --- a/src/Oms.Application/Messaging/OrderSideEffectPayload.cs +++ /dev/null @@ -1,34 +0,0 @@ -namespace Oms.Application.Messaging; - -/// -/// 后置步骤入参。轨迹授权头只在执行时使用，不要写进日志。 -/// -public sealed class OrderSideEffectPayload -{ - public int OperatorUserId { get; set; } - - public string? PlatformCode { get; set; } - - public int? CouponInstanceId { get; set; } - - /// 为 true 时轨迹创建成功后追加 A0101。 - public bool AppendInitialTrack { get; set; } - - public string? Authorization { get; set; } - - public int? CustomerId { get; set; } - - public int? SenderAddressId { get; set; } - - public int? WarehouseId { get; set; } - - public string? PickupDirection { get; set; } - - public string? CarrierCode { get; set; } - - public string? CarrierService { get; set; } - - public DateTime? ExpectPickupTime { get; set; } - - public string? InboundMethod { get; set; } -} diff --git a/src/Oms.Application/Messaging/OrderSideEffectQueue.cs b/src/Oms.Application/Messaging/OrderSideEffectQueue.cs deleted file mode 100644 index 1bd78a4..0000000 --- a/src/Oms.Application/Messaging/OrderSideEffectQueue.cs +++ /dev/null @@ -1,83 +0,0 @@ -using Fake.DependencyInjection; -using Newtonsoft.Json; -using Oms.Domain.OrdersAggregate.Consts; -using Oms.Domain.OrdersAggregate.Entities; -using Oms.Domain.OrdersAggregate.Repositories; - -namespace Oms.Application.Messaging; - -/// -/// 把创单后置步骤写入当前事务。事务回滚时这些行一起消失。 -/// -public sealed class OrderSideEffectQueue(IOrderSideEffectRepository repository) : IScopedDependency -{ - public const int MaxAttempts = 5; - - /// - /// 同一步骤已有 pending、running 或 done 时不再插入。用于 A0101、揽收和创单当下的自动触发。 - /// - public Task EnqueueOnceAsync( - int orderId, - string step, - OrderSideEffectPayload? payload, - CancellationToken cancellationToken = default) => - EnqueueAsync(orderId, step, payload, includeDone: true, cancellationToken); - - /// - /// 没有 pending 或 running 时才插入。已完成的步骤允许再入队，例如重新预报、排货后刷新轨迹。 - /// - public Task EnqueueIfIdleAsync( - int orderId, - string step, - OrderSideEffectPayload? payload, - CancellationToken cancellationToken = default) => - EnqueueAsync(orderId, step, payload, includeDone: false, cancellationToken); - - /// - /// 读取同一步骤最近一次入参。没有记录时返回 null。 - /// - public async Task FindLatestPayloadAsync( - int orderId, - string step, - CancellationToken cancellationToken = default) - { - var json = await repository.FindLatestPayloadJsonAsync(orderId, step, cancellationToken); - if (string.IsNullOrWhiteSpace(json)) - { - return null; - } - - return JsonConvert.DeserializeObject(json); - } - - private async Task EnqueueAsync( - int orderId, - string step, - OrderSideEffectPayload? payload, - bool includeDone, - CancellationToken cancellationToken) - { - if (orderId <= 0 || string.IsNullOrWhiteSpace(step)) - { - return; - } - - if (await repository.ExistsAsync(orderId, step, includeDone, cancellationToken)) - { - return; - } - - var now = DateTime.Now; - await repository.InsertAsync(new OrderSideEffect - { - OrderId = orderId, - Step = step, - Status = OrderSideEffectStatusCodes.Pending, - PayloadJson = payload == null ? null : JsonConvert.SerializeObject(payload), - AttemptCount = 0, - NextRunTime = now, - CreateTime = now, - UpdateTime = now - }, cancellationToken); - } -} diff --git a/src/Oms.Application/OmsApplicationModule.cs b/src/Oms.Application/OmsApplicationModule.cs index 1df839c..c24e699 100644 --- a/src/Oms.Application/OmsApplicationModule.cs +++ b/src/Oms.Application/OmsApplicationModule.cs @@ -65,6 +65,5 @@ public class OmsApplicationModule : FakeModule context.Services.AddAiLlmClient(); context.Services.AddHostedService(); context.Services.AddHostedService(); - context.Services.AddHostedService(); } } diff --git a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs index 8930e13..071ad64 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -182,7 +182,10 @@ public sealed class OrderBatchJobProcessor( text = text[..512]; } - await orderRepository.UpdatePredeclareFailureAsync(order.Id, text, cancellationToken: cancellationToken); + order.OrderStatus = OrderStatusCodes.Draft; + order.PredeclareFailReason = text; + order.UpdateTime = DateTime.Now; + await orderRepository.UpdateAsync(order, cancellationToken); } private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken) diff --git a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs index 69243c9..34d536e 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -17,7 +17,6 @@ public class AutoTriggerRuntimeService( OrderExceptionWorkflowService exceptionWorkflowService, OrderIssueNotificationService issueNotificationService, IOrderSideEffectEventPublisher sideEffectEventPublisher, - IOrderExceptionEventRepository orderExceptionEventRepository, ILogger logger) : IScopedDependency { private const string InterceptOperation = "拦截"; @@ -82,97 +81,6 @@ public class AutoTriggerRuntimeService( await FireAsync(order, timing, operatorUserId); } - /// - /// 预报计费完成后再评估。只跑依赖计费/渠道/费用的规则，以及草稿阶段没能打上的问题件。 - /// - public async Task FireSettledByOrderIdAsync(int orderId, int operatorUserId) - { - var order = await orderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null) - { - return; - } - - var context = await orderAutoTriggerService.BuildPostCommitContextAsync(order); - var triggers = await autoTriggerRepository.GetListAsync(x => x.Status == EnableStatus.Enabled); - var orderedTriggers = triggers - .Where(x => AutoTriggerTimingCodes.Matches(x.TriggerTiming, AutoTriggerTiming.OrderCreate)) - .Select(x => (Trigger: x, Priority: AutoTriggerRuleResolver.GetPriority(x))) - .OrderBy(x => x.Priority) - .ThenBy(x => x.Trigger.Id) - .Select(x => x.Trigger) - .ToList(); - - foreach (var trigger in orderedTriggers) - { - if (string.Equals(trigger.Operation?.Trim(), InterceptOperation, StringComparison.Ordinal)) - { - continue; - } - - var isProblem = string.Equals(trigger.Operation?.Trim(), "问题件", StringComparison.Ordinal); - if (!isProblem && !ReferencesSettlementFields(trigger.TriggerCondition)) - { - continue; - } - - if (isProblem && await AlreadyMarkedByTriggerAsync(order.Id, trigger.Id)) - { - continue; - } - - if (!AutoTriggerConditionMatcher.Matches(context, trigger.TriggerCondition)) - { - continue; - } - - try - { - await ExecuteTriggerAsync(order, trigger, operatorUserId); - } - catch (Exception ex) - { - logger.LogWarning(ex, "计费后自动触发失败 OrderId={OrderId} TriggerId={TriggerId}", order.Id, trigger.Id); - } - } - } - - private static readonly string[] SettlementFieldTokens = - [ - "计费重", - "体积重", - "收货渠道", - "下游服务商", - "渠道预报编码", - "预报失败原因", - "费用类型", - "费用项", - "费用金额", - "费用币种", - "费用下游服务商", - "价格编号", - "计费阶段", - "价格生效时间" - ]; - - private static bool ReferencesSettlementFields(string? condition) - { - if (string.IsNullOrWhiteSpace(condition)) - { - return false; - } - - return SettlementFieldTokens.Any(token => condition.Contains(token, StringComparison.Ordinal)); - } - - private async Task AlreadyMarkedByTriggerAsync(int orderId, int triggerId) - { - var remark = $"自动触发#{triggerId}"; - var existed = await orderExceptionEventRepository.FirstOrDefaultAsync(x => - x.OrderId == orderId && x.Remark == remark); - return existed != null; - } - private async Task ExecuteTriggerAsync( Order order, AutoTrigger trigger, @@ -184,7 +92,7 @@ public class AutoTriggerRuntimeService( if (order.IsIntercept == 0) { order.IsIntercept = 1; - await orderRepository.UpdateInterceptFlagAsync(order.Id, 1); + await orderRepository.UpdateAsync(order); } return; diff --git a/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs b/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs index 430e184..374b18a 100644 --- a/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs +++ b/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs @@ -42,10 +42,9 @@ public class OrderExceptionWorkflowService( int operatorUserId, bool sendNotification) { - var current = await orderRepository.FirstOrDefaultAsync(x => x.Id == order.Id) ?? order; - if (!OrderExceptionStatusRules.CanMarkException(current.OrderStatus)) + if (!OrderExceptionStatusRules.CanMarkException(order.OrderStatus)) { - return OrderExceptionStatusRules.GetForbiddenReason(current.OrderStatus); + return OrderExceptionStatusRules.GetForbiddenReason(order.OrderStatus); } IssueTemplate? template = null; @@ -57,14 +56,14 @@ public class OrderExceptionWorkflowService( IssueHandlerResolution handlerResolution; try { - handlerResolution = await issueHandlerResolver.ResolveAsync(current, template, request.HandlerNames); + handlerResolution = await issueHandlerResolver.ResolveAsync(order, template, request.HandlerNames); } catch (BusinessException ex) { return ex.Message; } - var context = IssuePlaceholderContext.FromOrder(current, request.IssueType, request.Title); + var context = IssuePlaceholderContext.FromOrder(order, request.IssueType, request.Title); var renderedTitle = IssuePlaceholderRenderer.Render(request.Title, context); var renderedContent = IssuePlaceholderRenderer.Render(request.Content, context); @@ -74,37 +73,28 @@ public class OrderExceptionWorkflowService( notifyMethod = IssueNotifyMethodCodes.InApp; } - var fromStatus = current.OrderStatus; + var fromStatus = order.OrderStatus; var isAppend = string.Equals(fromStatus, OrderStatusCodes.Problematic, StringComparison.OrdinalIgnoreCase); if (!isAppend) { - current.PreviousOrderStatus = fromStatus; - current.OrderStatus = OrderStatusCodes.Problematic; + order.PreviousOrderStatus = fromStatus; + order.OrderStatus = OrderStatusCodes.Problematic; } - current.HandleStatus = OrderHandleStatusCodes.Pending; - current.IssueType = request.IssueType; - current.IssueTitle = renderedTitle; - current.IssueContent = renderedContent; - current.IssueTemplateId = request.IssueTemplateId; - current.IssueRemark = request.Remark; - current.IssueTime = DateTime.Now; - current.ClaimType = null; - current.ClaimAmount = null; - current.ClaimCurrency = null; - current.ClaimReason = null; - current.ClaimTime = null; + order.HandleStatus = OrderHandleStatusCodes.Pending; + order.IssueType = request.IssueType; + order.IssueTitle = renderedTitle; + order.IssueContent = renderedContent; + order.IssueTemplateId = request.IssueTemplateId; + order.IssueRemark = request.Remark; + order.IssueTime = DateTime.Now; + order.ClaimType = null; + order.ClaimAmount = null; + order.ClaimCurrency = null; + order.ClaimReason = null; + order.ClaimTime = null; - await orderRepository.UpdateAsync(current); - order.OrderStatus = current.OrderStatus; - order.PreviousOrderStatus = current.PreviousOrderStatus; - order.HandleStatus = current.HandleStatus; - order.IssueType = current.IssueType; - order.IssueTitle = current.IssueTitle; - order.IssueContent = current.IssueContent; - order.IssueTemplateId = current.IssueTemplateId; - order.IssueRemark = current.IssueRemark; - order.IssueTime = current.IssueTime; + await orderRepository.UpdateAsync(order); await orderExceptionEventRepository.InsertAsync(new OrderExceptionEvent { OrderId = order.Id, @@ -130,7 +120,7 @@ public class OrderExceptionWorkflowService( ? handlerResolution.NotifyUserIds : null; await issueNotificationService.DispatchAsync( - current, + order, renderedTitle, renderedContent, notifyMethod, diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index 3270344..f1f1a5f 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -170,12 +170,11 @@ public partial class OrderService var weights = BuildInitialWeightRows(order.Id, input); await SaveParcelChildrenAsync(order, input, weights, country); - // 客户主动「先保存为草稿」：落库后返回。9610 / 揽收与订单同一事务记账，提交后各自执行。 + // 客户主动「先保存为草稿」：落库后返回。9610 / 揽收在事务提交后各自执行。 if (input.SaveAsDraft) { var draftOutput = await BuildParcelCreateOutputAsync(order); - ApplyPickupPlan(input, draftOutput); - await EnqueueCreateSideEffectsAsync(order.Id, input, includeForecast: false, operatorId); + await ScheduleCreateSideEffectsAsync(order.Id, input, draftOutput, includeForecast: false, operatorId); return draftOutput; } @@ -194,8 +193,7 @@ public partial class OrderService await OrderRepository.UpdateAsync(order); var output = await BuildParcelCreateOutputAsync(order); - ApplyPickupPlan(input, output); - await EnqueueCreateSideEffectsAsync(order.Id, input, includeForecast: true, operatorId); + await ScheduleCreateSideEffectsAsync(order.Id, input, output, includeForecast: true, operatorId); return output; } catch (BusinessException) @@ -2286,7 +2284,7 @@ public partial class OrderService } /// - /// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货写入后置步骤，提交后执行。 + /// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货投递 ForecastBilling MQ。 /// internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null) { @@ -2331,7 +2329,38 @@ public partial class OrderService await OrderRepository.UpdateAsync(order); var operatorId = operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); - await EnqueueDraftPredeclareSideEffectsAsync(order.Id, order.CouponInstanceId, operatorId); + var publisher = LazyServiceProvider.GetRequiredService(); + if (!publisher.IsEnabled) + { + var payloadJson = JsonConvert.SerializeObject(new ForecastBillingJobPayload + { + CouponInstanceId = order.CouponInstanceId + }); + await ProcessForecastBillingForBatchAsync(order.Id, payloadJson, operatorId); + } + else + { + var orderId = order.Id; + var couponInstanceId = order.CouponInstanceId; + await ScheduleForecastEnqueueAsync(orderId, couponInstanceId, operatorId); + } + + await ScheduleForecastSideEffectsAsync(order.Id, operatorId); + + // 同步回落失败时给出明确错误；入队成功则保持「处理中」由列表观察结果。 + var refreshed = await OrderRepository.FirstOrDefaultAsync(x => x.Id == order.Id); + if (refreshed != null + && string.Equals(refreshed.OrderStatus, OrderStatusCodes.Draft, StringComparison.OrdinalIgnoreCase) + && !string.IsNullOrWhiteSpace(refreshed.PredeclareFailReason) + && !string.Equals( + refreshed.PredeclareFailReason, + ForecastBillingProcessingReason, + StringComparison.Ordinal)) + { + throw CreateDraftSavedBusinessException( + refreshed, + StripDraftSavedMarker(refreshed.PredeclareFailReason)); + } } catch (BusinessException ex) { @@ -2358,11 +2387,10 @@ public partial class OrderService private async Task MarkParcelDraftFailedAsync(Order order, string reason) { - var text = TrimPendingReason(reason); order.OrderStatus = OrderStatusCodes.Draft; - order.PredeclareFailReason = text; + order.PredeclareFailReason = reason; order.UpdateTime = DateTime.Now; - await OrderRepository.UpdatePredeclareFailureAsync(order.Id, text, order.IsRemote); + await OrderRepository.UpdateAsync(order); } /// @@ -2769,7 +2797,7 @@ public partial class OrderService } // 排货失败不阻断预报成功（与原创单同步路径一致） - await TryConfirmServiceChannelAndPayableAsync( + var allocated = await TryConfirmServiceChannelAndPayableAsync( order, product, billing, @@ -2795,23 +2823,16 @@ public partial class OrderService // ignore } + order.OrderStatus = OrderStatusCodes.Predeclared; order.PredeclareFailReason = string.Empty; order.UpdateTime = DateTime.Now; - await OrderRepository.UpdateForecastCompletionAsync(order); - var persisted = await OrderRepository.FirstOrDefaultAsync(x => x.Id == order.Id); - if (persisted != null) - { - order.OrderStatus = persisted.OrderStatus; - order.PredeclareFailReason = persisted.PredeclareFailReason; - order.PreviousOrderStatus = persisted.PreviousOrderStatus; - order.LabelTemplateId = persisted.LabelTemplateId; - } - else - { - order.OrderStatus = OrderStatusCodes.Predeclared; - } + await OrderRepository.UpdateAsync(order); - await EnqueueSettlementSideEffectsAsync(order.Id, operatorUserId); + // 轨迹订单依赖排货后的渠道/ApiCode，只有排货成功才创建。 + if (allocated && !string.IsNullOrWhiteSpace(order.ChannelCode)) + { + await ScheduleTrackAfterAllocationAsync(order.Id); + } var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true); await OperationLogWriter.TryWriteAsync( @@ -2831,6 +2852,103 @@ public partial class OrderService public int? CouponInstanceId { get; set; } } + /// + /// 创单事务提交后，9610、预报计费、揽收、面单、自动触发各自提交。 + /// 轨迹订单等排货成功后再创建。 + /// + private async Task ScheduleCreateSideEffectsAsync( + int orderId, + OrderParcelCreateInput input, + OrderParcelCreateOutput output, + bool includeForecast, + int operatorUserId) + { + var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); + var scopeFactory = LazyServiceProvider.GetRequiredService(); + var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.OrderParcelCreate"); + var platformCode = input.PlatformCode; + var couponInstanceId = input.CouponInstanceId; + + await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () => + { + try + { + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "customs-9610", sp => + sp.GetRequiredService().PublishCustomsForOrderAsync(orderId, platformCode)); + + var forecastQueued = true; + if (includeForecast) + { + forecastQueued = await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp => + { + var service = sp.GetRequiredService(); + await service.EnqueueForecastBillingByIdAsync(orderId, couponInstanceId, operatorUserId); + }); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => + sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => + sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)); + } + + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp => + sp.GetRequiredService().TryAttachPickupAfterParcelCreateAsync(input, output)); + + if (includeForecast && !forecastQueued) + { + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp => + sp.GetRequiredService().MarkForecastEnqueueFailedIfStillProcessingAsync(orderId)); + } + } + catch (Exception ex) + { + logger.LogError(ex, "创单后置步骤调度失败 OrderId={OrderId}", orderId); + } + }); + } + + private async Task ScheduleForecastEnqueueAsync(int orderId, int? couponInstanceId, int operatorUserId) + { + var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); + var scopeFactory = LazyServiceProvider.GetRequiredService(); + var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); + await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => + OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", sp => + sp.GetRequiredService().EnqueueForecastBillingByIdAsync( + orderId, + couponInstanceId, + operatorUserId))); + } + + /// + /// 草稿重新预报提交后，面单和下单自动触发各自入队。轨迹等排货成功后再创建。 + /// + private async Task ScheduleForecastSideEffectsAsync(int orderId, int operatorUserId) + { + var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); + var scopeFactory = LazyServiceProvider.GetRequiredService(); + var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); + await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () => + { + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp => + sp.GetRequiredService().PublishForecastLabelAsync(orderId, operatorUserId)); + await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp => + sp.GetRequiredService().PublishForecastAutoTriggerAsync(orderId, operatorUserId)); + }); + } + + /// + /// 预报计费事务提交后，再创建轨迹订单（此时已有渠道和 ApiCode）。 + /// + private async Task ScheduleTrackAfterAllocationAsync(int orderId) + { + var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); + var scopeFactory = LazyServiceProvider.GetRequiredService(); + var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.ForecastBilling"); + await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => + OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp => + sp.GetRequiredService().PublishForecastTrackAsync(orderId))); + } + private async Task PublishCustomsForOrderAsync(int orderId, string? platformCode) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); @@ -2843,7 +2961,19 @@ public partial class OrderService await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, platformCode); } - internal async Task MarkForecastEnqueueFailedIfStillProcessingAsync(int orderId) + private async Task EnqueueForecastBillingByIdAsync(int orderId, int? couponInstanceId, int operatorUserId) + { + var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); + if (order == null + || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) + { + return; + } + + await EnqueueForecastBillingAsync(order, couponInstanceId, operatorUserId); + } + + private async Task MarkForecastEnqueueFailedIfStillProcessingAsync(int orderId) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); if (order == null @@ -2855,11 +2985,12 @@ public partial class OrderService await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason); } - private async Task PublishForecastTrackAsync(int orderId, bool appendInitialTrack, string? authorization) + private async Task PublishForecastTrackAsync(int orderId) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) + || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase) + || string.IsNullOrWhiteSpace(order.ChannelCode)) { return; } @@ -2868,12 +2999,8 @@ public partial class OrderService ? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) : null; var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiverInput: null); - request.AppendInitialTrackOnSuccess = appendInitialTrack; - if (!string.IsNullOrWhiteSpace(authorization)) - { - request.Authorization = authorization.Trim(); - } - + request.AppendInitialTrackOnSuccess = true; + ApplyTrackAuthorization(request); await ParcelCreateEventPublisher.PublishTrackUpsertAsync(order, request, customer?.Id); } diff --git a/src/Oms.Application/Services/Orders/OrderService.SideEffects.cs b/src/Oms.Application/Services/Orders/OrderService.SideEffects.cs deleted file mode 100644 index 204230f..0000000 --- a/src/Oms.Application/Services/Orders/OrderService.SideEffects.cs +++ /dev/null @@ -1,317 +0,0 @@ -using Fake.Application; -using Fake.UnitOfWork; -using Microsoft.AspNetCore.Http; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; -using Newtonsoft.Json; -using Oms.Application.Dtos.Orders; -using Oms.Application.Messaging; -using Oms.Domain.OrdersAggregate.Consts; -using Oms.Domain.OrdersAggregate.Entities; - -namespace Oms.Application.Services.Orders; - -public partial class OrderService -{ - private OrderSideEffectQueue SideEffectQueue => - LazyServiceProvider.GetRequiredService(); - - /// - /// 创单事务内写入后置步骤。保存草稿只写 9610 和揽收；确认下单再写预报、轨迹、面单和自动触发。 - /// - private async Task EnqueueCreateSideEffectsAsync( - int orderId, - OrderParcelCreateInput input, - bool includeForecast, - int operatorUserId) - { - var authorization = CaptureRequestAuthorization(); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.Customs9610, - new OrderSideEffectPayload { PlatformCode = input.PlatformCode }); - - if (includeForecast) - { - await SideEffectQueue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.ForecastBilling, - new OrderSideEffectPayload - { - CouponInstanceId = input.CouponInstanceId, - OperatorUserId = operatorUserId - }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.TrackUpsert, - new OrderSideEffectPayload - { - AppendInitialTrack = true, - Authorization = authorization, - OperatorUserId = operatorUserId - }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.LabelGenerate, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.AutoTrigger, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - } - - if (IsDoorPickup(input) && ValidateDoorPickup(input) == null) - { - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.PickupAttach, - BuildPickupPayload(input)); - } - - await KickSideEffectsAsync(); - } - - /// - /// 草稿重新预报：预报计费可再次入队；轨迹 A0101、面单和创单触发已做过则跳过。 - /// - private async Task EnqueueDraftPredeclareSideEffectsAsync( - int orderId, - int? couponInstanceId, - int operatorUserId) - { - var authorization = CaptureRequestAuthorization(); - await SideEffectQueue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.ForecastBilling, - new OrderSideEffectPayload - { - CouponInstanceId = couponInstanceId, - OperatorUserId = operatorUserId - }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.TrackUpsert, - new OrderSideEffectPayload - { - AppendInitialTrack = true, - Authorization = authorization, - OperatorUserId = operatorUserId - }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.LabelGenerate, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - await SideEffectQueue.EnqueueOnceAsync( - orderId, - OrderSideEffectSteps.AutoTrigger, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - await KickSideEffectsAsync(); - } - - /// - /// 预报计费成功后补轨迹渠道、系统面单，以及依赖计费结果的自动触发。 - /// - private async Task EnqueueSettlementSideEffectsAsync(int orderId, int operatorUserId) - { - var previousTrack = await SideEffectQueue.FindLatestPayloadAsync(orderId, OrderSideEffectSteps.TrackUpsert); - await SideEffectQueue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.TrackRefresh, - new OrderSideEffectPayload - { - AppendInitialTrack = false, - Authorization = previousTrack?.Authorization, - OperatorUserId = operatorUserId - }); - await SideEffectQueue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.LabelRefresh, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - await SideEffectQueue.EnqueueIfIdleAsync( - orderId, - OrderSideEffectSteps.AutoTriggerSettled, - new OrderSideEffectPayload { OperatorUserId = operatorUserId }); - await KickSideEffectsAsync(); - } - - private Task KickSideEffectsAsync() - { - var unitOfWorkManager = LazyServiceProvider.GetRequiredService(); - var scopeFactory = LazyServiceProvider.GetRequiredService(); - var logger = LazyServiceProvider.GetRequiredService().CreateLogger("Oms.OrderSideEffect"); - return OrderSideEffectExecutor.ScheduleAfterCommitAsync(unitOfWorkManager, scopeFactory, logger); - } - - /// - /// 执行一条已抢占的后置步骤。由独立事务调用，失败抛出以便重试。 - /// - internal async Task ExecuteSideEffectAsync(OrderSideEffect row, CancellationToken cancellationToken = default) - { - var payload = string.IsNullOrWhiteSpace(row.PayloadJson) - ? new OrderSideEffectPayload() - : JsonConvert.DeserializeObject(row.PayloadJson) ?? new OrderSideEffectPayload(); - - switch (row.Step) - { - case OrderSideEffectSteps.Customs9610: - await PublishCustomsForOrderAsync(row.OrderId, payload.PlatformCode); - break; - case OrderSideEffectSteps.ForecastBilling: - await ExecuteForecastBillingStepAsync(row.OrderId, payload); - break; - case OrderSideEffectSteps.TrackUpsert: - await PublishForecastTrackAsync(row.OrderId, appendInitialTrack: true, payload.Authorization); - break; - case OrderSideEffectSteps.TrackRefresh: - await PublishForecastTrackAsync(row.OrderId, appendInitialTrack: false, payload.Authorization); - break; - case OrderSideEffectSteps.LabelGenerate: - case OrderSideEffectSteps.LabelRefresh: - await PublishForecastLabelAsync(row.OrderId, payload.OperatorUserId); - break; - case OrderSideEffectSteps.AutoTrigger: - await PublishForecastAutoTriggerAsync(row.OrderId, payload.OperatorUserId); - break; - case OrderSideEffectSteps.AutoTriggerSettled: - await PublishSettledAutoTriggerAsync(row.OrderId, payload.OperatorUserId); - break; - case OrderSideEffectSteps.PickupAttach: - await ExecutePickupSideEffectAsync(row.OrderId, payload); - break; - default: - LazyServiceProvider.GetRequiredService() - .CreateLogger("Oms.OrderSideEffect") - .LogWarning("未知的订单后置步骤 Step={Step} OrderId={OrderId}", row.Step, row.OrderId); - break; - } - } - - private async Task ExecuteForecastBillingStepAsync(int orderId, OrderSideEffectPayload payload) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) - { - return; - } - - if (!string.Equals(order.PredeclareFailReason, ForecastBillingProcessingReason, StringComparison.Ordinal)) - { - return; - } - - await EnqueueForecastBillingAsync(order, payload.CouponInstanceId, payload.OperatorUserId); - } - - private async Task PublishSettledAutoTriggerAsync(int orderId, int operatorUserId) - { - var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); - if (order == null - || string.Equals(order.OrderStatus, OrderStatusCodes.Deleted, StringComparison.OrdinalIgnoreCase)) - { - return; - } - - await AutoTriggerRuntimeService.FireSettledByOrderIdAsync(orderId, operatorUserId); - } - - private async Task ExecutePickupSideEffectAsync(int orderId, OrderSideEffectPayload payload) - { - var input = new OrderParcelCreateInput - { - CustomerId = payload.CustomerId, - InboundMethod = payload.InboundMethod, - SenderAddressId = payload.SenderAddressId, - WarehouseId = payload.WarehouseId, - PickupDirection = payload.PickupDirection, - CarrierCode = payload.CarrierCode, - CarrierService = payload.CarrierService, - ExpectPickupTime = payload.ExpectPickupTime - }; - var output = new OrderParcelCreateOutput { OrderId = orderId }; - await TryAttachPickupAfterParcelCreateAsync(input, output); - if (string.IsNullOrWhiteSpace(output.PickupCreateError)) - { - return; - } - - if (output.PickupCreateError.Contains("已绑定", StringComparison.Ordinal) - || output.PickupCreateError.Contains("已在当前揽收单", StringComparison.Ordinal)) - { - return; - } - - throw new BusinessException(output.PickupCreateError); - } - - private void ApplyPickupPlan(OrderParcelCreateInput input, OrderParcelCreateOutput output) - { - if (!IsDoorPickup(input)) - { - return; - } - - var error = ValidateDoorPickup(input); - if (error != null) - { - output.PickupCreateError = error; - return; - } - - output.PickupQueued = true; - } - - private static bool IsDoorPickup(OrderParcelCreateInput input) => - string.Equals(input.InboundMethod?.Trim(), OrderInboundMethodCodes.DoorPickup, StringComparison.OrdinalIgnoreCase); - - private static string? ValidateDoorPickup(OrderParcelCreateInput input) - { - if (input.SenderAddressId is not > 0) - { - return "上门揽收需选择揽收地址"; - } - - if (input.ExpectPickupTime is null || input.ExpectPickupTime == default) - { - return "上门揽收需填写期望揽收时间"; - } - - if (input.WarehouseId is not > 0) - { - return "上门揽收需选择仓库"; - } - - return null; - } - - private static OrderSideEffectPayload BuildPickupPayload(OrderParcelCreateInput input) => - new() - { - CustomerId = input.CustomerId, - SenderAddressId = input.SenderAddressId, - WarehouseId = input.WarehouseId, - PickupDirection = input.PickupDirection, - CarrierCode = input.CarrierCode, - CarrierService = input.CarrierService, - ExpectPickupTime = input.ExpectPickupTime, - InboundMethod = input.InboundMethod - }; - - private string? CaptureRequestAuthorization() - { - try - { - return LazyServiceProvider - .GetRequiredService() - .HttpContext? - .Request - .Headers - .Authorization - .ToString(); - } - catch (ObjectDisposedException) - { - return null; - } - } -} diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index ff7c50b..2e5fe6d 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -262,10 +262,10 @@ public partial class OrderService : ApplicationService } /// - /// 执行订单预报。校验通过后订单保持草稿并标记处理中，计费和排货在事务提交后执行。 + /// 执行订单预报。 /// /// 入参：订单操作参数。 - /// 无返回体。预报结果随后写回订单状态和失败原因。 + /// 无返回体，仅完成订单预报处理。 [HttpPost("/order/pre-declare")] [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task PreDeclareAsync(OrderActionInput input) @@ -603,10 +603,34 @@ public partial class OrderService : ApplicationService } /// - /// 创建订单包裹。 + /// 创建订单包裹（PC/H5 确认下单或保存草稿）。 + /// + /// 【同步，本接口事务内】RequestId 幂等 → 校验入参/销售产品 → 黑名单 → + /// 自动触发「拦截」规则（命中直接失败）→ 生成运单号 → 落库订单与子表。 + /// 确认下单：同步写偏远标记，状态草稿 +「预报计费/排货处理中」后返回； + /// 保存草稿：仅落库，不做预报计费。失败会回滚订单，不落草稿。 + /// + /// + /// 【创单事务提交后各自投递 MQ（互不等待；失败不回滚订单）】 + /// 1) 9610 订单 Upsert（customs-9610）— 草稿与确认下单都会投； + /// 2) 预报计费/排货任务（forecast-billing 批量任务）— 仅确认下单； + /// 队列关闭时在后台同步跑同一套计费逻辑； + /// 3) 系统面单 PDF（label-generate）— 仅确认下单； + /// 4) 下单自动触发（OrderCreate：警告/问题件等，不含拦截）— 仅确认下单； + /// 5) 上门揽收挂载（pickup-attach）— 仅上门揽收且地址/时间/仓库齐全时； + /// 投递失败则同步回落建揽收单。 + /// + /// + /// 【预报计费消费后】计费重、应收、排货应付、营销快照 → 订单改已预报。 + /// 排货成功且已有 ChannelCode 时，再投递轨迹订单 + A0101（track-upsert）； + /// 排货失败不阻断预报，也不建轨迹（可页面补做）。 + /// /// - /// 入参：订单包裹创建参数。 - /// 订单包裹创建结果。 + /// 入参：订单包裹创建参数。SaveAsDraft=true 为仅保存草稿。 + /// + /// 订单包裹创建结果。确认下单时 ForecastBillingQueued 表示预报计费已排队； + /// 上门揽收时 PickupQueued / PickupNo 视投递或同步回落结果而定。 + /// [HttpPost("/order/parcel/create")] [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateParcelAsync(OrderParcelCreateInput input) diff --git a/src/Oms.Application/Workers/OrderSideEffectWorker.cs b/src/Oms.Application/Workers/OrderSideEffectWorker.cs deleted file mode 100644 index 9f28943..0000000 --- a/src/Oms.Application/Workers/OrderSideEffectWorker.cs +++ /dev/null @@ -1,41 +0,0 @@ -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Oms.Application.Messaging; - -namespace Oms.Application.Workers; - -/// -/// 轮询已提交的创单后置步骤。进程在事务提交后、步骤执行前退出时，靠这里补跑。 -/// -public sealed class OrderSideEffectWorker( - IServiceScopeFactory scopeFactory, - ILogger logger) : BackgroundService -{ - private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(2); - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - while (!stoppingToken.IsCancellationRequested) - { - try - { - await OrderSideEffectExecutor.RecoverStuckForecastAsync(scopeFactory, logger, stoppingToken); - await OrderSideEffectExecutor.ProcessDueAsync(scopeFactory, logger, stoppingToken); - } - catch (Exception ex) when (ex is not OperationCanceledException) - { - logger.LogError(ex, "创单后置步骤轮询失败"); - } - - try - { - await Task.Delay(PollInterval, stoppingToken); - } - catch (OperationCanceledException) - { - break; - } - } - } -} diff --git a/src/Oms.Domain/OrdersAggregate/Consts/OrderSideEffectCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderSideEffectCodes.cs deleted file mode 100644 index 9fc3d01..0000000 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderSideEffectCodes.cs +++ /dev/null @@ -1,44 +0,0 @@ -namespace Oms.Domain.OrdersAggregate.Consts; - -/// -/// 创单后置步骤状态。同一订单、同一步骤不建唯一索引，是否重复入队由业务判断。 -/// -public static class OrderSideEffectStatusCodes -{ - public const string Pending = "pending"; - - public const string Running = "running"; - - public const string Done = "done"; - - public const string Failed = "failed"; -} - -/// -/// 创单/预报提交后要单独执行的步骤。 -/// -public static class OrderSideEffectSteps -{ - public const string Customs9610 = "customs-9610"; - - public const string ForecastBilling = "forecast-billing"; - - /// 创建轨迹订单并追加 A0101，同一订单只做一次。 - public const string TrackUpsert = "track-upsert"; - - /// 排货完成后按最新渠道再同步一次轨迹，不追加 A0101。 - public const string TrackRefresh = "track-refresh"; - - public const string LabelGenerate = "label-generate"; - - /// 排货完成后重做系统面单，覆盖创单时还没有渠道的那张。 - public const string LabelRefresh = "label-refresh"; - - /// 下单时机的自动触发。此时还没有计费重、应收和渠道。 - public const string AutoTrigger = "auto-trigger"; - - /// 预报计费成功后再跑依赖计费结果的规则，以及草稿阶段没能打上的问题件。 - public const string AutoTriggerSettled = "auto-trigger-settled"; - - public const string PickupAttach = "pickup-attach"; -} diff --git a/src/Oms.Domain/OrdersAggregate/Entities/OrderSideEffect.cs b/src/Oms.Domain/OrdersAggregate/Entities/OrderSideEffect.cs deleted file mode 100644 index feb5471..0000000 --- a/src/Oms.Domain/OrdersAggregate/Entities/OrderSideEffect.cs +++ /dev/null @@ -1,30 +0,0 @@ -using Oms.Domain.Shared; -using SqlSugar; - -namespace Oms.Domain.OrdersAggregate.Entities; - -/// -/// 与订单同一事务写入的后置步骤。事务回滚则步骤一起消失；提交后由后台执行，进程重启仍可继续。 -/// -[SugarTable("t_order_side_effect")] -public class OrderSideEffect : OmsAuditedAggregateRoot -{ - public int OrderId { get; set; } - - [SugarColumn(Length = 64)] - public string Step { get; set; } = string.Empty; - - [SugarColumn(Length = 32)] - public string Status { get; set; } = string.Empty; - - [SugarColumn(ColumnDataType = "text", IsNullable = true)] - public string? PayloadJson { get; set; } - - public int AttemptCount { get; set; } - - [SugarColumn(IsNullable = true)] - public DateTime? NextRunTime { get; set; } - - [SugarColumn(Length = 512, IsNullable = true)] - public string? LastError { get; set; } -} diff --git a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs index cf35dfd..c5d8ab0 100644 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs +++ b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs @@ -137,23 +137,4 @@ public interface IOrderRepository : IRepository /// 只更新标签模板 Id，避免面单生成覆盖预报计费正在写入的订单字段。 /// Task UpdateLabelTemplateIdAsync(int orderId, int labelTemplateId, CancellationToken cancellationToken = default); - - /// - /// 预报失败时只写状态和原因，不覆盖渠道、面单模板、问题件字段。 - /// - Task UpdatePredeclareFailureAsync( - int orderId, - string reason, - byte? isRemote = null, - CancellationToken cancellationToken = default); - - /// - /// 预报计费成功时只写计费/排货字段。订单已是问题件或理赔件时不改状态，只把异常前状态改成已预报。 - /// - Task UpdateForecastCompletionAsync(Order order, CancellationToken cancellationToken = default); - - /// - /// 只更新拦截标记。 - /// - Task UpdateInterceptFlagAsync(int orderId, byte isIntercept, CancellationToken cancellationToken = default); } diff --git a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderSideEffectRepository.cs b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderSideEffectRepository.cs deleted file mode 100644 index 7d7e724..0000000 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderSideEffectRepository.cs +++ /dev/null @@ -1,55 +0,0 @@ -using Fake.Domain.Repositories; -using Oms.Domain.OrdersAggregate.Entities; - -namespace Oms.Domain.OrdersAggregate.Repositories; - -public interface IOrderSideEffectRepository : IRepository -{ - /// - /// 是否已有未结束（以及可选的已完成）步骤。用于避免重复入队。 - /// - Task ExistsAsync( - int orderId, - string step, - bool includeDone, - CancellationToken cancellationToken = default); - - Task> ListDueIdsAsync(int take, DateTime now, CancellationToken cancellationToken = default); - - /// - /// 取同一步骤最近一条入参，用于把创单时捕获的轨迹授权头带到排货后的刷新。 - /// - Task FindLatestPayloadJsonAsync( - int orderId, - string step, - CancellationToken cancellationToken = default); - - /// - /// 仅当当前仍是 pending 时改为 running，并增加尝试次数。 - /// - Task TryClaimAsync(int id, DateTime now, CancellationToken cancellationToken = default); - - Task MarkDoneAsync(int id, DateTime now, CancellationToken cancellationToken = default); - - Task MarkRetryAsync( - int id, - string status, - string? lastError, - DateTime? nextRunTime, - DateTime now, - CancellationToken cancellationToken = default); - - /// - /// 执行过程中进程退出时，把长时间停在 running 的步骤放回 pending。 - /// - Task ResetStaleRunningAsync(DateTime staleBefore, DateTime now, CancellationToken cancellationToken = default); - - /// - /// 仍显示「预报计费/排货处理中」、且没有待执行、执行中或已完成预报步骤的订单。 - /// - Task> ListStuckForecastOrderIdsAsync( - string processingReason, - DateTime olderThan, - int take, - CancellationToken cancellationToken = default); -} diff --git a/src/Oms.SqlSugarCore/Orders/OrderRepository.cs b/src/Oms.SqlSugarCore/Orders/OrderRepository.cs index f9ad549..c85d43f 100644 --- a/src/Oms.SqlSugarCore/Orders/OrderRepository.cs +++ b/src/Oms.SqlSugarCore/Orders/OrderRepository.cs @@ -1505,85 +1505,4 @@ public class OrderRepository : OmsSqlSugarRepository, IOrderRepository, I .Where(x => x.OrderStatus != OrderStatusCodes.Deleted) .ToListAsync(cancellationToken); } - - public async Task UpdatePredeclareFailureAsync( - int orderId, - string reason, - byte? isRemote = null, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - var now = DateTime.Now; - var update = db.Updateable() - .SetColumns(x => x.OrderStatus == OrderStatusCodes.Draft) - .SetColumns(x => x.PredeclareFailReason == reason) - .SetColumns(x => x.UpdateTime == now); - if (isRemote.HasValue) - { - var remote = isRemote.Value; - update = update.SetColumns(x => x.IsRemote == remote); - } - - await update.Where(x => x.Id == orderId).ExecuteCommandAsync(cancellationToken); - } - - public async Task UpdateForecastCompletionAsync(Order order, CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - var now = DateTime.Now; - await db.Updateable() - .SetColumns(x => x.IsRemote == order.IsRemote) - .SetColumns(x => x.ChannelCode == order.ChannelCode) - .SetColumns(x => x.DownstreamSupplierCode == order.DownstreamSupplierCode) - .SetColumns(x => x.DownstreamSupplierName == order.DownstreamSupplierName) - .SetColumns(x => x.DownstreamAppKey == order.DownstreamAppKey) - .SetColumns(x => x.DownstreamAppId == order.DownstreamAppId) - .SetColumns(x => x.ChannelForecastCode == order.ChannelForecastCode) - .SetColumns(x => x.DownstreamServiceNo == order.DownstreamServiceNo) - .SetColumns(x => x.TrackNo == order.TrackNo) - .SetColumns(x => x.AllocationPendingReason == order.AllocationPendingReason) - .SetColumns(x => x.AllocationScheme == order.AllocationScheme) - .SetColumns(x => x.AllocationTraceJson == order.AllocationTraceJson) - .SetColumns(x => x.AllocationTraceTime == order.AllocationTraceTime) - .SetColumns(x => x.PredeclareFailReason == string.Empty) - .SetColumns(x => x.MemberLevelId == order.MemberLevelId) - .SetColumns(x => x.PriceTierKey == order.PriceTierKey) - .SetColumns(x => x.ReceivableSubtotal == order.ReceivableSubtotal) - .SetColumns(x => x.LevelDiscountAmount == order.LevelDiscountAmount) - .SetColumns(x => x.CouponInstanceId == order.CouponInstanceId) - .SetColumns(x => x.CouponDiscountAmount == order.CouponDiscountAmount) - .SetColumns(x => x.PayableAmount == order.PayableAmount) - .SetColumns(x => x.UpdateTime == now) - .Where(x => x.Id == order.Id) - .ExecuteCommandAsync(cancellationToken); - - await db.Updateable() - .SetColumns(x => x.OrderStatus == OrderStatusCodes.Predeclared) - .SetColumns(x => x.UpdateTime == now) - .Where(x => x.Id == order.Id - && x.OrderStatus != OrderStatusCodes.Problematic - && x.OrderStatus != OrderStatusCodes.Claimed) - .ExecuteCommandAsync(cancellationToken); - - await db.Updateable() - .SetColumns(x => x.PreviousOrderStatus == OrderStatusCodes.Predeclared) - .SetColumns(x => x.UpdateTime == now) - .Where(x => x.Id == order.Id - && (x.OrderStatus == OrderStatusCodes.Problematic - || x.OrderStatus == OrderStatusCodes.Claimed)) - .ExecuteCommandAsync(cancellationToken); - } - - public async Task UpdateInterceptFlagAsync( - int orderId, - byte isIntercept, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - await db.Updateable() - .SetColumns(x => x.IsIntercept == isIntercept) - .SetColumns(x => x.UpdateTime == DateTime.Now) - .Where(x => x.Id == orderId) - .ExecuteCommandAsync(cancellationToken); - } } diff --git a/src/Oms.SqlSugarCore/Orders/OrderSideEffectRepository.cs b/src/Oms.SqlSugarCore/Orders/OrderSideEffectRepository.cs deleted file mode 100644 index 4d6290a..0000000 --- a/src/Oms.SqlSugarCore/Orders/OrderSideEffectRepository.cs +++ /dev/null @@ -1,153 +0,0 @@ -using Fake.DependencyInjection; -using Oms.Domain.OrdersAggregate.Consts; -using SqlSugar; -using Oms.Domain.OrdersAggregate.Entities; -using Oms.Domain.OrdersAggregate.Repositories; -using Oms.SqlSugarCore.Repositories; - -namespace Oms.SqlSugarCore.Orders; - -public class OrderSideEffectRepository : OmsSqlSugarRepository, IOrderSideEffectRepository, IScopedDependency -{ - public async Task ExistsAsync( - int orderId, - string step, - bool includeDone, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - var statuses = includeDone - ? new[] - { - OrderSideEffectStatusCodes.Pending, - OrderSideEffectStatusCodes.Running, - OrderSideEffectStatusCodes.Done - } - : new[] - { - OrderSideEffectStatusCodes.Pending, - OrderSideEffectStatusCodes.Running - }; - - return await db.Queryable() - .AnyAsync( - x => x.OrderId == orderId - && x.Step == step - && statuses.Contains(x.Status) - && x.IsDeleted == false, - cancellationToken); - } - - public async Task> ListDueIdsAsync(int take, DateTime now, CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - return await db.Queryable() - .Where(x => x.Status == OrderSideEffectStatusCodes.Pending - && x.IsDeleted == false - && (x.NextRunTime == null || x.NextRunTime <= now)) - .OrderBy(x => x.Id) - .Take(take) - .Select(x => x.Id) - .ToListAsync(cancellationToken); - } - - public async Task FindLatestPayloadJsonAsync( - int orderId, - string step, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - var rows = await db.Queryable() - .Where(x => x.OrderId == orderId && x.Step == step && x.IsDeleted == false) - .OrderByDescending(x => x.Id) - .Take(1) - .Select(x => x.PayloadJson) - .ToListAsync(cancellationToken); - return rows.FirstOrDefault(); - } - - public async Task TryClaimAsync(int id, DateTime now, CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - var affected = await db.Updateable() - .SetColumns(x => x.Status == OrderSideEffectStatusCodes.Running) - .SetColumns(x => x.UpdateTime == now) - .SetColumns(x => x.AttemptCount == x.AttemptCount + 1) - .Where(x => x.Id == id - && x.Status == OrderSideEffectStatusCodes.Pending - && x.IsDeleted == false) - .ExecuteCommandAsync(cancellationToken); - return affected == 1; - } - - public async Task MarkDoneAsync(int id, DateTime now, CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - await db.Updateable() - .SetColumns(x => x.Status == OrderSideEffectStatusCodes.Done) - .SetColumns(x => x.UpdateTime == now) - .SetColumns(x => x.LastError == null) - .Where(x => x.Id == id) - .ExecuteCommandAsync(cancellationToken); - } - - public async Task MarkRetryAsync( - int id, - string status, - string? lastError, - DateTime? nextRunTime, - DateTime now, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - await db.Updateable() - .SetColumns(x => x.Status == status) - .SetColumns(x => x.LastError == lastError) - .SetColumns(x => x.NextRunTime == nextRunTime) - .SetColumns(x => x.UpdateTime == now) - .Where(x => x.Id == id) - .ExecuteCommandAsync(cancellationToken); - } - - public async Task ResetStaleRunningAsync( - DateTime staleBefore, - DateTime now, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - return await db.Updateable() - .SetColumns(x => x.Status == OrderSideEffectStatusCodes.Pending) - .SetColumns(x => x.NextRunTime == now) - .SetColumns(x => x.UpdateTime == now) - .Where(x => x.Status == OrderSideEffectStatusCodes.Running - && x.IsDeleted == false - && x.UpdateTime < staleBefore) - .ExecuteCommandAsync(cancellationToken); - } - - public async Task> ListStuckForecastOrderIdsAsync( - string processingReason, - DateTime olderThan, - int take, - CancellationToken cancellationToken = default) - { - var db = await GetDbContextAsync(cancellationToken); - return await db.Queryable() - .Where(order => order.IsDeleted == false - && order.PredeclareFailReason == processingReason - && order.UpdateTime < olderThan - && order.UpdateTime > DateTime.MinValue) - .Where(order => !SqlSugar.SqlFunc.Subqueryable() - .Where(effect => effect.OrderId == order.Id - && effect.Step == OrderSideEffectSteps.ForecastBilling - && effect.IsDeleted == false - && (effect.Status == OrderSideEffectStatusCodes.Pending - || effect.Status == OrderSideEffectStatusCodes.Running - || effect.Status == OrderSideEffectStatusCodes.Done)) - .Any()) - .OrderBy(order => order.Id) - .Take(take) - .Select(order => order.Id) - .ToListAsync(cancellationToken); - } -}