import { buildScheduledTaskVerifiedHtmlUrls, deliveryTextPromisesPublicHtml, executeScheduledTask, formatScheduledTaskDeliveryMessage, looksLikeScheduledTaskNonDelivery, reconcileStuckStaticPageDeliveryContracts, resendScheduledTaskWechatForReadyPage, } from './scheduled-task-executor.mjs'; export function startScheduledTaskWorker({ scheduledTaskService, scheduleService = null, executeTask = executeScheduledTask, userAuth = null, tkmindProxy = null, sessionSnapshotService = null, notificationDispatcher = null, pool = null, h5Root = null, logger = console, intervalMs = Number(process.env.H5_SCHEDULED_TASK_SCAN_INTERVAL_MS ?? 30_000), maxAttempts = Number(process.env.H5_SCHEDULED_TASK_MAX_ATTEMPTS ?? 3), executionTimeoutMs = Number(process.env.H5_SCHEDULED_TASK_EXECUTION_TIMEOUT_MS ?? 15 * 60 * 1000), runOnStart = true, } = {}) { const sendScheduleNotification = notificationDispatcher?.sendScheduleNotification?.bind(notificationDispatcher) ?? null; const canExecute = scheduledTaskService && typeof executeTask === 'function' && userAuth && tkmindProxy; if (!canExecute) { return { stop() {} }; } let stopped = false; let running = false; const deliverTaskResult = async (task, deliveryText, { readyPaths = [] } = {}) => { const text = formatScheduledTaskDeliveryMessage(task, deliveryText); const notifyChannel = task.notifyChannel ?? 'both'; const verifiedHtmlUrls = buildScheduledTaskVerifiedHtmlUrls(task.userId, readyPaths); const promisesHtml = deliveryTextPromisesPublicHtml(deliveryText); let wechatDelivery = null; if (scheduleService?.createUserNotification && (notifyChannel === 'web' || notifyChannel === 'both')) { await scheduleService.createUserNotification({ userId: task.userId, channel: 'web', notificationType: 'scheduled_task_result', title: `定时任务完成:${task.title}`, body: text, data: { taskId: task.id, recurrence: task.recurrence, }, }).catch((err) => { logger.warn?.('Scheduled task web notification failed:', err); }); } if ( (notifyChannel === 'wechat' || notifyChannel === 'both') && typeof sendScheduleNotification === 'function' ) { if (promisesHtml && verifiedHtmlUrls.length === 0) { logger.warn?.('[ScheduledTask] skip wechat until public html is ready', { taskId: task.id, userId: task.userId, }); } else { const sent = await sendScheduleNotification({ userId: task.userId, text, verifiedHtmlUrls, }).catch((err) => { logger.warn?.('Scheduled task wechat notification failed:', err); return false; }); if (sent && verifiedHtmlUrls.length > 0) { wechatDelivery = { sentAt: Date.now(), relativePaths: readyPaths.map((value) => String(value ?? '').trim()).filter(Boolean), source: 'scheduled_task_worker', }; } } } return { wechatDelivery }; }; const runOnce = async () => { if (running || stopped) return; running = true; try { if (pool && h5Root) { const reconciled = await reconcileStuckStaticPageDeliveryContracts({ pool, h5Root, logger, }).catch((err) => { logger.warn?.('Scheduled task delivery reconcile failed:', err); return []; }); for (const item of reconciled) { await resendScheduledTaskWechatForReadyPage({ pool, userId: item.userId, relativePath: item.relativePath, notificationDispatcher, logger, }).catch((err) => { logger.warn?.('Scheduled task reconcile wechat resend failed:', err); }); } } const dueTasks = await scheduledTaskService.listDueTasks({ limit: 10 }); for (const candidate of dueTasks) { const task = await scheduledTaskService.lockTask(candidate.id); if (!task) continue; try { await scheduledTaskService.markTaskRunning(task); const result = await executeTask(task, { userAuth, tkmindProxy, sessionSnapshotService, pool, h5Root, timeoutMs: executionTimeoutMs, logger, }); if (looksLikeScheduledTaskNonDelivery(result.deliveryText, { readyPaths: result.readyPaths, })) { const err = new Error('定时任务未产出可交付结果'); err.code = 'SCHEDULED_TASK_NON_DELIVERY'; throw err; } const deliveryMeta = await deliverTaskResult(task, result.deliveryText, { readyPaths: result.readyPaths, }); await scheduledTaskService.markTaskSucceeded(task, { result: { deliveryText: result.deliveryText, ...(deliveryMeta.wechatDelivery ? { wechatDelivery: deliveryMeta.wechatDelivery } : {}), }, deliveryText: result.deliveryText, sessionId: result.sessionId, requestId: result.requestId, }); } catch (err) { logger.warn?.('Scheduled task execution failed:', err); const failedTask = await scheduledTaskService.markTaskFailed(task, err, { maxAttempts, }); if (failedTask.status === 'failed' && scheduleService?.createUserNotification) { await scheduleService.createUserNotification({ userId: task.userId, channel: 'web', notificationType: 'scheduled_task_failed', title: `定时任务失败:${task.title}`, body: String(failedTask.lastError ?? '执行失败'), data: { taskId: task.id }, }).catch(() => {}); } if ( failedTask.status === 'failed' && (task.notifyChannel === 'wechat' || task.notifyChannel === 'both') && typeof sendScheduleNotification === 'function' ) { await sendScheduleNotification({ userId: task.userId, text: `定时任务失败:${task.title}\n${failedTask.lastError ?? '执行失败'}`, }).catch(() => {}); } } } } catch (err) { logger.warn?.('Scheduled task worker failed:', err); } finally { running = false; } }; const timer = setInterval(runOnce, Math.max(1000, Number(intervalMs) || 30_000)); timer.unref?.(); if (runOnStart) void runOnce(); return { runOnce, stop() { stopped = true; clearInterval(timer); }, }; }