memind/scheduled-task-service.mjs

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