import { createTask, listTasksByOrderId, updateTask } from '../../repositories/task-repo.js' import { createTaskEvent } from '../../repositories/task-event-repo.js' import { getFulfillmentProfileByKey } from '../../repositories/fulfillment-profile-repo.js' import { createTaskClaimToken } from '../claim/claim-service.js' import { notifyTaskAutoManualReview } from '../notification/domain-notifications.js' import { planFulfillmentTaskForOrderItem, type FulfillmentBindingLike, } from '../fulfillment/planner.js' import { preparePaidFulfillmentTask } from '../fulfillment/executors/registry.js' import { FULFILLMENT_EXECUTOR_KEYS, type FulfillmentPrepareDeps, } from '../fulfillment/executors/types.js' import { nowIso } from '../../utils/time.js' import { randomId } from '../../utils/random.js' import { logInfo, logWarn } from '../../utils/logger.js' import { TASK_STATUS, resolveInitialPaidTaskStatus } from '../../domain/task-status.js' import type { JsonObject } from '../../types/json.js' import type { OrderItemRow, OrderRow, TaskRow } from '../../types/repository/rows.js' import { findConflictingCarrier, loadActiveCarrierEvidence, } from '../fulfillment/carrier-evidence-service.js' import { isInactiveCarrierTaskStatus, type FulfillmentChannel, } from '../fulfillment/channel-types.js' type ClaimTokenLike = { token: string expired_at: string [key: string]: unknown } type DeliveryTaskRow = TaskRow & { skuCode?: string skuName?: string } type DeliveryTaskDeps = { createTask?: typeof createTask listTasksByOrderId?: typeof listTasksByOrderId loadActiveCarrierEvidence?: typeof loadActiveCarrierEvidence updateTask?: typeof updateTask getFulfillmentProfileByKey?: (profileKey: string) => Promise createTaskClaimToken?: (taskId: number | string) => Promise notifyTaskAutoManualReview?: (payload: { task: unknown reason: string source: string }) => Promise | unknown nowIso?: () => string randomId?: (prefix?: string) => string createTaskEvent?: typeof createTaskEvent logInfo?: typeof logInfo logWarn?: typeof logWarn allowInactiveTaskRetry?: boolean } export async function syncDeliveryTasksForOrder( order: OrderRow, orderItems: OrderItemRow[], options: { allowInactiveTaskRetry?: boolean } = {}, ): Promise { return syncDeliveryTasksForOrderWithDeps(order, orderItems, options) } export async function syncDeliveryTasksForOrderWithDeps( order: OrderRow, orderItems: OrderItemRow[], deps: DeliveryTaskDeps = {}, ): Promise { const { createTask: createDeliveryTask = createTask, listTasksByOrderId: listTasks = listTasksByOrderId, loadActiveCarrierEvidence: loadCarrierEvidence = loadActiveCarrierEvidence, updateTask: updateDeliveryTask = updateTask, getFulfillmentProfileByKey: getProfileByKey = getFulfillmentProfileByKey, createTaskClaimToken: createClaimToken = createTaskClaimToken, notifyTaskAutoManualReview: notifyManualReview = notifyTaskAutoManualReview, nowIso: getNowIso = nowIso, randomId: createRandomId = randomId, createTaskEvent: appendTaskEvent = createTaskEvent, logInfo: writeInfoLog = logInfo, logWarn: writeWarnLog = logWarn, allowInactiveTaskRetry = false, } = deps const runtimeDeps: FulfillmentPrepareDeps = { updateTask: updateDeliveryTask, createTaskClaimToken: createClaimToken, notifyTaskAutoManualReview: notifyManualReview, nowIso: getNowIso, } const existingTasks = await listTasks(order.id) const tasks: DeliveryTaskRow[] = [...existingTasks] const existingTaskKeys = new Set( (allowInactiveTaskRetry ? existingTasks.filter((task) => !isInactiveCarrierTaskStatus(task.task_status)) : existingTasks ).map(buildTaskUnitKey), ) const carrierEvidence = await loadCarrierEvidence(orderItems.map((item) => Number(item.id))) writeInfoLog('[delivery-task-service]', '开始核对订单履约任务完整性', { ...buildOrderLogContext(order), orderItemCount: orderItems.length, existingTaskCount: existingTasks.length, }) for (const item of orderItems) { const quantity = Math.max(1, Number(item.quantity || 1)) const missingUnitIndexes = Array.from({ length: quantity }, (_, index) => index + 1).filter( (unitIndex) => !existingTaskKeys.has(buildOrderItemUnitKey(item.id, unitIndex)), ) if (missingUnitIndexes.length === 0) { continue } let plan try { plan = await planFulfillmentTaskForOrderItem({ order, item, getProfileByKey, }) } catch (error) { writeWarnLog('[delivery-task-service]', '订单项履约任务规划执行异常', { ...buildOrderLogContext(order), ...buildOrderItemLogContext(item), missingUnitIndexes, planningDiagnostic: buildPlanningDiagnostic(item), error: error instanceof Error ? error.message : String(error), }) throw error } if (!plan) { writeWarnLog('[delivery-task-service]', '订单项已配置履约路由但无法生成任务计划', { ...buildOrderLogContext(order), ...buildOrderItemLogContext(item), missingUnitIndexes, planningDiagnostic: buildPlanningDiagnostic(item), }) continue } const plannedChannel = executorKeyToFulfillmentChannel(plan.executorKey) if (plannedChannel) { const assignedChannel = String(item.fulfillment_channel || '').trim() if (item.fulfillment_status === 'assigned' && assignedChannel !== plannedChannel) { writeInfoLog('[delivery-task-service]', '订单项已分配其它渠道,跳过载体补建', { ...buildOrderItemLogContext(item), assignedChannel, plannedChannel, }) continue } const conflictingCarrier = findConflictingCarrier( carrierEvidence.get(Number(item.id)), plannedChannel, ) if (conflictingCarrier) { writeWarnLog('[delivery-task-service]', '订单项已有其它真实载体,跳过载体补建', { ...buildOrderItemLogContext(item), existingCarrier: conflictingCarrier, plannedChannel, }) continue } } for (const unitIndex of missingUnitIndexes) { const createdAt = getNowIso() const initialStatus = order.pay_status === 'paid' ? resolvePaidTaskStatus(plan.profile) : TASK_STATUS.PENDING_PAYMENT let createdFresh = true let task: TaskRow | null try { task = await createDeliveryTask({ orderId: order.id, orderItemId: item.id, unitIndex, provider: order.provider, platform: order.platform, shopId: order.shop_id, shopName: order.shop_name, platformOrderId: order.platform_order_id, taskNo: createRandomId('DT'), profileId: plan.profileId, executorKey: plan.executorKey, taskStatus: initialStatus, deliveryStatus: 'pending', resultCode: '', resultMessage: '', claimToken: '', claimExpiresAt: null, automationMode: plan.autoDispatch ? 'automatic' : 'manual', requiresClaim: plan.requiresClaim, userActionStatus: plan.requiresClaim ? 'pending_claim' : 'not_required', attemptCount: 0, lastError: '', contextJson: JSON.stringify(plan.context), createdAt, updatedAt: createdAt, }) } catch (error) { const concurrentTasks = await listTasks(order.id) task = concurrentTasks.find( (candidate) => (allowInactiveTaskRetry ? !isInactiveCarrierTaskStatus(candidate.task_status) : true) && buildTaskUnitKey(candidate) === buildOrderItemUnitKey(item.id, unitIndex), ) || null if (!task) { writeWarnLog('[delivery-task-service]', '履约任务创建执行异常', { ...buildOrderLogContext(order), ...buildOrderItemLogContext(item), unitIndex, profileId: plan.profileId, profileKey: plan.profileKey, executorKey: plan.executorKey, error: error instanceof Error ? error.message : String(error), }) throw error } createdFresh = false writeInfoLog('[delivery-task-service]', '并发同步已创建相同履约任务,本次直接复用', { ...buildOrderLogContext(order), ...buildOrderItemLogContext(item), unitIndex, taskId: task.id, taskNo: task.task_no, }) } if (task) { const deliveryTask = { ...task, skuCode: item.sku_code, skuName: item.sku_name, } tasks.push(deliveryTask) existingTaskKeys.add(buildTaskUnitKey(task)) if (createdFresh) { await appendTaskEvent( task.id, 'fulfillment_task_planned', { source: 'order_task_reconciliation', orderId: order.id, platformOrderId: order.platform_order_id, orderItemId: item.id, unitIndex, skuCode: item.sku_code, skuName: item.sku_name, profileId: plan.profileId, profileKey: plan.profileKey, executorKey: plan.executorKey, automationMode: plan.autoDispatch ? 'automatic' : 'manual', initialStatus, }, createdAt, ) } } else { writeWarnLog('[delivery-task-service]', '履约任务创建未返回记录', { ...buildOrderLogContext(order), ...buildOrderItemLogContext(item), unitIndex, profileId: plan.profileId, profileKey: plan.profileKey, executorKey: plan.executorKey, }) } } } writeInfoLog('[delivery-task-service]', '订单履约任务完整性核对完成', { ...buildOrderLogContext(order), existingTaskCount: existingTasks.length, createdTaskCount: tasks.length - existingTasks.length, totalTaskCount: tasks.length, expectedTaskCount: orderItems.reduce( (total, item) => total + Math.max(1, Number(item.quantity || 1)), 0, ), }) if (order.pay_status !== 'paid') { return tasks } const preparedTasks = await Promise.all( tasks.map((task) => preparePaidTaskWithDiagnostics(order, task, runtimeDeps, writeWarnLog)), ) return preparedTasks.filter(isTaskRow) } function executorKeyToFulfillmentChannel(executorKey: string): FulfillmentChannel | null { if (executorKey === FULFILLMENT_EXECUTOR_KEYS.AFFILIATE_DASH) return 'affiliate_dash' if (executorKey === FULFILLMENT_EXECUTOR_KEYS.MANUAL_DISPATCH) return 'manual' return null } async function preparePaidTaskWithDiagnostics( order: OrderRow, task: TaskRow, deps: FulfillmentPrepareDeps, writeWarnLog: typeof logWarn, ) { try { const prepared = await preparePaidFulfillmentTask(task, deps) if (!prepared) { writeWarnLog('[delivery-task-service]', '已支付履约任务准备未返回任务记录', { ...buildOrderLogContext(order), taskId: task.id, taskNo: task.task_no, orderItemId: task.order_item_id, unitIndex: task.unit_index, executorKey: task.executor_key, taskStatus: task.task_status, }) } return prepared } catch (error) { writeWarnLog('[delivery-task-service]', '已支付履约任务准备执行异常', { ...buildOrderLogContext(order), taskId: task.id, taskNo: task.task_no, orderItemId: task.order_item_id, unitIndex: task.unit_index, executorKey: task.executor_key, taskStatus: task.task_status, error: error instanceof Error ? error.message : String(error), }) throw error } } function resolvePaidTaskStatus(profile: FulfillmentBindingLike | null | undefined): string { return resolveInitialPaidTaskStatus(profile) } function isTaskRow(task: TaskRow | DeliveryTaskRow | null | undefined): task is TaskRow { return Boolean(task && Number(task.id || 0) > 0) } function buildTaskUnitKey(task: Pick) { return buildOrderItemUnitKey(task.order_item_id, task.unit_index) } function buildOrderItemUnitKey(orderItemId: number | string, unitIndex: number | string) { return `${Number(orderItemId)}:${Math.max(1, Number(unitIndex || 1))}` } function buildOrderLogContext(order: OrderRow) { return { orderId: order.id, provider: order.provider, platform: order.platform, shopId: order.shop_id, platformOrderId: order.platform_order_id, payStatus: order.pay_status, } } function buildOrderItemLogContext(item: OrderItemRow) { return { orderItemId: item.id, skuCode: item.sku_code, skuName: item.sku_name, quantity: item.quantity, } } function buildPlanningDiagnostic(item: OrderItemRow) { const snapshot = parseJsonObject(item.item_snapshot_json) const route = parseJsonObject(snapshot.fulfillmentRoute) const cloudtentacles = parseJsonObject(snapshot.cloudtentacles) const kuaishouFeifei = parseJsonObject(snapshot.kuaishouFeifei) const affiliateDash = parseJsonObject(snapshot.affiliateDash) const kuaishouSendCode = parseJsonObject(snapshot.kuaishouSendCode) return { snapshotIsConfigured: Boolean(snapshot.isConfigured), selectedExecutorKey: String(route.selectedExecutorKey || '').trim(), routeReason: String(route.reason || '').trim(), candidateAvailability: route.candidates || route.candidateAvailability || [], sourceItemTitle: String(kuaishouSendCode.itemTitle || '').trim(), sourceSkuNick: String(kuaishouSendCode.skuNick || '').trim(), cloudSkuId: Number(cloudtentacles.cloudSkuId || 0) || 0, cloudSkuName: String(cloudtentacles.cloudSkuName || '').trim(), cloudSourceKeys: Array.isArray(cloudtentacles.cloudSourceKeys) ? cloudtentacles.cloudSourceKeys : [], feifeiProductCode: String(kuaishouFeifei.productCode || '').trim(), affiliateDashSku: String(affiliateDash.sku || '').trim(), } } function parseJsonObject(value: unknown): JsonObject { if (value && typeof value === 'object' && !Array.isArray(value)) { return value as JsonObject } try { const parsed = JSON.parse(String(value || '{}')) return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as JsonObject) : {} } catch { return {} } }