1.05 #7

Merged
zzh merged 19 commits from test into master 2026-10-15 13:59:09 +08:00
20 changed files with 1307 additions and 203 deletions
Showing only changes of commit 34c47a3173 - Show all commits

21
db.sql
View File

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

View File

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

View File

@ -1,11 +1,9 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Fake.UnitOfWork;
namespace Oms.Application.Messaging;
/// <summary>
/// 把创单/预报之后的副作用延后到当前事务提交成功再执行,每一步单独提交。
/// 当前事务提交并释放后再执行。回滚不会执行。
/// </summary>
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<UnitOfWorkEventArgs> 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();
}
/// <summary>
/// 新开独立事务执行一步。失败只记日志,不抛回调用方。
/// </summary>
public static async Task<bool> RunStepAsync(
IServiceScopeFactory scopeFactory,
ILogger logger,
string step,
Func<IServiceProvider, Task> work)
{
try
{
using var scope = scopeFactory.CreateScope();
var unitOfWorkManager = scope.ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
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;
}
}
}

View File

@ -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;
/// <summary>
/// 每一步后置任务单独开事务执行。失败按次数重试,超过次数记失败。
/// </summary>
public sealed class OrderSideEffectExecutor(
IOrderSideEffectRepository repository,
IUnitOfWorkManager unitOfWorkManager,
OrderService orderService,
ILogger<OrderSideEffectExecutor> logger) : IScopedDependency
{
private static readonly TimeSpan StaleRunning = TimeSpan.FromMinutes(2);
/// <summary>
/// 当前事务提交成功后再执行到期步骤。回滚不会触发。
/// </summary>
public static Task ScheduleAfterCommitAsync(
IUnitOfWorkManager unitOfWorkManager,
IServiceScopeFactory scopeFactory,
ILogger logger)
{
return OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () => ProcessDueAsync(scopeFactory, logger));
}
/// <summary>
/// 捞出到期步骤并逐条执行。可重复调用,抢占失败的步骤会被跳过。
/// </summary>
public static async Task ProcessDueAsync(
IServiceScopeFactory scopeFactory,
ILogger logger,
CancellationToken cancellationToken = default)
{
List<int> ids;
try
{
using var scope = scopeFactory.CreateScope();
var repository = scope.ServiceProvider.GetRequiredService<IOrderSideEffectRepository>();
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<OrderSideEffectExecutor>();
await executor.ExecuteAsync(id, cancellationToken);
}
catch (Exception ex)
{
logger.LogError(ex, "订单后置步骤执行失败 SideEffectId={SideEffectId}", id);
}
}
}
/// <summary>
/// 仍停在「预报计费/排货处理中」且没有未完成预报步骤的订单,补一条预报计费步骤。
/// </summary>
public static async Task RecoverStuckForecastAsync(
IServiceScopeFactory scopeFactory,
ILogger logger,
CancellationToken cancellationToken = default)
{
try
{
using var scope = scopeFactory.CreateScope();
var unitOfWorkManager = scope.ServiceProvider.GetRequiredService<IUnitOfWorkManager>();
using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Independent());
var repository = scope.ServiceProvider.GetRequiredService<IOrderSideEffectRepository>();
var queue = scope.ServiceProvider.GetRequiredService<OrderSideEffectQueue>();
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<bool> 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<bool> 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);
}
}
}

View File

@ -0,0 +1,34 @@
namespace Oms.Application.Messaging;
/// <summary>
/// 后置步骤入参。轨迹授权头只在执行时使用,不要写进日志。
/// </summary>
public sealed class OrderSideEffectPayload
{
public int OperatorUserId { get; set; }
public string? PlatformCode { get; set; }
public int? CouponInstanceId { get; set; }
/// <summary>为 true 时轨迹创建成功后追加 A0101。</summary>
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; }
}

View File

@ -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;
/// <summary>
/// 把创单后置步骤写入当前事务。事务回滚时这些行一起消失。
/// </summary>
public sealed class OrderSideEffectQueue(IOrderSideEffectRepository repository) : IScopedDependency
{
public const int MaxAttempts = 5;
/// <summary>
/// 同一步骤已有 pending、running 或 done 时不再插入。用于 A0101、揽收和创单当下的自动触发。
/// </summary>
public Task EnqueueOnceAsync(
int orderId,
string step,
OrderSideEffectPayload? payload,
CancellationToken cancellationToken = default) =>
EnqueueAsync(orderId, step, payload, includeDone: true, cancellationToken);
/// <summary>
/// 没有 pending 或 running 时才插入。已完成的步骤允许再入队,例如重新预报、排货后刷新轨迹。
/// </summary>
public Task EnqueueIfIdleAsync(
int orderId,
string step,
OrderSideEffectPayload? payload,
CancellationToken cancellationToken = default) =>
EnqueueAsync(orderId, step, payload, includeDone: false, cancellationToken);
/// <summary>
/// 读取同一步骤最近一次入参。没有记录时返回 null。
/// </summary>
public async Task<OrderSideEffectPayload?> 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<OrderSideEffectPayload>(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);
}
}

