memind/scheduled-task-worker.mjs

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);
    },
  };
}