diff --git a/db.sql b/db.sql index 3f1111e..199ec6e 100644 --- a/db.sql +++ b/db.sql @@ -101,6 +101,7 @@ 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`; @@ -306,6 +307,26 @@ 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/migrate_order_side_effect.sql b/sql/migrate_order_side_effect.sql new file mode 100644 index 0000000..20fbf7a --- /dev/null +++ b/sql/migrate_order_side_effect.sql @@ -0,0 +1,39 @@ +-- 创单后置步骤表：与订单同一事务写入，提交后由后台执行。 +-- 可重复执行。不建唯一索引。 + +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 9b31ee7..deda445 100644 --- a/src/Oms.Application/Messaging/OmsAfterCommit.cs +++ b/src/Oms.Application/Messaging/OmsAfterCommit.cs @@ -1,11 +1,9 @@ -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; using Fake.UnitOfWork; namespace Oms.Application.Messaging; /// -/// 把创单/预报之后的副作用延后到当前事务提交成功再执行，每一步单独提交。 +/// 当前事务提交并释放后再执行。回滚不会执行。 /// internal static class OmsAfterCommit { @@ -14,13 +12,36 @@ internal static class OmsAfterCommit var current = unitOfWorkManager.Current; if (current != null && !current.IsCompleted) { - // OnCompleted 仍处在创单事务里，此时再写 Outbox 征用不到连接，事件会丢。 - // 等工作单元释放后再开独立事务。 + // OnCompleted 仍处在业务事务里，此时写 Outbox 征用不到连接。 + // 等工作单元成功释放后再开独立事务。回滚时 IsCompleted 为 false，不跑。 EventHandler handler = null!; handler = (_, _) => { current.Disposed -= handler; - _ = Task.Run(action); + if (!current.IsCompleted) + { + return; + } + + var suppressed = ExecutionContext.SuppressFlow(); + try + { + _ = Task.Run(async () => + { + try + { + await action(); + } + catch + { + // 具体步骤在执行器里记日志。这里避免未观察的任务异常。 + } + }); + } + finally + { + suppressed.Dispose(); + } }; current.Disposed += handler; return Task.CompletedTask; @@ -28,30 +49,4 @@ 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 new file mode 100644 index 0000000..bf46bc5 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffectExecutor.cs @@ -0,0 +1,202 @@ +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 new file mode 100644 index 0000000..8f629c5 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffectPayload.cs @@ -0,0 +1,34 @@ +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 new file mode 100644 index 0000000..1bd78a4 --- /dev/null +++ b/src/Oms.Application/Messaging/OrderSideEffectQueue.cs @@ -0,0 +1,83 @@ +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 c24e699..1df839c 100644 --- a/src/Oms.Application/OmsApplicationModule.cs +++ b/src/Oms.Application/OmsApplicationModule.cs @@ -65,5 +65,6 @@ 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 071ad64..8930e13 100644 --- a/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs +++ b/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs @@ -182,10 +182,7 @@ public sealed class OrderBatchJobProcessor( text = text[..512]; } - order.OrderStatus = OrderStatusCodes.Draft; - order.PredeclareFailReason = text; - order.UpdateTime = DateTime.Now; - await orderRepository.UpdateAsync(order, cancellationToken); + await orderRepository.UpdatePredeclareFailureAsync(order.Id, text, cancellationToken: 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 34d536e..69243c9 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -17,6 +17,7 @@ public class AutoTriggerRuntimeService( OrderExceptionWorkflowService exceptionWorkflowService, OrderIssueNotificationService issueNotificationService, IOrderSideEffectEventPublisher sideEffectEventPublisher, + IOrderExceptionEventRepository orderExceptionEventRepository, ILogger logger) : IScopedDependency { private const string InterceptOperation = "拦截"; @@ -81,6 +82,97 @@ 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, @@ -92,7 +184,7 @@ public class AutoTriggerRuntimeService( if (order.IsIntercept == 0) { order.IsIntercept = 1; - await orderRepository.UpdateAsync(order); + await orderRepository.UpdateInterceptFlagAsync(order.Id, 1); } return; diff --git a/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs b/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs index 374b18a..430e184 100644 --- a/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs +++ b/src/Oms.Application/Services/Orders/Issues/OrderExceptionWorkflowService.cs @@ -42,9 +42,10 @@ public class OrderExceptionWorkflowService( int operatorUserId, bool sendNotification) { - if (!OrderExceptionStatusRules.CanMarkException(order.OrderStatus)) + var current = await orderRepository.FirstOrDefaultAsync(x => x.Id == order.Id) ?? order; + if (!OrderExceptionStatusRules.CanMarkException(current.OrderStatus)) { - return OrderExceptionStatusRules.GetForbiddenReason(order.OrderStatus); + return OrderExceptionStatusRules.GetForbiddenReason(current.OrderStatus); } IssueTemplate? template = null; @@ -56,14 +57,14 @@ public class OrderExceptionWorkflowService( IssueHandlerResolution handlerResolution; try { - handlerResolution = await issueHandlerResolver.ResolveAsync(order, template, request.HandlerNames); + handlerResolution = await issueHandlerResolver.ResolveAsync(current, template, request.HandlerNames); } catch (BusinessException ex) { return ex.Message; } - var context = IssuePlaceholderContext.FromOrder(order, request.IssueType, request.Title); + var context = IssuePlaceholderContext.FromOrder(current, request.IssueType, request.Title); var renderedTitle = IssuePlaceholderRenderer.Render(request.Title, context); var renderedContent = IssuePlaceholderRenderer.Render(request.Content, context); @@ -73,28 +74,37 @@ public class OrderExceptionWorkflowService( notifyMethod = IssueNotifyMethodCodes.InApp; } - var fromStatus = order.OrderStatus; + var fromStatus = current.OrderStatus; var isAppend = string.Equals(fromStatus, OrderStatusCodes.Problematic, StringComparison.OrdinalIgnoreCase); if (!isAppend) { - order.PreviousOrderStatus = fromStatus; - order.OrderStatus = OrderStatusCodes.Problematic; + current.PreviousOrderStatus = fromStatus; + current.OrderStatus = OrderStatusCodes.Problematic; } - 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; + 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; - await orderRepository.UpdateAsync(order); + 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 orderExceptionEventRepository.InsertAsync(new OrderExceptionEvent { OrderId = order.Id, @@ -120,7 +130,7 @@ public class OrderExceptionWorkflowService( ? handlerResolution.NotifyUserIds : null; await issueNotificationService.DispatchAsync( - order, + current, renderedTitle, renderedContent, notifyMethod, diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index 2d66718..3270344 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -170,11 +170,12 @@ 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); - await ScheduleCreateSideEffectsAsync(order.Id, input, draftOutput, includeForecast: false, operatorId); + ApplyPickupPlan(input, draftOutput); + await EnqueueCreateSideEffectsAsync(order.Id, input, includeForecast: false, operatorId); return draftOutput; } @@ -193,7 +194,8 @@ public partial class OrderService await OrderRepository.UpdateAsync(order); var output = await BuildParcelCreateOutputAsync(order); - await ScheduleCreateSideEffectsAsync(order.Id, input, output, includeForecast: true, operatorId); + ApplyPickupPlan(input, output); + await EnqueueCreateSideEffectsAsync(order.Id, input, includeForecast: true, operatorId); return output; } catch (BusinessException) @@ -2284,7 +2286,7 @@ public partial class OrderService } /// - /// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货投递 ForecastBilling MQ。 + /// 草稿重新预报：同步做黑名单等硬校验，计重/应收/排货写入后置步骤，提交后执行。 /// internal async Task CompleteDraftPredeclareAsync(Order order, int? operatorUserId = null) { @@ -2329,38 +2331,7 @@ public partial class OrderService await OrderRepository.UpdateAsync(order); var operatorId = operatorUserId ?? (CurrentUser.IsAuthenticated ? CurrentUser.GetId() : 0); - 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)); - } + await EnqueueDraftPredeclareSideEffectsAsync(order.Id, order.CouponInstanceId, operatorId); } catch (BusinessException ex) { @@ -2387,10 +2358,11 @@ public partial class OrderService private async Task MarkParcelDraftFailedAsync(Order order, string reason) { + var text = TrimPendingReason(reason); order.OrderStatus = OrderStatusCodes.Draft; - order.PredeclareFailReason = reason; + order.PredeclareFailReason = text; order.UpdateTime = DateTime.Now; - await OrderRepository.UpdateAsync(order); + await OrderRepository.UpdatePredeclareFailureAsync(order.Id, text, order.IsRemote); } /// @@ -2823,10 +2795,23 @@ public partial class OrderService // ignore } - order.OrderStatus = OrderStatusCodes.Predeclared; order.PredeclareFailReason = string.Empty; order.UpdateTime = DateTime.Now; - await OrderRepository.UpdateAsync(order); + 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 EnqueueSettlementSideEffectsAsync(order.Id, operatorUserId); var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeFees: true, includeWeights: true); await OperationLogWriter.TryWriteAsync( @@ -2846,93 +2831,6 @@ 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, "track-upsert", sp => - sp.GetRequiredService().PublishForecastTrackAsync(orderId)); - 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, "track-upsert", sp => - sp.GetRequiredService().PublishForecastTrackAsync(orderId)); - 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)); - }); - } - private async Task PublishCustomsForOrderAsync(int orderId, string? platformCode) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); @@ -2945,19 +2843,7 @@ public partial class OrderService await ParcelCreateEventPublisher.PublishCustoms9610UpsertAsync(order, platformCode); } - 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) + internal async Task MarkForecastEnqueueFailedIfStillProcessingAsync(int orderId) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); if (order == null @@ -2969,7 +2855,7 @@ public partial class OrderService await MarkParcelDraftFailedAsync(order, ForecastBillingEnqueueFailReason); } - private async Task PublishForecastTrackAsync(int orderId) + private async Task PublishForecastTrackAsync(int orderId, bool appendInitialTrack, string? authorization) { var order = await OrderRepository.FirstOrDefaultAsync(x => x.Id == orderId); if (order == null @@ -2982,8 +2868,12 @@ public partial class OrderService ? await CustomerRepository.FirstOrDefaultAsync(x => x.Id == order.CustomerId.Value) : null; var request = await BuildSmallBagTrackUpsertRequestAsync(order, receiverInput: null); - request.AppendInitialTrackOnSuccess = true; - ApplyTrackAuthorization(request); + request.AppendInitialTrackOnSuccess = appendInitialTrack; + if (!string.IsNullOrWhiteSpace(authorization)) + { + request.Authorization = authorization.Trim(); + } + 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 new file mode 100644 index 0000000..204230f --- /dev/null +++ b/src/Oms.Application/Services/Orders/OrderService.SideEffects.cs @@ -0,0 +1,317 @@ +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 eaccbe2..ff7c50b 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) diff --git a/src/Oms.Application/Workers/OrderSideEffectWorker.cs b/src/Oms.Application/Workers/OrderSideEffectWorker.cs new file mode 100644 index 0000000..9f28943 --- /dev/null +++ b/src/Oms.Application/Workers/OrderSideEffectWorker.cs @@ -0,0 +1,41 @@ +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 new file mode 100644 index 0000000..9fc3d01 --- /dev/null +++ b/src/Oms.Domain/OrdersAggregate/Consts/OrderSideEffectCodes.cs @@ -0,0 +1,44 @@ +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 new file mode 100644 index 0000000..feb5471 --- /dev/null +++ b/src/Oms.Domain/OrdersAggregate/Entities/OrderSideEffect.cs @@ -0,0 +1,30 @@ +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 c5d8ab0..cf35dfd 100644 --- a/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs +++ b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderRepository.cs @@ -137,4 +137,23 @@ 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 new file mode 100644 index 0000000..7d7e724 --- /dev/null +++ b/src/Oms.Domain/OrdersAggregate/Repositories/IOrderSideEffectRepository.cs @@ -0,0 +1,55 @@ +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 c85d43f..f9ad549 100644 --- a/src/Oms.SqlSugarCore/Orders/OrderRepository.cs +++ b/src/Oms.SqlSugarCore/Orders/OrderRepository.cs @@ -1505,4 +1505,85 @@ 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 new file mode 100644 index 0000000..4d6290a --- /dev/null +++ b/src/Oms.SqlSugarCore/Orders/OrderSideEffectRepository.cs @@ -0,0 +1,153 @@ +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); + } +}