View File

@ -65,5 +65,6 @@ public class OmsApplicationModule : FakeModule
context.Services.AddAiLlmClient();
context.Services.AddHostedService<XhsScheduledPublishWorker>();
context.Services.AddHostedService<MarketingBackgroundWorker>();
context.Services.AddHostedService<OrderSideEffectWorker>();
}
}

View File

@ -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)

View File

@ -17,6 +17,7 @@ public class AutoTriggerRuntimeService(
OrderExceptionWorkflowService exceptionWorkflowService,
OrderIssueNotificationService issueNotificationService,
IOrderSideEffectEventPublisher sideEffectEventPublisher,
IOrderExceptionEventRepository orderExceptionEventRepository,
ILogger<AutoTriggerRuntimeService> logger) : IScopedDependency
{
private const string InterceptOperation = "拦截";
@ -81,6 +82,97 @@ public class AutoTriggerRuntimeService(
await FireAsync(order, timing, operatorUserId);
}
/// <summary>
/// 预报计费完成后再评估。只跑依赖计费/渠道/费用的规则,以及草稿阶段没能打上的问题件。
/// </summary>
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<bool> 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;

View File

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

View File

@ -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
}
/// <summary>
/// 草稿重新预报:同步做黑名单等硬校验,计重/应收/排货投递 ForecastBilling MQ。
/// 草稿重新预报:同步做黑名单等硬校验,计重/应收/排货写入后置步骤,提交后执行。
/// </summary>
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<int>() : 0);
var publisher = LazyServiceProvider.GetRequiredService<IOrderBatchJobPublisher>();
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);
}
/// <summary>
@ -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; }
}
/// <summary>
/// 创单事务提交后,9610、预报计费、揽收、轨迹、面单、自动触发各自提交,互不等待。
/// </summary>
private async Task ScheduleCreateSideEffectsAsync(
int orderId,
OrderParcelCreateInput input,
OrderParcelCreateOutput output,
bool includeForecast,
int operatorUserId)
{
var unitOfWorkManager = LazyServiceProvider.GetRequiredService<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().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<OrderService>().PublishCustomsForOrderAsync(orderId, platformCode));
var forecastQueued = true;
if (includeForecast)
{
forecastQueued = await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", async sp =>
{
var service = sp.GetRequiredService<OrderService>();
await service.EnqueueForecastBillingByIdAsync(orderId, couponInstanceId, operatorUserId);
});
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp =>
sp.GetRequiredService<OrderService>().PublishForecastTrackAsync(orderId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
sp.GetRequiredService<OrderService>().PublishForecastAutoTriggerAsync(orderId, operatorUserId));
}
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "pickup-attach", sp =>
sp.GetRequiredService<OrderService>().TryAttachPickupAfterParcelCreateAsync(input, output));
if (includeForecast && !forecastQueued)
{
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing-fail", sp =>
sp.GetRequiredService<OrderService>().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<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.ForecastBilling");
await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, () =>
OmsAfterCommit.RunStepAsync(scopeFactory, logger, "forecast-billing", sp =>
sp.GetRequiredService<OrderService>().EnqueueForecastBillingByIdAsync(
orderId,
couponInstanceId,
operatorUserId)));
}
/// <summary>
/// 创单或草稿重新预报提交后,轨迹、面单、下单自动触发各自入队,不等预报计费。
/// </summary>
private async Task ScheduleForecastSideEffectsAsync(int orderId, int operatorUserId)
{
var unitOfWorkManager = LazyServiceProvider.GetRequiredService<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.ForecastBilling");
await OmsAfterCommit.ScheduleAsync(unitOfWorkManager, async () =>
{
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "track-upsert", sp =>
sp.GetRequiredService<OrderService>().PublishForecastTrackAsync(orderId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "label-generate", sp =>
sp.GetRequiredService<OrderService>().PublishForecastLabelAsync(orderId, operatorUserId));
await OmsAfterCommit.RunStepAsync(scopeFactory, logger, "auto-trigger", sp =>
sp.GetRequiredService<OrderService>().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);
}

View File

@ -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<OrderSideEffectQueue>();
/// <summary>
/// 创单事务内写入后置步骤。保存草稿只写 9610 和揽收;确认下单再写预报、轨迹、面单和自动触发。
/// </summary>
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();
}
/// <summary>
/// 草稿重新预报:预报计费可再次入队;轨迹 A0101、面单和创单触发已做过则跳过。
/// </summary>
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();
}
/// <summary>
/// 预报计费成功后补轨迹渠道、系统面单,以及依赖计费结果的自动触发。
/// </summary>
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<IUnitOfWorkManager>();
var scopeFactory = LazyServiceProvider.GetRequiredService<IServiceScopeFactory>();
var logger = LazyServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger("Oms.OrderSideEffect");
return OrderSideEffectExecutor.ScheduleAfterCommitAsync(unitOfWorkManager, scopeFactory, logger);
}
/// <summary>
/// 执行一条已抢占的后置步骤。由独立事务调用,失败抛出以便重试。
/// </summary>
internal async Task ExecuteSideEffectAsync(OrderSideEffect row, CancellationToken cancellationToken = default)
{
var payload = string.IsNullOrWhiteSpace(row.PayloadJson)
? new OrderSideEffectPayload()
: JsonConvert.DeserializeObject<OrderSideEffectPayload>(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<ILoggerFactory>()
.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<IHttpContextAccessor>()
.HttpContext?
.Request
.Headers
.Authorization
.ToString();
}
catch (ObjectDisposedException)
{
return null;
}
}
}

