Add scheduled task automation skill with worker execution pipeline.

Introduce scheduled-task-automation for H5 and WeChat, persist tasks in h5_scheduled_tasks, and run due jobs via a dedicated worker that executes agent tasks and delivers results. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-31 08:02:49 +08:00
parent 4186522c72
commit 81e63c16d3
27 changed files with 1736 additions and 2 deletions
@@ -0,0 +1,375 @@
+import crypto from 'node:crypto';
+import {
+  nextDailyRunAt,
+  nextWeeklyRunAt,
+  normalizeTimezone,
+  parseLocalDateTimeString,
+} from './schedule-time.mjs';
+
+const DEFAULT_TIMEZONE = 'Asia/Shanghai';
+const VALID_RECURRENCES = new Set(['once', 'daily', 'weekly']);
+const VALID_NOTIFY_CHANNELS = new Set(['wechat', 'web', 'both']);
+
+function nowMs() {
+  return Date.now();
+}
+
+function rowToTask(row) {
+  if (!row) return null;
+  return {
+    id: row.id,
+    userId: row.user_id,
+    title: row.title,
+    taskSpec: row.task_spec,
+    recurrence: row.recurrence,
+    hour: row.hour == null ? null : Number(row.hour),
+    minute: Number(row.minute ?? 0),
+    weekday: row.weekday == null ? null : Number(row.weekday),
+    timezone: row.timezone || DEFAULT_TIMEZONE,
+    nextRunAt: Number(row.next_run_at),
+    lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at),
+    notifyChannel: row.notify_channel || 'both',
+    status: row.status,
+    attempts: Number(row.attempts ?? 0),
+    lastError: row.last_error ?? null,
+    lastResult: parseJsonColumn(row.last_result_json),
+    sourceChannel: row.source_channel ?? 'agent',
+    sourceSessionId: row.source_session_id ?? null,
+    sourceMessageId: row.source_message_id ?? null,
+    sourceText: row.source_text ?? null,
+    createdAt: Number(row.created_at),
+    updatedAt: Number(row.updated_at),
+  };
+}
+
+function parseJsonColumn(value) {
+  if (value == null || value === '') return null;
+  if (typeof value === 'string') {
+    try { return JSON.parse(value); } catch { return null; }
+  }
+  if (typeof value === 'object') return value;
+  return null;
+}
+
+export function computeScheduledTaskNextRunAt({
+  recurrence,
+  runAtLocal = null,
+  hour = null,
+  minute = 0,
+  weekday = null,
+  timezone = DEFAULT_TIMEZONE,
+  now = Date.now(),
+} = {}) {
+  const safeRecurrence = String(recurrence ?? '').trim();
+  if (!VALID_RECURRENCES.has(safeRecurrence)) {
+    throw new Error('recurrence 无效，仅支持 once / daily / weekly');
+  }
+  const tz = normalizeTimezone(timezone);
+  if (safeRecurrence === 'once') {
+    const runAt = parseLocalDateTimeString(runAtLocal, tz);
+    if (runAt == null) throw new Error('一次性任务需要 runAtLocal（YYYY-MM-DD HH:mm）');
+    if (runAt <= now) throw new Error('一次性任务的执行时间必须在未来');
+    return runAt;
+  }
+  const safeHour = Number(hour);
+  const safeMinute = Number(minute ?? 0);
+  if (!Number.isInteger(safeHour) || safeHour < 0 || safeHour > 23) {
+    throw new Error('daily/weekly 任务需要有效 hour（0-23）');
+  }
+  if (!Number.isInteger(safeMinute) || safeMinute < 0 || safeMinute > 59) {
+    throw new Error('minute 无效（0-59）');
+  }
+  if (safeRecurrence === 'weekly') {
+    return nextWeeklyRunAt({
+      weekday,
+      hour: safeHour,
+      minute: safeMinute,
+      timezone: tz,
+      now,
+    });
+  }
+  return nextDailyRunAt({
+    hour: safeHour,
+    minute: safeMinute,
+    timezone: tz,
+    now,
+  });
+}
+
+export function createScheduledTaskService(pool, { defaultTimezone = DEFAULT_TIMEZONE, clock = { now: nowMs } } = {}) {
+  if (!pool) throw new Error('缺少数据库连接');
+
+  const createTask = async ({
+    userId,
+    title,
+    taskSpec,
+    recurrence = 'daily',
+    runAtLocal = null,
+    hour = null,
+    minute = 0,
+    weekday = null,
+    timezone = defaultTimezone,
+    notifyChannel = 'both',
+    sourceChannel = 'agent',
+    sourceSessionId = null,
+    sourceMessageId = null,
+    sourceText = null,
+  }) => {
+    if (!userId) throw new Error('缺少用户');
+    const safeTitle = String(title ?? '').trim();
+    const safeTaskSpec = String(taskSpec ?? '').trim();
+    if (!safeTaskSpec) throw new Error('缺少 taskSpec（执行内容）');
+    const safeRecurrence = String(recurrence ?? 'daily').trim();
+    if (!VALID_RECURRENCES.has(safeRecurrence)) {
+      throw new Error('recurrence 无效，仅支持 once / daily / weekly');
+    }
+    const safeNotifyChannel = String(notifyChannel ?? 'both').trim();
+    if (!VALID_NOTIFY_CHANNELS.has(safeNotifyChannel)) {
+      throw new Error('notifyChannel 无效，仅支持 wechat / web / both');
+    }
+    const tz = normalizeTimezone(timezone);
+    const nextRunAt = computeScheduledTaskNextRunAt({
+      recurrence: safeRecurrence,
+      runAtLocal,
+      hour,
+      minute,
+      weekday,
+      timezone: tz,
+      now: clock.now(),
+    });
+    const id = crypto.randomUUID();
+    const ts = clock.now();
+    await pool.query(
+      `INSERT INTO h5_scheduled_tasks
+       (id, user_id, title, task_spec, recurrence, hour, minute, weekday, timezone,
+        next_run_at, notify_channel, status, source_channel, source_session_id,
+        source_message_id, source_text, created_at, updated_at)
+       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?)`,
+      [
+        id,
+        userId,
+        safeTitle || safeTaskSpec.slice(0, 80),
+        safeTaskSpec,
+        safeRecurrence,
+        hour == null ? null : Number(hour),
+        Number(minute ?? 0),
+        weekday == null ? null : Number(weekday),
+        tz,
+        nextRunAt,
+        safeNotifyChannel,
+        sourceChannel,
+        sourceSessionId,
+        sourceMessageId,
+        sourceText,
+        ts,
+        ts,
+      ],
+    );
+    const [rows] = await pool.query(
+      `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`,
+      [id],
+    );
+    return rowToTask(rows[0]);
+  };
+
+  const listTasks = async ({
+    userId,
+    status = 'active',
+    limit = 20,
+  } = {}) => {
+    if (!userId) throw new Error('缺少用户');
+    const clauses = ['user_id = ?'];
+    const params = [userId];
+    if (status && status !== 'all') {
+      const statuses = Array.isArray(status) ? status : [status];
+      clauses.push(`status IN (${statuses.map(() => '?').join(', ')})`);
+      params.push(...statuses);
+    }
+    params.push(Math.max(1, Math.min(200, Number(limit) || 20)));
+    const [rows] = await pool.query(
+      `SELECT *
+       FROM h5_scheduled_tasks
+       WHERE ${clauses.join(' AND ')}
+       ORDER BY next_run_at ASC, created_at DESC
+       LIMIT ?`,
+      params,
+    );
+    return rows.map(rowToTask);
+  };
+
+  const cancelTask = async ({
+    userId,
+    taskId = null,
+    titleMatch = null,
+  } = {}) => {
+    if (!userId) throw new Error('缺少用户');
+    const ts = clock.now();
+    if (taskId) {
+      const [result] = await pool.query(
+        `UPDATE h5_scheduled_tasks
+         SET status = 'cancelled', updated_at = ?
+         WHERE id = ? AND user_id = ? AND status IN ('active', 'locked', 'failed')`,
+        [ts, taskId, userId],
+      );
+      if (Number(result?.affectedRows ?? 0) !== 1) {
+        throw new Error('未找到可取消的定时任务');
+      }
+      const [rows] = await pool.query(
+        `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`,
+        [taskId],
+      );
+      return rowToTask(rows[0]);
+    }
+    const safeTitleMatch = String(titleMatch ?? '').trim();
+    if (!safeTitleMatch) throw new Error('取消任务需要 taskId 或 titleMatch');
+    const [rows] = await pool.query(
+      `SELECT *
+       FROM h5_scheduled_tasks
+       WHERE user_id = ?
+         AND status IN ('active', 'locked', 'failed')
+         AND title LIKE ?
+       ORDER BY created_at DESC
+       LIMIT 1`,
+      [userId, `%${safeTitleMatch}%`],
+    );
+    const target = rows[0];
+    if (!target) throw new Error('未找到可取消的定时任务');
+    await pool.query(
+      `UPDATE h5_scheduled_tasks
+       SET status = 'cancelled', updated_at = ?
+       WHERE id = ?`,
+      [ts, target.id],
+    );
+    return rowToTask({ ...target, status: 'cancelled', updated_at: ts });
+  };
+
+  const listDueTasks = async ({ now = clock.now(), limit = 50 } = {}) => {
+    const [rows] = await pool.query(
+      `SELECT *
+       FROM h5_scheduled_tasks
+       WHERE next_run_at <= ?
+         AND (status = 'active' OR (status = 'locked' AND locked_until IS NOT NULL AND locked_until <= ?))
+       ORDER BY next_run_at ASC
+       LIMIT ?`,
+      [now, now, Math.max(1, Math.min(200, Number(limit) || 50))],
+    );
+    return rows.map(rowToTask);
+  };
+
+  const lockTask = async (id, { now = clock.now(), lockMs = 300_000 } = {}) => {
+    const lockedUntil = now + lockMs;
+    const [result] = await pool.query(
+      `UPDATE h5_scheduled_tasks
+       SET status = 'locked', locked_until = ?, attempts = attempts + 1, updated_at = ?
+       WHERE id = ? AND (status = 'active' OR (status = 'locked' AND locked_until IS NOT NULL AND locked_until <= ?))`,
+      [lockedUntil, now, id, now],
+    );
+    if (Number(result?.affectedRows ?? 0) !== 1) return null;
+    const [rows] = await pool.query(
+      `SELECT * FROM h5_scheduled_tasks WHERE id = ? LIMIT 1`,
+      [id],
+    );
+    return rowToTask(rows[0]);
+  };
+
+  const markTaskRunning = async (task, { now = clock.now() } = {}) => {
+    await pool.query(
+      `UPDATE h5_scheduled_tasks
+       SET status = 'running', updated_at = ?
+       WHERE id = ?`,
+      [now, task.id],
+    );
+    return { ...task, status: 'running' };
+  };
+
+  const markTaskSucceeded = async (task, {
+    now = clock.now(),
+    result = null,
+    deliveryText = null,
+    sessionId = null,
+    requestId = null,
+  } = {}) => {
+    const lastResult = {
+      ...(result && typeof result === 'object' ? result : {}),
+      deliveryText: deliveryText ?? null,
+      sessionId: sessionId ?? null,
+      requestId: requestId ?? null,
+      finishedAt: now,
+    };
+    if (task.recurrence === 'once') {
+      await pool.query(
+        `UPDATE h5_scheduled_tasks
+         SET status = 'completed', last_run_at = ?, locked_until = NULL,
+             last_error = NULL, last_result_json = ?, updated_at = ?
+         WHERE id = ?`,
+        [now, JSON.stringify(lastResult), now, task.id],
+      );
+      return {
+        ...task,
+        status: 'completed',
+        lastRunAt: now,
+        lastResult,
+      };
+    }
+    const nextRunAt = computeScheduledTaskNextRunAt({
+      recurrence: task.recurrence,
+      hour: task.hour,
+      minute: task.minute,
+      weekday: task.weekday,
+      timezone: task.timezone,
+      now: now + 1000,
+    });
+    await pool.query(
+      `UPDATE h5_scheduled_tasks
+       SET status = 'active', next_run_at = ?, last_run_at = ?, locked_until = NULL,
+           last_error = NULL, last_result_json = ?, updated_at = ?
+       WHERE id = ?`,
+      [nextRunAt, now, JSON.stringify(lastResult), now, task.id],
+    );
+    return {
+      ...task,
+      status: 'active',
+      nextRunAt,
+      lastRunAt: now,
+      lastResult,
+    };
+  };
+
+  const markTaskFailed = async (task, error, {
+    now = clock.now(),
+    retryMs = 10 * 60 * 1000,
+    maxAttempts = 5,
+  } = {}) => {
+    const attempts = Number(task.attempts ?? 0);
+    const message = String(error?.message ?? error ?? '执行失败').slice(0, 500);
+    const terminal = attempts >= maxAttempts;
+    const nextStatus = terminal ? 'failed' : 'active';
+    const nextRunAt = terminal
+      ? task.nextRunAt
+      : now + retryMs;
+    await pool.query(
+      `UPDATE h5_scheduled_tasks
+       SET status = ?, next_run_at = ?, locked_until = NULL, last_error = ?, updated_at = ?
+       WHERE id = ?`,
+      [nextStatus, nextRunAt, message, now, task.id],
+    );
+    return {
+      ...task,
+      status: nextStatus,
+      nextRunAt,
+      lastError: message,
+    };
+  };
+
+  return {
+    createTask,
+    listTasks,
+    cancelTask,
+    listDueTasks,
+    lockTask,
+    markTaskRunning,
+    markTaskSucceeded,
+    markTaskFailed,
+    computeScheduledTaskNextRunAt,
+  };
+}