import crypto from 'node:crypto'; const DEFAULT_TIMEZONE = 'Asia/Shanghai'; function parseJson(value) { if (value == null || value === '') return null; if (typeof value === 'object') return value; try { return JSON.parse(value); } catch { return null; } } function rowToUnifiedTask(row) { if (!row) return null; return { id: row.id, userId: row.user_id, type: row.type, title: row.title, spec: parseJson(row.spec_json) ?? {}, trigger: parseJson(row.trigger_json) ?? {}, action: parseJson(row.action_json) ?? {}, actionLevel: Number(row.action_level ?? 1), notifyChannel: row.notify_channel ?? 'both', status: row.status, nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at), lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at), legacyRef: parseJson(row.legacy_ref_json), sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }; } function legacyKey(ref) { if (!ref?.table || !ref?.id) return null; return `${ref.table}:${ref.id}`; } /** MySQL JSON column key order is not stable; match by extracted fields. */ function legacyRefMatchSql(column = 'legacy_ref_json') { return `${column} IS NOT NULL AND JSON_UNQUOTE(JSON_EXTRACT(${column}, '$.table')) = ? AND JSON_UNQUOTE(JSON_EXTRACT(${column}, '$.id')) = ?`; } function legacyRefMatchParams(legacyRef) { return [legacyRef.table, legacyRef.id]; } function formatTriggerLabel(task, timezone = DEFAULT_TIMEZONE) { const trigger = task.trigger ?? {}; if (trigger.at) return trigger.at; if (trigger.runAtLocal) return trigger.runAtLocal; if (trigger.hour != null) { return `${String(trigger.hour).padStart(2, '0')}:${String(trigger.minute ?? 0).padStart(2, '0')}`; } if (task.nextRunAt) { return new Date(task.nextRunAt).toLocaleString('zh-CN', { timeZone: trigger.timezone || timezone, hour12: false, }); } return '待确认'; } export function formatUnifiedTaskListReply(tasks = [], { timezone = DEFAULT_TIMEZONE } = {}) { if (!tasks.length) return '你当前没有进行中的提醒或定时任务。'; const lines = ['你的任务一览:']; for (const task of tasks.slice(0, 10)) { const when = formatTriggerLabel(task, timezone); const repeat = task.trigger?.repeat ? ` · ${task.trigger.repeat}` : ''; const typeLabel = { reminder: '提醒', todo: '待办', digest: '待办摘要', automation: '自动任务', condition: '条件提醒', }[task.type] ?? task.type; lines.push(`- [${typeLabel}] ${task.title}(${when}${repeat})`); } return lines.join('\n'); } function mapCommitToUnifiedTask({ userId, kind, committed, timezone = DEFAULT_TIMEZONE }) { const now = Date.now(); const base = { userId, actionLevel: kind === 'scheduled_task' ? 2 : 1, notifyChannel: 'wechat', status: 'active', sourceChannel: committed?.item?.sourceChannel ?? committed?.task?.sourceChannel ?? committed?.subscription?.sourceChannel ?? 'wechat', sourceMessageId: committed?.item?.sourceMessageId ?? committed?.task?.sourceMessageId ?? committed?.subscription?.sourceMessageId ?? null, sourceText: committed?.item?.sourceText ?? committed?.task?.sourceText ?? committed?.subscription?.sourceText ?? null, createdAt: now, updatedAt: now, }; if (kind === 'timed_reminder') { const item = committed.item ?? {}; const reminder = committed.reminder ?? {}; return { ...base, type: 'reminder', title: item.title ?? '提醒', spec: { itemId: item.id, reminderId: reminder.id }, trigger: { kind: 'at', at: item.startAt ?? reminder.remindAt, timezone: item.timezone ?? timezone, }, action: { kind: 'notify', channel: reminder.channel ?? 'wechat' }, nextRunAt: Number(reminder.remindAt ?? item.startAt ?? null) || null, legacyRef: { table: 'h5_schedule_reminders', id: reminder.id }, }; } if (kind === 'create_todo') { const item = committed.item ?? {}; return { ...base, type: 'todo', title: item.title ?? '待办', spec: { itemId: item.id }, trigger: { kind: 'none', timezone: item.timezone ?? timezone }, action: { kind: 'none' }, nextRunAt: null, legacyRef: { table: 'h5_schedule_items', id: item.id }, }; } if (kind === 'create_daily_todo_digest') { const subscription = committed.subscription ?? {}; return { ...base, type: 'digest', title: '每日待办摘要', spec: { subscriptionId: subscription.id }, trigger: { kind: 'cron', repeat: 'daily', hour: subscription.hour, minute: subscription.minute ?? 0, timezone: subscription.timezone ?? timezone, }, action: { kind: 'notify', channel: subscription.channel ?? 'wechat' }, nextRunAt: subscription.nextRunAt ?? null, legacyRef: { table: 'h5_schedule_digest_subscriptions', id: subscription.id }, }; } if (kind === 'create_balance_alert') { const subscription = committed.subscription ?? {}; return { ...base, type: 'condition', title: '余额低提醒', spec: { subscriptionId: subscription.id, thresholdCents: subscription.thresholdCents }, trigger: { kind: 'condition', watch: 'balance', op: 'lt', value: subscription.thresholdCents, }, action: { kind: 'notify', channel: subscription.channel ?? 'wechat' }, nextRunAt: subscription.nextRunAt ?? null, actionLevel: 2, legacyRef: { table: 'h5_balance_alert_subscriptions', id: subscription.id }, }; } if (kind === 'scheduled_task') { const task = committed.task ?? {}; return { ...base, type: 'automation', title: task.title ?? '定时自动任务', spec: { taskSpec: task.taskSpec, scheduledTaskId: task.id }, trigger: { kind: 'cron', repeat: task.recurrence, hour: task.hour, minute: task.minute ?? 0, weekday: task.weekday, timezone: task.timezone ?? timezone, }, action: { kind: 'agent_run', notifyChannel: task.notifyChannel ?? 'both' }, nextRunAt: task.nextRunAt ?? null, actionLevel: 2, legacyRef: { table: 'h5_scheduled_tasks', id: task.id }, }; } return null; } export function mapLegacyScheduledTaskRow(row, timezone = DEFAULT_TIMEZONE) { return { userId: row.user_id, type: 'automation', title: row.title, spec: { taskSpec: row.task_spec, scheduledTaskId: row.id }, trigger: { kind: 'cron', repeat: 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 || timezone, }, action: { kind: 'agent_run' }, actionLevel: 2, notifyChannel: row.notify_channel ?? 'both', status: row.status ?? 'active', nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at), lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at), legacyRef: { table: 'h5_scheduled_tasks', id: row.id }, sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, }; } export function mapLegacyReminderRow(row, timezone = DEFAULT_TIMEZONE) { return { userId: row.user_id ?? row.userId, type: 'reminder', title: row.title, spec: { itemId: row.item_id, reminderId: row.reminder_id ?? row.id }, trigger: { kind: 'at', at: Number(row.remind_at), timezone: row.timezone || timezone, }, action: { kind: 'notify', channel: row.channel ?? 'wechat' }, actionLevel: 1, notifyChannel: row.channel ?? 'wechat', status: row.status ?? 'pending', nextRunAt: Number(row.remind_at), legacyRef: { table: 'h5_schedule_reminders', id: row.reminder_id ?? row.id }, sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, }; } export function mapLegacyDigestRow(row, timezone = DEFAULT_TIMEZONE) { return { userId: row.user_id, type: 'digest', title: '每日待办摘要', spec: { subscriptionId: row.id }, trigger: { kind: 'cron', repeat: 'daily', hour: Number(row.hour), minute: Number(row.minute ?? 0), timezone: row.timezone || timezone, }, action: { kind: 'notify', channel: row.channel ?? 'wechat' }, actionLevel: 2, notifyChannel: row.channel ?? 'wechat', status: row.status ?? 'active', nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at), legacyRef: { table: 'h5_schedule_digest_subscriptions', id: row.id }, sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, }; } export function mapLegacyBalanceAlertRow(row) { return { userId: row.user_id, type: 'condition', title: '余额低提醒', spec: { subscriptionId: row.id, thresholdCents: Number(row.threshold_cents) }, trigger: { kind: 'condition', watch: 'balance', op: 'lt', value: Number(row.threshold_cents), }, action: { kind: 'notify', channel: row.channel ?? 'wechat' }, actionLevel: 2, notifyChannel: row.channel ?? 'wechat', status: row.status ?? 'active', nextRunAt: row.next_run_at == null ? null : Number(row.next_run_at), legacyRef: { table: 'h5_balance_alert_subscriptions', id: row.id }, sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, }; } export function mapLegacyTodoItemRow(row, timezone = DEFAULT_TIMEZONE) { return { userId: row.user_id, type: 'todo', title: row.title, spec: { itemId: row.id }, trigger: { kind: 'none', timezone: row.timezone || timezone }, action: { kind: 'none' }, actionLevel: 1, notifyChannel: 'both', status: row.status ?? 'active', nextRunAt: null, legacyRef: { table: 'h5_schedule_items', id: row.id }, sourceChannel: row.source_channel ?? null, sourceMessageId: row.source_message_id ?? null, sourceText: row.source_text ?? null, }; } export function createTaskUnifiedService(pool, { clock = { now: () => Date.now() } } = {}) { if (!pool) throw new Error('缺少数据库连接'); const insertUnifiedTask = async (task) => { const id = crypto.randomUUID(); const now = clock.now(); await pool.query( `INSERT INTO h5_tasks (id, user_id, type, title, spec_json, trigger_json, action_json, action_level, notify_channel, status, next_run_at, last_run_at, legacy_ref_json, source_channel, source_message_id, source_text, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [ id, task.userId, task.type, task.title, JSON.stringify(task.spec ?? {}), JSON.stringify(task.trigger ?? {}), JSON.stringify(task.action ?? {}), task.actionLevel ?? 1, task.notifyChannel ?? 'both', task.status ?? 'active', task.nextRunAt ?? null, task.lastRunAt ?? null, task.legacyRef ? JSON.stringify(task.legacyRef) : null, task.sourceChannel ?? null, task.sourceMessageId ?? null, task.sourceText ?? null, now, now, ], ); const [rows] = await pool.query(`SELECT * FROM h5_tasks WHERE id = ? LIMIT 1`, [id]); return rowToUnifiedTask(rows?.[0]); }; const upsertUnifiedTask = async (mapped) => { if (!mapped?.userId || !mapped?.type || !mapped?.title) return null; if (mapped.legacyRef?.table && mapped.legacyRef?.id) { const [existing] = await pool.query( `SELECT * FROM h5_tasks WHERE user_id = ? AND ${legacyRefMatchSql()} LIMIT 1`, [mapped.userId, ...legacyRefMatchParams(mapped.legacyRef)], ); if (existing?.[0]) return rowToUnifiedTask(existing[0]); } return insertUnifiedTask(mapped); }; const syncFromCommit = async ({ userId, kind, committed }) => { const mapped = mapCommitToUnifiedTask({ userId, kind, committed }); return upsertUnifiedTask(mapped); }; const syncFromLegacyMapped = async (mapped) => upsertUnifiedTask(mapped); const markLegacyCancelled = async ({ userId, legacyRef }) => { if (!userId || !legacyRef?.table || !legacyRef?.id) return 0; const [result] = await pool.query( `UPDATE h5_tasks SET status = 'cancelled', updated_at = ? WHERE user_id = ? AND ${legacyRefMatchSql()}`, [clock.now(), userId, ...legacyRefMatchParams(legacyRef)], ); return Number(result?.affectedRows ?? 0); }; const syncLegacyLifecycle = async ({ userId, legacyRef, status, nextRunAt = undefined, lastRunAt = undefined, lastError = undefined, }) => { if (!userId || !legacyRef?.table || !legacyRef?.id || !status) return 0; const sets = ['status = ?', 'updated_at = ?']; const params = [status, clock.now()]; if (nextRunAt !== undefined) { sets.push('next_run_at = ?'); params.push(nextRunAt); } if (lastRunAt !== undefined) { sets.push('last_run_at = ?'); params.push(lastRunAt); } if (lastError !== undefined) { const [rows] = await pool.query( `SELECT spec_json FROM h5_tasks WHERE user_id = ? AND ${legacyRefMatchSql()} LIMIT 1`, [userId, ...legacyRefMatchParams(legacyRef)], ); const spec = parseJson(rows?.[0]?.spec_json) ?? {}; spec.lastError = lastError ?? null; sets.push('spec_json = ?'); params.push(JSON.stringify(spec)); } params.push(userId, ...legacyRefMatchParams(legacyRef)); const [result] = await pool.query( `UPDATE h5_tasks SET ${sets.join(', ')} WHERE user_id = ? AND ${legacyRefMatchSql()}`, params, ); return Number(result?.affectedRows ?? 0); }; const listStoredTasks = async ({ userId, status = 'active', limit = 20 } = {}) => { const [rows] = await pool.query( `SELECT * FROM h5_tasks WHERE user_id = ? AND status = ? ORDER BY COALESCE(next_run_at, 9223372036854775807), created_at DESC LIMIT ?`, [userId, status, Math.max(1, Math.min(100, Number(limit) || 20))], ); return rows.map(rowToUnifiedTask); }; const listLegacyTasks = async ({ userId, limit = 20, timezone = DEFAULT_TIMEZONE } = {}) => { const tasks = []; const cap = Math.max(1, Math.min(100, Number(limit) || 20)); const [automations] = await pool.query( `SELECT id, user_id, title, task_spec, recurrence, hour, minute, weekday, timezone, next_run_at, status, source_channel, source_message_id, source_text, created_at, updated_at FROM h5_scheduled_tasks WHERE user_id = ? AND status = 'active' ORDER BY next_run_at ASC LIMIT ?`, [userId, cap], ); for (const row of automations) { tasks.push({ id: `legacy:scheduled:${row.id}`, ...mapLegacyScheduledTaskRow(row, timezone), createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }); } const [reminders] = await pool.query( `SELECT r.id AS reminder_id, r.user_id, r.remind_at, r.channel, r.status, i.id AS item_id, i.title, i.timezone, i.source_channel, i.source_message_id, i.source_text, i.created_at, i.updated_at FROM h5_schedule_reminders r JOIN h5_schedule_items i ON i.id = r.item_id WHERE r.user_id = ? AND r.status IN ('pending', 'locked') AND i.deleted_at IS NULL ORDER BY r.remind_at ASC LIMIT ?`, [userId, cap], ); for (const row of reminders) { tasks.push({ id: `legacy:reminder:${row.reminder_id}`, ...mapLegacyReminderRow(row, timezone), createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }); } const [digests] = await pool.query( `SELECT id, hour, minute, timezone, channel, status, next_run_at, source_channel, source_message_id, source_text, created_at, updated_at FROM h5_schedule_digest_subscriptions WHERE user_id = ? AND status = 'active' LIMIT ?`, [userId, cap], ); for (const row of digests) { tasks.push({ id: `legacy:digest:${row.id}`, ...mapLegacyDigestRow(row, timezone), createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }); } const [alerts] = await pool.query( `SELECT id, user_id, threshold_cents, channel, status, next_run_at, source_channel, source_message_id, source_text, created_at, updated_at FROM h5_balance_alert_subscriptions WHERE user_id = ? AND status = 'active' LIMIT ?`, [userId, cap], ); for (const row of alerts) { tasks.push({ id: `legacy:condition:${row.id}`, ...mapLegacyBalanceAlertRow(row), createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }); } const [todos] = await pool.query( `SELECT id, user_id, title, timezone, status, source_channel, source_message_id, source_text, created_at, updated_at FROM h5_schedule_items WHERE user_id = ? AND kind = 'task' AND status = 'active' AND deleted_at IS NULL ORDER BY created_at DESC LIMIT ?`, [userId, cap], ); for (const row of todos) { tasks.push({ id: `legacy:todo:${row.id}`, ...mapLegacyTodoItemRow(row, timezone), createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }); } return tasks.slice(0, cap); }; const collectLegacyMigrationCandidates = async ({ userId = null } = {}) => { const mapped = []; const [automations] = userId ? await pool.query(`SELECT * FROM h5_scheduled_tasks WHERE user_id = ? AND status = 'active'`, [userId]) : await pool.query(`SELECT * FROM h5_scheduled_tasks WHERE status = 'active'`); mapped.push(...automations.map((row) => mapLegacyScheduledTaskRow(row))); const reminderSql = `SELECT r.id AS reminder_id, r.user_id, r.remind_at, r.channel, r.status, i.id AS item_id, i.title, i.timezone, i.source_channel, i.source_message_id, i.source_text FROM h5_schedule_reminders r JOIN h5_schedule_items i ON i.id = r.item_id WHERE ${userId ? 'r.user_id = ? AND' : ''} r.status IN ('pending', 'locked') AND i.deleted_at IS NULL`; const [reminders] = userId ? await pool.query(reminderSql, [userId]) : await pool.query(reminderSql); mapped.push(...reminders.map((row) => mapLegacyReminderRow(row))); const [digests] = userId ? await pool.query(`SELECT * FROM h5_schedule_digest_subscriptions WHERE user_id = ? AND status = 'active'`, [userId]) : await pool.query(`SELECT * FROM h5_schedule_digest_subscriptions WHERE status = 'active'`); mapped.push(...digests.map((row) => mapLegacyDigestRow(row))); const [alerts] = userId ? await pool.query(`SELECT * FROM h5_balance_alert_subscriptions WHERE user_id = ? AND status = 'active'`, [userId]) : await pool.query(`SELECT * FROM h5_balance_alert_subscriptions WHERE status = 'active'`); mapped.push(...alerts.map((row) => mapLegacyBalanceAlertRow(row))); const [todos] = userId ? await pool.query(`SELECT * FROM h5_schedule_items WHERE user_id = ? AND kind = 'task' AND status = 'active' AND deleted_at IS NULL`, [userId]) : await pool.query(`SELECT * FROM h5_schedule_items WHERE kind = 'task' AND status = 'active' AND deleted_at IS NULL`); mapped.push(...todos.map((row) => mapLegacyTodoItemRow(row))); return mapped; }; const migrateLegacyTasks = async ({ userId = null, dryRun = true } = {}) => { const candidates = await collectLegacyMigrationCandidates({ userId }); const stats = { scanned: candidates.length, inserted: 0, skipped: 0, errors: 0 }; if (dryRun) return { ...stats, dryRun: true, sample: candidates.slice(0, 5) }; for (const candidate of candidates) { try { const before = await pool.query( `SELECT id FROM h5_tasks WHERE user_id = ? AND ${legacyRefMatchSql()} LIMIT 1`, [candidate.userId, ...legacyRefMatchParams(candidate.legacyRef)], ); if (before?.[0]?.[0]) { stats.skipped += 1; continue; } await upsertUnifiedTask(candidate); stats.inserted += 1; } catch { stats.errors += 1; } } return { ...stats, dryRun: false }; }; const listUserTasks = async ({ userId, status = 'active', limit = 20, timezone = DEFAULT_TIMEZONE } = {}) => { if (!userId) throw new Error('缺少用户'); const stored = await listStoredTasks({ userId, status, limit }); const mirrored = new Set( stored.map((task) => legacyKey(task.legacyRef)).filter(Boolean), ); const legacy = await listLegacyTasks({ userId, limit, timezone }); const merged = [...stored]; for (const task of legacy) { const key = legacyKey(task.legacyRef); if (key && mirrored.has(key)) continue; merged.push(task); } merged.sort((a, b) => { const aRun = a.nextRunAt ?? Number.MAX_SAFE_INTEGER; const bRun = b.nextRunAt ?? Number.MAX_SAFE_INTEGER; if (aRun !== bRun) return aRun - bRun; return b.createdAt - a.createdAt; }); return merged.slice(0, Math.max(1, Math.min(100, Number(limit) || 20))); }; return { syncFromCommit, syncFromLegacyMapped, markLegacyCancelled, syncLegacyLifecycle, migrateLegacyTasks, listUserTasks, listStoredTasks, listLegacyTasks, formatUnifiedTaskListReply, }; } export { formatTriggerLabel, mapCommitToUnifiedTask };