View File

@ -262,10 +262,10 @@ public partial class OrderService : ApplicationService
}
/// <summary>
/// 执行订单预报。
/// 执行订单预报。校验通过后订单保持草稿并标记处理中,计费和排货在事务提交后执行。
/// </summary>
/// <param name="input">入参:订单操作参数。</param>
/// <returns>无返回体,仅完成订单预报处理。</returns>
/// <returns>无返回体。预报结果随后写回订单状态和失败原因。</returns>
[HttpPost("/order/pre-declare")]
[UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)]
public async Task PreDeclareAsync(OrderActionInput input)

View File

@ -0,0 +1,41 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Oms.Application.Messaging;
namespace Oms.Application.Workers;
/// <summary>
/// 轮询已提交的创单后置步骤。进程在事务提交后、步骤执行前退出时,靠这里补跑。
/// </summary>
public sealed class OrderSideEffectWorker(
IServiceScopeFactory scopeFactory,
ILogger<OrderSideEffectWorker> 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;
}
}
}
}

View File

@ -0,0 +1,44 @@
namespace Oms.Domain.OrdersAggregate.Consts;
/// <summary>
/// 创单后置步骤状态。同一订单、同一步骤不建唯一索引,是否重复入队由业务判断。
/// </summary>
public static class OrderSideEffectStatusCodes
{
public const string Pending = "pending";
public const string Running = "running";
public const string Done = "done";
public const string Failed = "failed";
}
/// <summary>
/// 创单/预报提交后要单独执行的步骤。
/// </summary>
public static class OrderSideEffectSteps
{
public const string Customs9610 = "customs-9610";
public const string ForecastBilling = "forecast-billing";
/// <summary>创建轨迹订单并追加 A0101,同一订单只做一次。</summary>
public const string TrackUpsert = "track-upsert";
/// <summary>排货完成后按最新渠道再同步一次轨迹,不追加 A0101。</summary>
public const string TrackRefresh = "track-refresh";
public const string LabelGenerate = "label-generate";
/// <summary>排货完成后重做系统面单,覆盖创单时还没有渠道的那张。</summary>
public const string LabelRefresh = "label-refresh";
/// <summary>下单时机的自动触发。此时还没有计费重、应收和渠道。</summary>
public const string AutoTrigger = "auto-trigger";
/// <summary>预报计费成功后再跑依赖计费结果的规则,以及草稿阶段没能打上的问题件。</summary>
public const string AutoTriggerSettled = "auto-trigger-settled";
public const string PickupAttach = "pickup-attach";
}

View File

@ -0,0 +1,30 @@
using Oms.Domain.Shared;
using SqlSugar;
namespace Oms.Domain.OrdersAggregate.Entities;
/// <summary>
/// 与订单同一事务写入的后置步骤。事务回滚则步骤一起消失;提交后由后台执行,进程重启仍可继续。
/// </summary>
[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; }
}

View File

