262 lines
8.2 KiB
TypeScript
262 lines
8.2 KiB
TypeScript
import { query as dbQuery } from '../../db/client.js'
|
|
import { TASK_STATUS } from '../../domain/task-status.js'
|
|
import { logInfo, logWarn } from '../../utils/logger.js'
|
|
import { parseJsonObject } from '../../utils/task-json.js'
|
|
import type { TaskRow } from '../../types/repository/rows.js'
|
|
import {
|
|
isKuaishouCloudBindingMutationFrozen,
|
|
isKuaishouCloudTask,
|
|
normalizeKuaishouCloudFlow,
|
|
} from '../fulfillment/kuaishou-cloud/domain.js'
|
|
import { refreshKuaishouCloudTaskBindUrl } from '../fulfillment/kuaishou-cloud/prepare-fulfillment.js'
|
|
import { returnKuaishouCloudFulfillmentTask } from '../fulfillment/kuaishou-cloud/return-fulfillment.js'
|
|
|
|
/** bindUrl 过期超过该时长才回收（避免刚过期就触发上游操作） */
|
|
const STALE_BIND_URL_GRACE_MS = 15 * 60 * 1000
|
|
/** 单轮最多处理的任务数（health 每 2.5 分钟跑一次，防止一次扫太多） */
|
|
const RECYCLE_SCAN_LIMIT = 50
|
|
/** "权限不足"类孤儿号码最多重试次数：上游判定号码不属于当前会话时重试无意义，3 次后停止打上游 */
|
|
const ORPHAN_MAX_RETRIES = 3
|
|
/** 孤儿计数超过该时长后自动重置，允许上游/账号恢复后重新尝试 */
|
|
const ORPHAN_RESET_AFTER_MS = 60 * 60 * 1000
|
|
/** 连续失败记录：taskId -> { count, lastAt } */
|
|
const recycleFailRecords = new Map<number, { count: number; lastAt: number }>()
|
|
|
|
type RecycleNumberResult = {
|
|
scanned: number
|
|
skippedFrozen: number
|
|
skippedMissingContext: number
|
|
refreshed: number
|
|
failed: number
|
|
/** 上游判定无权操作（账号重新登录换 token 后的孤儿号码），已停止重试的任务数 */
|
|
orphanCount: number
|
|
orphanTaskIds: number[]
|
|
}
|
|
|
|
/** 上游"权限不足/没有权限"类错误：号码归属当前账号会话，重试无法恢复 */
|
|
function isOrphanNumberError(error: unknown) {
|
|
const message = error instanceof Error ? error.message : String(error || '')
|
|
return message.includes('权限不足') || message.includes('没有权限')
|
|
}
|
|
|
|
function getRecycleFailCount(taskId: number) {
|
|
const record = recycleFailRecords.get(taskId)
|
|
if (!record) {
|
|
return 0
|
|
}
|
|
if (Date.now() - record.lastAt > ORPHAN_RESET_AFTER_MS) {
|
|
recycleFailRecords.delete(taskId)
|
|
return 0
|
|
}
|
|
return record.count
|
|
}
|
|
|
|
function isRecycleOrphanQuotaReached(taskId: number) {
|
|
return getRecycleFailCount(taskId) >= ORPHAN_MAX_RETRIES
|
|
}
|
|
|
|
function recordRecycleFailure(taskId: number, isOrphan: boolean) {
|
|
if (isOrphan) {
|
|
const record = recycleFailRecords.get(taskId)
|
|
recycleFailRecords.set(taskId, {
|
|
count: (record?.count || 0) + 1,
|
|
lastAt: Date.now(),
|
|
})
|
|
} else {
|
|
recycleFailRecords.delete(taskId)
|
|
}
|
|
}
|
|
|
|
function recordRecycleSuccess(taskId: number) {
|
|
recycleFailRecords.delete(taskId)
|
|
}
|
|
|
|
export async function recycleCloudtentaclesStaleNumbers(): Promise<RecycleNumberResult> {
|
|
const staleBeforeIso = new Date(Date.now() - STALE_BIND_URL_GRACE_MS).toISOString()
|
|
const result: RecycleNumberResult = {
|
|
scanned: 0,
|
|
skippedFrozen: 0,
|
|
skippedMissingContext: 0,
|
|
refreshed: 0,
|
|
failed: 0,
|
|
orphanCount: 0,
|
|
orphanTaskIds: [],
|
|
}
|
|
|
|
const rows = await dbQuery<TaskRow>(
|
|
`
|
|
SELECT *
|
|
FROM fulfillment_tasks ft
|
|
WHERE ft.executor_key = 'kuaishou_ct_assisted'
|
|
AND COALESCE(ft.context_json #>> '{kuaishouCloudFulfillment,binding,prepareStatus}', '') = 'ready'
|
|
AND COALESCE(ft.context_json #>> '{kuaishouCloudFulfillment,binding,bindExpiresAt}', '') <> ''
|
|
AND (ft.context_json #>> '{kuaishouCloudFulfillment,binding,bindExpiresAt}')::timestamptz < $1
|
|
ORDER BY ft.id DESC
|
|
LIMIT $2
|
|
`,
|
|
[staleBeforeIso, RECYCLE_SCAN_LIMIT]
|
|
)
|
|
|
|
for (const task of rows.rows) {
|
|
result.scanned += 1
|
|
const taskId = Number(task.id || 0)
|
|
|
|
// 已确认孤儿号码：停止无效重试，避免每轮继续打上游
|
|
if (isRecycleOrphanQuotaReached(taskId)) {
|
|
result.orphanCount += 1
|
|
result.orphanTaskIds.push(taskId)
|
|
continue
|
|
}
|
|
|
|
try {
|
|
if (!isKuaishouCloudTask(task)) {
|
|
result.skippedMissingContext += 1
|
|
continue
|
|
}
|
|
|
|
const flow = normalizeKuaishouCloudFlow(
|
|
parseJsonObject(task.context_json).kuaishouCloudFulfillment
|
|
)
|
|
|
|
// 已发货/兑换中：禁止换号覆盖，跳过
|
|
if (isKuaishouCloudBindingMutationFrozen(task, flow)) {
|
|
result.skippedFrozen += 1
|
|
continue
|
|
}
|
|
|
|
if (!flow.binding.vnId || !flow.binding.vnKey || !flow.binding.bindUrl) {
|
|
result.skippedMissingContext += 1
|
|
continue
|
|
}
|
|
|
|
await refreshKuaishouCloudTaskBindUrl(task, {
|
|
source: 'scheduler_stale_number_recycle',
|
|
actor: { source: 'system' },
|
|
})
|
|
result.refreshed += 1
|
|
recordRecycleSuccess(taskId)
|
|
} catch (error) {
|
|
const isOrphan = isOrphanNumberError(error)
|
|
recordRecycleFailure(taskId, isOrphan)
|
|
if (isOrphan && isRecycleOrphanQuotaReached(taskId)) {
|
|
result.orphanCount += 1
|
|
result.orphanTaskIds.push(taskId)
|
|
logWarn(
|
|
'[scheduler/cloudtentacles-number-recycle]',
|
|
'孤儿号码达到重试上限，已停止自动重试（需人工或上游处理）',
|
|
{
|
|
taskId,
|
|
error: error instanceof Error ? error.message : String(error || ''),
|
|
}
|
|
)
|
|
continue
|
|
}
|
|
result.failed += 1
|
|
logWarn('[scheduler/cloudtentacles-number-recycle]', '过期绑定资源回收失败', {
|
|
taskId,
|
|
error: error instanceof Error ? error.message : String(error || ''),
|
|
})
|
|
}
|
|
}
|
|
|
|
if (result.scanned > 0) {
|
|
logInfo('[scheduler/cloudtentacles-number-recycle]', '过期绑定资源回收完成', {
|
|
...result,
|
|
taskIds: rows.rows.map((row) => Number(row.id || 0)),
|
|
})
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
/** 重试发货后未完成的退号（DISPATCHED_PENDING_RETURN 且未退号成功） */
|
|
export async function retryCloudtentaclesPendingReturns(): Promise<{
|
|
scanned: number
|
|
returned: number
|
|
failed: number
|
|
orphanCount: number
|
|
orphanTaskIds: number[]
|
|
}> {
|
|
const result = {
|
|
scanned: 0,
|
|
returned: 0,
|
|
failed: 0,
|
|
orphanCount: 0,
|
|
orphanTaskIds: [] as number[],
|
|
}
|
|
const rows = await dbQuery<TaskRow>(
|
|
`
|
|
SELECT *
|
|
FROM fulfillment_tasks ft
|
|
WHERE ft.executor_key = 'kuaishou_ct_assisted'
|
|
AND ft.task_status = $1
|
|
AND COALESCE(ft.context_json #>> '{kuaishouCloudFulfillment,returnNumber,status}', '') <> 'success'
|
|
ORDER BY ft.id DESC
|
|
LIMIT $2
|
|
`,
|
|
[TASK_STATUS.DISPATCHED_PENDING_RETURN, RECYCLE_SCAN_LIMIT]
|
|
)
|
|
|
|
for (const task of rows.rows) {
|
|
result.scanned += 1
|
|
const taskId = Number(task.id || 0)
|
|
|
|
if (isRecycleOrphanQuotaReached(taskId)) {
|
|
result.orphanCount += 1
|
|
result.orphanTaskIds.push(taskId)
|
|
continue
|
|
}
|
|
|
|
try {
|
|
if (!isKuaishouCloudTask(task)) {
|
|
result.failed += 1
|
|
continue
|
|
}
|
|
const flow = normalizeKuaishouCloudFlow(
|
|
parseJsonObject(task.context_json).kuaishouCloudFulfillment
|
|
)
|
|
if (!flow.binding.vnId || !flow.binding.vnKey) {
|
|
result.failed += 1
|
|
continue
|
|
}
|
|
|
|
await returnKuaishouCloudFulfillmentTask(task, {
|
|
source: 'scheduler_retry_return_number',
|
|
actor: { source: 'system' },
|
|
})
|
|
result.returned += 1
|
|
recordRecycleSuccess(taskId)
|
|
} catch (error) {
|
|
const isOrphan = isOrphanNumberError(error)
|
|
recordRecycleFailure(taskId, isOrphan)
|
|
if (isOrphan && isRecycleOrphanQuotaReached(taskId)) {
|
|
result.orphanCount += 1
|
|
result.orphanTaskIds.push(taskId)
|
|
logWarn(
|
|
'[scheduler/cloudtentacles-number-recycle]',
|
|
'发货后退号的孤儿号码达到重试上限，已停止自动重试（需人工或上游处理）',
|
|
{
|
|
taskId,
|
|
error: error instanceof Error ? error.message : String(error || ''),
|
|
}
|
|
)
|
|
continue
|
|
}
|
|
result.failed += 1
|
|
logWarn('[scheduler/cloudtentacles-number-recycle]', '发货后退号重试失败', {
|
|
taskId,
|
|
error: error instanceof Error ? error.message : String(error || ''),
|
|
})
|
|
}
|
|
}
|
|
|
|
if (result.scanned > 0) {
|
|
logInfo('[scheduler/cloudtentacles-number-recycle]', '发货后退号重试完成', {
|
|
...result,
|
|
taskIds: rows.rows.map((row) => Number(row.id || 0)),
|
|
})
|
|
}
|
|
|
|
return result
|
|
}
|