refactor(fulfillment): 移除飞飞/乐玩通道,仅保留 affiliate_dash

新单只路由 affiliate_dash(未命中走 manual_dispatch);历史飞飞/乐玩任务保留只读 stub,避免订单重同步失败。同步清理 cloudtentacles 调度、Admin 写路径与相关 env。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
yml2213
2026-09-14 17:29:47 +08:00
co-authored by Cursor
parent 0cbcb89dd8
commit b3aad0044c
155 changed files with 661 additions and 23351 deletions
@@ -1,358 +0,0 @@
import { getCloudtentaclesAsset } from '../platforms/cloudtentacles/catalog-service.js'
import type { JsonObject } from '../../types/json.js'
import {
normalizeCloudtentaclesDeviceId,
normalizeCloudtentaclesDeviceType,
} from '../platforms/cloudtentacles/defaults.js'
import {
getCloudtentaclesSourceByKey,
listCloudtentaclesSources,
} from '../platforms/cloudtentacles/source-config-service.js'
import { getCloudtentaclesSessionStateByKey } from '../platforms/cloudtentacles/session-state-service.js'
import {
notifyCloudtentaclesAssetLow,
notifyCloudtentaclesAuthExpired,
} from '../notification/domain-notifications.js'
import { retryCloudtentaclesPendingReturns } from './cloudtentacles-number-recycle-service.js'
type CloudtentaclesHealthAccountResult = {
sourceKey: string
label: string
enabled: boolean
hasToken: boolean
ok: boolean
skipped: boolean
status: string
message: string
asset: number | null
threshold: number
checkedAt: string
}
export async function runCloudtentaclesHealthJob(job: JsonObject) {
const accounts = resolveCloudtentaclesHealthAccounts(job)
const cooldownSeconds = Number(job.cooldownSeconds || 1800)
const [results, returnRetryResult] = await Promise.all([
Promise.all(
accounts.map((account) => runCloudtentaclesHealthAccount(account, cooldownSeconds)),
),
retryCloudtentaclesPendingReturns(),
])
const checkedResults = results.filter((item) => !item.skipped)
const failedResults = checkedResults.filter((item) => item.ok === false)
const lowAssetResults = checkedResults.filter((item) => item.status === 'asset_low')
const okResults = checkedResults.filter((item) => item.status === 'ok')
const assetValues = checkedResults
.map((item) => item.asset)
.filter((value): value is number => typeof value === 'number' && Number.isFinite(value))
const summary = buildCloudtentaclesHealthSummary(results)
const status = resolveCloudtentaclesHealthStatus(results)
const orphanTaskCount = Number(returnRetryResult?.orphanCount || 0)
const recycleNote =
orphanTaskCount > 0
? `,另有 ${orphanTaskCount} 个孤儿号码待人工/上游清理`
: returnRetryResult?.returned
? ',发货后退号重试正常'
: ''
const finalMessage = `${summary}${recycleNote}`
return {
ok: failedResults.length === 0,
status,
message: finalMessage,
accountCount: results.length,
checkedCount: checkedResults.length,
okCount: okResults.length,
lowAssetCount: lowAssetResults.length,
failedCount: failedResults.length,
asset: assetValues.length > 0 ? Math.min(...assetValues) : null,
threshold:
checkedResults.length > 0
? Math.max(...checkedResults.map((item) => Number(item.threshold || 0)))
: normalizeNonNegativeInteger(job.config?.assetThreshold, 500),
accounts: results,
pendingReturnRetry: returnRetryResult,
}
}
async function runCloudtentaclesHealthAccount(
account: JsonObject,
cooldownSeconds: number,
): Promise<CloudtentaclesHealthAccountResult> {
const sourceKey = String(account.sourceKey || '').trim() || 'default'
const source = getCloudtentaclesSourceByKey(sourceKey)
const session: JsonObject = getCloudtentaclesSessionStateByKey(sourceKey) || {}
const label =
String(account.label || source?.label || source?.username || sourceKey).trim() || sourceKey
const threshold = normalizeNonNegativeInteger(account.assetThreshold, 500)
const checkedAt = new Date().toISOString()
if (!source) {
return {
sourceKey,
label,
enabled: account.enabled !== false,
hasToken: false,
ok: false,
skipped: false,
status: 'source_missing',
message: `账号 ${label} 不存在`,
asset: null,
threshold,
checkedAt,
}
}
if (account.enabled === false || source.enabled === false) {
return {
sourceKey,
label,
enabled: false,
hasToken: Boolean(String(session.token || '').trim()),
ok: true,
skipped: true,
status: 'disabled',
message: `账号 ${label} 已停用`,
asset: null,
threshold,
checkedAt,
}
}
const token = String(session.token || '').trim()
if (!token) {
await notifyCloudtentaclesAuthExpired({
pathname: '/user/get_asset',
errorCode: 'cloudtentacles_token_missing',
message: `账号 ${label} 没有可用 token,请到后台重新登录`,
cooldownSeconds,
sourceKey,
accountLabel: label,
})
return {
ok: false,
status: 'auth_missing',
message: `账号 ${label} token 缺失`,
asset: null,
threshold,
sourceKey,
label,
enabled: true,
hasToken: false,
skipped: false,
checkedAt,
}
}
try {
const assetResult = await getCloudtentaclesAsset({
sourceKey,
accountLabel: label,
baseUrl: String(session.baseUrl || source.baseUrl || '').trim() || 'https://123.207.217.176',
token,
deviceId: normalizeCloudtentaclesDeviceId(session.deviceId || source.deviceId),
deviceType: normalizeCloudtentaclesDeviceType(session.deviceType ?? source.deviceType),
})
const asset = Number(assetResult.asset || 0)
if (asset < threshold) {
await notifyCloudtentaclesAssetLow({
asset,
threshold,
cooldownSeconds,
sourceKey,
accountLabel: label,
})
}
return {
ok: true,
status: asset < threshold ? 'asset_low' : 'ok',
message:
asset < threshold
? `账号 ${label} 余额 ${asset} 低于阈值 ${threshold}`
: `账号 ${label} 余额 ${asset} 正常`,
asset,
threshold,
sourceKey,
label,
enabled: true,
hasToken: true,
skipped: false,
checkedAt,
}
} catch (error) {
const message = error instanceof Error ? error.message : String(error || '余额检查失败')
return {
ok: false,
status: 'failed',
message: `账号 ${label} 检查失败:${message}`,
asset: null,
threshold,
sourceKey,
label,
enabled: true,
hasToken: true,
skipped: false,
checkedAt,
}
}
}
function resolveCloudtentaclesHealthAccounts(job: JsonObject) {
const config = isPlainObject(job.config) ? job.config : {}
const rawAccounts = Array.isArray(config.accounts) ? config.accounts : []
const defaultThreshold = normalizeNonNegativeInteger(config.assetThreshold, 500)
const sourceConfig = listCloudtentaclesSources()
const sources = Array.isArray(sourceConfig.sources) ? sourceConfig.sources : []
const sourceMap = new Map(
sources
.map((source) => [String(source.key || '').trim(), source] as const)
.filter(([sourceKey]) => Boolean(sourceKey)),
)
const configuredMap = new Map(
rawAccounts
.map((item) => normalizeCloudtentaclesHealthAccount(item, defaultThreshold, sourceMap))
.filter(Boolean)
.map((item) => [String(item!.sourceKey || '').trim(), item!] as const)
.filter(([sourceKey]) => Boolean(sourceKey)),
)
// 以平台配置中的全部账号为准,合并任务里已保存的阈值/开关;新增账号自动纳入监控。
const mergedAccounts = sources
.map((source) => {
const sourceKey = String(source.key || '').trim()
if (!sourceKey) {
return null
}
const configured = configuredMap.get(sourceKey)
return {
sourceKey,
label:
String(configured?.label || source.label || source.username || sourceKey).trim() ||
sourceKey,
enabled: configured ? configured.enabled !== false : source.enabled !== false,
assetThreshold: normalizeNonNegativeInteger(configured?.assetThreshold, defaultThreshold),
}
})
.filter(Boolean) as JsonObject[]
// 保留已配置但源账号已删除的项,便于报 source_missing。
for (const [sourceKey, configured] of configuredMap.entries()) {
if (sourceMap.has(sourceKey)) {
continue
}
mergedAccounts.push(configured)
}
if (mergedAccounts.length > 0) {
return dedupeAccounts(mergedAccounts)
}
// 兼容旧配置:没有任何平台账号时,仍检查历史 default。
const fallback = configuredMap.get('default')
return [
{
sourceKey: 'default',
label: String(fallback?.label || '默认账号').trim() || '默认账号',
enabled: fallback ? fallback.enabled !== false : true,
assetThreshold: normalizeNonNegativeInteger(fallback?.assetThreshold, defaultThreshold),
},
]
}
function normalizeCloudtentaclesHealthAccount(
rawValue: unknown,
defaultThreshold: number,
sourceMap: Map<string, JsonObject>,
) {
const source = isPlainObject(rawValue) ? rawValue : {}
const sourceKey = String(source.sourceKey || source.key || '').trim()
if (!sourceKey) {
return null
}
const sourceConfig = sourceMap.get(sourceKey) || {}
return {
sourceKey,
label: String(source.label || sourceConfig.label || sourceConfig.username || sourceKey).trim(),
enabled: source.enabled !== false,
assetThreshold: normalizeNonNegativeInteger(
source.assetThreshold ?? source.threshold,
defaultThreshold,
),
}
}
function dedupeAccounts(accounts: JsonObject[]) {
const seen = new Set<string>()
return accounts.filter((item) => {
const sourceKey = String(item.sourceKey || '').trim()
if (!sourceKey || seen.has(sourceKey)) {
return false
}
seen.add(sourceKey)
return true
})
}
function buildCloudtentaclesHealthSummary(results: CloudtentaclesHealthAccountResult[]) {
if (results.length === 0) {
return '未配置 kuaishou-lewan 监控账号'
}
const checkedResults = results.filter((item) => !item.skipped)
const lowAssetResults = checkedResults.filter((item) => item.status === 'asset_low')
const failedResults = checkedResults.filter((item) => item.ok === false)
const disabledCount = results.length - checkedResults.length
if (checkedResults.length === 0) {
return `所有账号均已停用,共 ${disabledCount} 个`
}
if (failedResults.length > 0 || lowAssetResults.length > 0) {
return [
`检查 ${checkedResults.length} 个账号`,
lowAssetResults.length > 0 ? `余额预警 ${lowAssetResults.length} 个` : '',
failedResults.length > 0 ? `异常 ${failedResults.length} 个` : '',
disabledCount > 0 ? `停用 ${disabledCount} 个` : '',
]
.filter(Boolean)
.join(',')
}
return `检查 ${checkedResults.length} 个账号,余额均正常${disabledCount > 0 ? `,停用 ${disabledCount} 个` : ''}`
}
function resolveCloudtentaclesHealthStatus(results: CloudtentaclesHealthAccountResult[]) {
const checkedResults = results.filter((item) => !item.skipped)
if (checkedResults.length === 0) {
return 'skipped'
}
if (
checkedResults.some((item) =>
['auth_missing', 'source_missing', 'failed'].includes(item.status),
)
) {
return 'failed'
}
if (checkedResults.some((item) => item.status === 'asset_low')) {
return 'asset_low'
}
return 'ok'
}
function normalizeNonNegativeInteger(value: unknown, fallback: number) {
const parsed = Number(value)
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback
}
function isPlainObject(value: unknown): value is JsonObject {
return Boolean(value) && typeof value === 'object' && !Array.isArray(value)
}
@@ -1,281 +0,0 @@
import { createTaskEvent } from '../../repositories/task-event-repo.js'
import { updateTask } from '../../repositories/task-repo.js'
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 { nowIso } from '../../utils/time.js'
import type { TaskRow } from '../../types/repository/rows.js'
import {
isKuaishouCloudBindingMutationFrozen,
isKuaishouCloudTask,
maskPhone,
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
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('没有权限') ||
message.includes('不存在') ||
message.includes('已释放') ||
message.includes('已回收')
)
}
/** 清空任务侧已失效的旧绑定(号码已退/已回收),任务下次领取时重新取号 */
async function clearStaleKuaishouCloudBinding(task: TaskRow, errorMessage: string) {
const now = nowIso()
const taskContext = parseJsonObject(task.context_json)
const flow = normalizeKuaishouCloudFlow(taskContext.kuaishouCloudFulfillment)
const oldVnId = flow.binding.vnId
const oldVnPhone = flow.binding.vnPhone
const nextContext = {
...taskContext,
kuaishouCloudFulfillment: {
...flow,
binding: {
...flow.binding,
prepareStatus: 'pending',
vnId: 0,
vnPhone: '',
bindUrl: '',
bindPreparedAt: null,
bindExpiresAt: null,
bindProbeAt: now,
bindProbeStatus: 'stale_cleared',
bindProbeMessage: errorMessage,
roleName: '',
roleId: '',
},
role: {
status: 'pending',
name: '',
rid: '',
refreshedAt: now,
errorMessage: '旧绑定号码已失效,已清空绑定,将重新取号',
rawInfo: null,
},
},
}
await updateTask(task.id, {
task_status: TASK_STATUS.PENDING_BINDING_PREPARE,
role_id: '',
role_name: '',
last_error: `旧绑定号码已失效(${errorMessage}),已清空绑定`,
context_json: JSON.stringify(nextContext),
updated_at: now,
})
await createTaskEvent(
task.id,
'kuaishou_cloud_stale_binding_cleared',
{
source: 'scheduler_stale_number_recycle',
oldVnId,
oldVnPhoneMasked: maskPhone(oldVnPhone),
errorMessage,
actor: { source: 'system' },
},
now,
)
}
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 ft.task_status NOT IN ('redeeming', 'dispatched_pending_return', 'completed', 'redeemed', 'expired', 'closed')
AND COALESCE(ft.context_json #>> '{kuaishouCloudFulfillment,dispatch,status}', '') <> 'success'
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)
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
} catch (error) {
const message = error instanceof Error ? error.message : String(error || '')
if (isOrphanNumberError(error)) {
result.orphanCount += 1
result.orphanTaskIds.push(taskId)
try {
await clearStaleKuaishouCloudBinding(task, message)
logWarn(
'[scheduler/cloudtentacles-number-recycle]',
'旧绑定号码无操作权限(号码已退/已回收),已清空任务绑定,等待重新取号',
{ taskId, error: message },
)
} catch (clearError) {
result.failed += 1
logWarn('[scheduler/cloudtentacles-number-recycle]', '清空旧绑定失败', {
taskId,
error: clearError instanceof Error ? clearError.message : String(clearError || ''),
})
}
continue
}
result.failed += 1
logWarn('[scheduler/cloudtentacles-number-recycle]', '过期绑定资源回收失败', {
taskId,
error: message,
})
}
}
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)
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
} catch (error) {
const message = error instanceof Error ? error.message : String(error || '')
if (isOrphanNumberError(error)) {
result.orphanCount += 1
result.orphanTaskIds.push(taskId)
logWarn(
'[scheduler/cloudtentacles-number-recycle]',
'发货收尾退号无操作权限(号码已退/已回收),请人工确认任务收尾',
{ taskId, error: message },
)
continue
}
result.failed += 1
logWarn('[scheduler/cloudtentacles-number-recycle]', '发货后退号重试失败', {
taskId,
error: message,
})
}
}
if (result.scanned > 0) {
logInfo('[scheduler/cloudtentacles-number-recycle]', '发货后退号重试完成', {
...result,
taskIds: rows.rows.map((row) => Number(row.id || 0)),
})
}
return result
}
@@ -1,9 +1,9 @@
import test from 'node:test'
import assert from 'node:assert/strict'
import test from 'node:test'
import { normalizeScheduledJobsConfig } from './config-service.js'
test('normalizeScheduledJobsConfig 将旧版 cloudtentacles 阈值迁移为默认账号配置', () => {
test('normalizeScheduledJobsConfig drops retired cloudtentacles_health jobs', () => {
const config = normalizeScheduledJobsConfig({
enabled: true,
jobs: [
@@ -11,97 +11,34 @@ test('normalizeScheduledJobsConfig 将旧版 cloudtentacles 阈值迁移为默
id: 'cloudtentacles-health',
type: 'cloudtentacles_health',
enabled: true,
intervalSeconds: 18000,
cooldownSeconds: 1800,
config: {
assetThreshold: 100,
},
intervalSeconds: 300,
config: { assetThreshold: 100 },
},
],
})
assert.deepEqual(config.jobs[0], {
id: 'cloudtentacles-health',
type: 'cloudtentacles_health',
enabled: true,
intervalSeconds: 18000,
cooldownSeconds: 1800,
config: {
assetThreshold: 100,
accounts: [
{
sourceKey: 'default',
label: '',
enabled: true,
assetThreshold: 100,
},
],
},
})
})
test('normalizeScheduledJobsConfig 支持多账号去重并保留 0 阈值', () => {
const config = normalizeScheduledJobsConfig({
jobs: [
{
id: 'cloudtentacles-health',
type: 'cloudtentacles_health',
id: 'work-order-timeout',
type: 'work_order_timeout',
enabled: true,
intervalSeconds: 30,
cooldownSeconds: 10,
config: {
assetThreshold: 500,
accounts: [
{
sourceKey: 'default',
label: '默认账号',
enabled: true,
assetThreshold: 0,
},
{
sourceKey: 'default',
label: '重复账号',
enabled: false,
assetThreshold: 999,
},
{
sourceKey: 'account2',
enabled: false,
threshold: 200,
},
],
},
intervalSeconds: 60,
config: { scanLimit: 50 },
},
],
})
assert.deepEqual(config.jobs[0].config.accounts, [
{
sourceKey: 'default',
label: '默认账号',
enabled: true,
assetThreshold: 0,
},
{
sourceKey: 'account2',
label: '',
enabled: false,
assetThreshold: 200,
},
])
assert.equal(config.jobs[0].intervalSeconds, 60)
assert.equal(config.jobs[0].cooldownSeconds, 60)
assert.equal(
config.jobs.some((item) => item.type === 'cloudtentacles_health'),
false,
)
assert.equal(
config.jobs.some((item) => item.id === 'work-order-timeout'),
true,
)
})
test('normalizeScheduledJobsConfig 自动补齐数据保留清理任务', () => {
test('normalizeScheduledJobsConfig ensures default jobs exist', () => {
const config = normalizeScheduledJobsConfig({ enabled: true, jobs: [] })
const retentionJob = config.jobs.find((job) => job.type === 'data_retention')
assert.deepEqual(retentionJob, {
id: 'data-retention',
type: 'data_retention',
enabled: true,
intervalSeconds: 86400,
config: {},
})
const ids = config.jobs.map((item) => item.id)
assert.ok(ids.includes('work-order-timeout'))
assert.ok(ids.includes('deposit-unfreeze'))
assert.ok(ids.includes('data-retention'))
assert.ok(ids.includes('pdd_order_sync'))
})
@@ -6,7 +6,6 @@ import { PROJECT_ROOT } from '../../config/runtime.js'
import { readAppConfigEntry, saveAppConfigEntry } from '../config/app-config-store.js'
const SCHEDULED_JOBS_FILE_PATH = path.join(PROJECT_ROOT, 'data', 'scheduled-jobs.json')
const CLOUDTENTACLES_HEALTH_JOB_ID = 'cloudtentacles-health'
const WORK_ORDER_TIMEOUT_JOB_ID = 'work-order-timeout'
const DEPOSIT_UNFREEZE_JOB_ID = 'deposit-unfreeze'
const DATA_RETENTION_JOB_ID = 'data-retention'
@@ -34,14 +33,6 @@ export function saveScheduledJobsConfig(rawValue: unknown) {
})
}
export function getCloudtentaclesHealthJob(config: JsonObject = getScheduledJobsConfig()) {
return (
(Array.isArray(config.jobs) ? config.jobs : []).find(
(item) => String(item.id || '').trim() === CLOUDTENTACLES_HEALTH_JOB_ID,
) || createDefaultCloudtentaclesHealthJob()
)
}
export function getWorkOrderTimeoutJob(config: JsonObject = getScheduledJobsConfig()) {
return (
(Array.isArray(config.jobs) ? config.jobs : []).find(
@@ -80,15 +71,11 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
const jobs = rawJobs
.map((item) => normalizeScheduledJob(item))
.filter((item): item is NonNullable<ReturnType<typeof normalizeScheduledJob>> => Boolean(item))
const hasCloudtentaclesHealth = jobs.some((item) => item.id === CLOUDTENTACLES_HEALTH_JOB_ID)
const hasWorkOrderTimeout = jobs.some((item) => item.id === WORK_ORDER_TIMEOUT_JOB_ID)
const hasDepositUnfreeze = jobs.some((item) => item.id === DEPOSIT_UNFREEZE_JOB_ID)
const hasDataRetention = jobs.some((item) => item.id === DATA_RETENTION_JOB_ID)
const hasPddOrderSync = jobs.some((item) => item.id === PDD_ORDER_SYNC_JOB_ID)
if (!hasCloudtentaclesHealth) {
jobs.push(createDefaultCloudtentaclesHealthJob())
}
if (!hasWorkOrderTimeout) {
jobs.push(createDefaultWorkOrderTimeoutJob())
}
@@ -114,8 +101,9 @@ function normalizeScheduledJob(rawValue: unknown) {
}
const type = String(rawValue.type || '').trim()
// 已下线的 cloudtentacles_health 配置直接丢弃
if (type === 'cloudtentacles_health') {
return normalizeCloudtentaclesHealthJob(rawValue)
return null
}
if (type === 'work_order_timeout') {
return normalizeWorkOrderTimeoutJob(rawValue)
@@ -185,78 +173,10 @@ function normalizeWorkOrderTimeoutJob(rawValue: JsonObject) {
}
}
function normalizeCloudtentaclesHealthJob(rawValue: JsonObject) {
const config = isPlainObject(rawValue.config) ? rawValue.config : {}
const defaultAssetThreshold = normalizeRangeInteger(config.assetThreshold, 500, 0, 999999)
return {
id: CLOUDTENTACLES_HEALTH_JOB_ID,
type: 'cloudtentacles_health',
enabled: rawValue.enabled === true,
intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 86400),
cooldownSeconds: normalizeRangeInteger(rawValue.cooldownSeconds, 1800, 60, 86400),
config: {
assetThreshold: defaultAssetThreshold,
accounts: normalizeCloudtentaclesHealthAccounts(config, defaultAssetThreshold),
},
}
}
function normalizeCloudtentaclesHealthAccounts(config: JsonObject, defaultAssetThreshold: number) {
const rawAccounts = Array.isArray(config.accounts)
? config.accounts
: Array.isArray(config.sourceKeys)
? config.sourceKeys.map((sourceKey) => ({ sourceKey }))
: []
const accounts = rawAccounts
.map((item) => normalizeCloudtentaclesHealthAccount(item, defaultAssetThreshold))
.filter(Boolean) as ReturnType<typeof normalizeCloudtentaclesHealthAccount>[]
if (accounts.length === 0) {
accounts.push(
normalizeCloudtentaclesHealthAccount(
{
sourceKey: config.sourceKey || 'default',
assetThreshold: defaultAssetThreshold,
},
defaultAssetThreshold,
),
)
}
const seen = new Set<string>()
return accounts.filter((item) => {
const sourceKey = String(item?.sourceKey || '').trim()
if (!sourceKey || seen.has(sourceKey)) {
return false
}
seen.add(sourceKey)
return true
})
}
function normalizeCloudtentaclesHealthAccount(rawValue: unknown, defaultAssetThreshold: number) {
const source = isPlainObject(rawValue) ? rawValue : {}
const sourceKey = String(source.sourceKey || source.key || '').trim() || 'default'
return {
sourceKey,
label: String(source.label || '').trim(),
enabled: source.enabled !== false,
assetThreshold: normalizeRangeInteger(
source.assetThreshold ?? source.threshold,
defaultAssetThreshold,
0,
999999,
),
}
}
function createDefaultScheduledJobsConfig() {
return {
enabled: true,
jobs: [
createDefaultCloudtentaclesHealthJob(),
createDefaultWorkOrderTimeoutJob(),
createDefaultDepositUnfreezeJob(),
createDefaultDataRetentionJob(),
@@ -311,27 +231,6 @@ function createDefaultWorkOrderTimeoutJob() {
}
}
function createDefaultCloudtentaclesHealthJob() {
return {
id: CLOUDTENTACLES_HEALTH_JOB_ID,
type: 'cloudtentacles_health',
enabled: false,
intervalSeconds: 300,
cooldownSeconds: 1800,
config: {
assetThreshold: 500,
accounts: [
{
sourceKey: 'default',
label: '默认账号',
enabled: true,
assetThreshold: 500,
},
],
},
}
}
function normalizeRangeInteger(value: unknown, fallback: number, min: number, max: number) {
const parsed = Number(value)
if (!Number.isInteger(parsed)) {
@@ -2,7 +2,6 @@ import { logError, logInfo, logWarn } from '../../utils/logger.js'
import { createHttpError } from '../../utils/http.js'
import type { JsonObject } from '../../types/json.js'
import { getScheduledJobById, getScheduledJobsConfig } from './config-service.js'
import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js'
import { runDepositUnfreezeJob } from './deposit-unfreeze-job.js'
import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js'
import { runDataRetentionJob } from './data-retention-job.js'
@@ -186,9 +185,6 @@ async function runJob(job: JsonObject, { manual = false }: { manual?: boolean }
}
function dispatchJob(job: JsonObject) {
if (job.type === 'cloudtentacles_health') {
return runCloudtentaclesHealthJob(job)
}
if (job.type === 'work_order_timeout') {
return runWorkOrderTimeoutJob(job)
}