@ -137,4 +137,23 @@ public interface IOrderRepository : IRepository<Order>
/// 只更新标签模板 Id,避免面单生成覆盖预报计费正在写入的订单字段。
/// </summary>
Task UpdateLabelTemplateIdAsync(int orderId, int labelTemplateId, CancellationToken cancellationToken = default);
/// <summary>
/// 预报失败时只写状态和原因,不覆盖渠道、面单模板、问题件字段。
/// </summary>
Task UpdatePredeclareFailureAsync(
int orderId,
string reason,
byte? isRemote = null,
CancellationToken cancellationToken = default);
/// <summary>
/// 预报计费成功时只写计费/排货字段。订单已是问题件或理赔件时不改状态,只把异常前状态改成已预报。
/// </summary>
Task UpdateForecastCompletionAsync(Order order, CancellationToken cancellationToken = default);
/// <summary>
/// 只更新拦截标记。
/// </summary>
Task UpdateInterceptFlagAsync(int orderId, byte isIntercept, CancellationToken cancellationToken = default);
}

View File

@ -0,0 +1,55 @@
using Fake.Domain.Repositories;
using Oms.Domain.OrdersAggregate.Entities;
namespace Oms.Domain.OrdersAggregate.Repositories;
public interface IOrderSideEffectRepository : IRepository<OrderSideEffect>
{
/// <summary>
/// 是否已有未结束(以及可选的已完成)步骤。用于避免重复入队。
/// </summary>
Task<bool> ExistsAsync(
int orderId,
string step,
bool includeDone,
CancellationToken cancellationToken = default);
Task<List<int>> ListDueIdsAsync(int take, DateTime now, CancellationToken cancellationToken = default);
/// <summary>
/// 取同一步骤最近一条入参,用于把创单时捕获的轨迹授权头带到排货后的刷新。
/// </summary>
Task<string?> FindLatestPayloadJsonAsync(
int orderId,
string step,
CancellationToken cancellationToken = default);
/// <summary>
/// 仅当当前仍是 pending 时改为 running,并增加尝试次数。
/// </summary>
Task<bool> 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);
/// <summary>
/// 执行过程中进程退出时,把长时间停在 running 的步骤放回 pending。
/// </summary>
Task<int> ResetStaleRunningAsync(DateTime staleBefore, DateTime now, CancellationToken cancellationToken = default);
/// <summary>
/// 仍显示「预报计费/排货处理中」、且没有待执行、执行中或已完成预报步骤的订单。
/// </summary>
Task<List<int>> ListStuckForecastOrderIdsAsync(
string processingReason,
DateTime olderThan,
int take,
CancellationToken cancellationToken = default);
}

View File

@ -1505,4 +1505,85 @@ public class OrderRepository : OmsSqlSugarRepository<Order>, 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<Order>()
.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<Order>()
.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<Order>()
.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<Order>()
.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<Order>()
.SetColumns(x => x.IsIntercept == isIntercept)
.SetColumns(x => x.UpdateTime == DateTime.Now)
.Where(x => x.Id == orderId)
.ExecuteCommandAsync(cancellationToken);
}
}

View File

@ -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<OrderSideEffect>, IOrderSideEffectRepository, IScopedDependency
{
public async Task<bool> 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<OrderSideEffect>()
.AnyAsync(
x => x.OrderId == orderId
&& x.Step == step
&& statuses.Contains(x.Status)
&& x.IsDeleted == false,
cancellationToken);
}
public async Task<List<int>> ListDueIdsAsync(int take, DateTime now, CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
return await db.Queryable<OrderSideEffect>()
.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<string?> FindLatestPayloadJsonAsync(
int orderId,
string step,
CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
var rows = await db.Queryable<OrderSideEffect>()
.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<bool> TryClaimAsync(int id, DateTime now, CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
var affected = await db.Updateable<OrderSideEffect>()
.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<OrderSideEffect>()
.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<OrderSideEffect>()
.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<int> ResetStaleRunningAsync(
DateTime staleBefore,
DateTime now,
CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
return await db.Updateable<OrderSideEffect>()
.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<List<int>> ListStuckForecastOrderIdsAsync(
string processingReason,
DateTime olderThan,
int take,
CancellationToken cancellationToken = default)
{
var db = await GetDbContextAsync(cancellationToken);
return await db.Queryable<Order>()
.Where(order => order.IsDeleted == false
&& order.PredeclareFailReason == processingReason
&& order.UpdateTime < olderThan
&& order.UpdateTime > DateTime.MinValue)
.Where(order => !SqlSugar.SqlFunc.Subqueryable<OrderSideEffect>()
.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);
}
}