import crypto from 'node:crypto'; import { addLocalDays, formatLocalTime, localDateLabel, nextDailyRunAt, normalizeTimezone, startOfLocalDay, } from './schedule-time.mjs'; const DEFAULT_TIMEZONE = 'Asia/Shanghai'; function nowMs() { return Date.now(); } function rowToItem(row) { if (!row) return null; return { id: row.id, userId: row.user_id, kind: row.kind, title: row.title, description: row.description ?? null, status: row.status, startAt: row.start_at == null ? null : Number(row.start_at), endAt: row.end_at == null ? null : Number(row.end_at), dueAt: row.due_at == null ? null : Number(row.due_at), allDay: Boolean(row.all_day), timezone: row.timezone || DEFAULT_TIMEZONE, location: row.location ?? null, createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), }; } function rowToDigest(row) { if (!row) return null; return { id: row.id, userId: row.user_id, hour: Number(row.hour), minute: Number(row.minute), timezone: row.timezone || DEFAULT_TIMEZONE, channel: row.channel || 'wechat', status: row.status, nextRunAt: Number(row.next_run_at), lastRunAt: row.last_run_at == null ? null : Number(row.last_run_at), attempts: Number(row.attempts ?? 0), }; } function formatTodoDigest(items, { now = nowMs(), timezone = DEFAULT_TIMEZONE } = {}) { const date = localDateLabel(now, timezone); if (!items.length) { return `${date} 待办记录\n\n今天暂时没有待办。`; } const lines = [`${date} 待办记录`, '']; items.forEach((item, index) => { const time = item.dueAt || item.startAt ? `(${formatLocalTime(item.dueAt || item.startAt, timezone)})` : ''; lines.push(`${index + 1}. ${item.title}${time}`); }); return lines.join('\n'); } export function createScheduleService(pool, options = {}) { const defaultTimezone = normalizeTimezone(options.defaultTimezone || process.env.H5_DEFAULT_TIMEZONE); const clock = options.clock || { now: nowMs }; const createItem = async ({ userId, kind = 'task', title, description = null, startAt = null, endAt = null, dueAt = null, allDay = false, timezone = defaultTimezone, location = null, sourceChannel = 'agent', sourceSessionId = null, sourceMessageId = null, sourceText = null, metadata = null, }) => { if (!userId) throw new Error('缺少用户'); const cleanTitle = String(title ?? '').trim(); if (!cleanTitle) throw new Error('缺少待办标题'); const safeKind = kind === 'event' ? 'event' : 'task'; const id = crypto.randomUUID(); const ts = clock.now(); await pool.query( `INSERT INTO h5_schedule_items (id, user_id, kind, title, description, status, start_at, end_at, due_at, all_day, timezone, location, source_channel, source_session_id, source_message_id, source_text, metadata_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, [ id, userId, safeKind, cleanTitle, description, startAt, endAt, dueAt, allDay ? 1 : 0, normalizeTimezone(timezone), location, sourceChannel, sourceSessionId, sourceMessageId, sourceText, metadata ? JSON.stringify(metadata) : null, ts, ts, ], ); return { id, userId, kind: safeKind, title: cleanTitle, startAt, endAt, dueAt, allDay: Boolean(allDay), timezone: normalizeTimezone(timezone), status: 'active', createdAt: ts, updatedAt: ts, }; }; const listItems = async ({ userId, from = null, to = null, status = 'active', limit = 100 } = {}) => { if (!userId) throw new Error('缺少用户'); const clauses = ['user_id = ?', 'deleted_at IS NULL']; const params = [userId]; if (status) { clauses.push('status = ?'); params.push(status); } if (from != null && to != null) { clauses.push( `( (start_at IS NOT NULL AND start_at >= ? AND start_at < ?) OR (due_at IS NOT NULL AND due_at >= ? AND due_at < ?) OR (start_at IS NULL AND due_at IS NULL) )`, ); params.push(from, to, from, to); } params.push(Math.max(1, Math.min(500, Number(limit) || 100))); const [rows] = await pool.query( `SELECT * FROM h5_schedule_items WHERE ${clauses.join(' AND ')} ORDER BY COALESCE(start_at, due_at, 9223372036854775807), created_at LIMIT ?`, params, ); return rows.map(rowToItem); }; const listTodayTodoItems = async ({ userId, timezone = defaultTimezone, now = clock.now() } = {}) => { const start = startOfLocalDay(now, timezone); const end = addLocalDays(start, 1, timezone); return listItems({ userId, from: start, to: end, status: 'active', limit: 200 }); }; const createDailyTodoDigest = async ({ userId, hour, minute = 0, timezone = defaultTimezone, channel = 'wechat', sourceChannel = 'agent', sourceSessionId = null, sourceMessageId = null, sourceText = null, }) => { if (!userId) throw new Error('缺少用户'); const safeHour = Number(hour); const safeMinute = Number(minute); if (!Number.isInteger(safeHour) || safeHour < 0 || safeHour > 23) { throw new Error('提醒小时无效'); } if (!Number.isInteger(safeMinute) || safeMinute < 0 || safeMinute > 59) { throw new Error('提醒分钟无效'); } const tz = normalizeTimezone(timezone); const nextRunAt = nextDailyRunAt({ hour: safeHour, minute: safeMinute, timezone: tz, now: clock.now(), }); const id = crypto.randomUUID(); const ts = clock.now(); await pool.query( `INSERT INTO h5_schedule_digest_subscriptions (id, user_id, digest_type, hour, minute, timezone, channel, status, next_run_at, source_channel, source_session_id, source_message_id, source_text, created_at, updated_at) VALUES (?, ?, 'todo_day', ?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE hour = VALUES(hour), minute = VALUES(minute), timezone = VALUES(timezone), channel = VALUES(channel), status = 'active', next_run_at = VALUES(next_run_at), source_channel = VALUES(source_channel), source_session_id = VALUES(source_session_id), source_message_id = VALUES(source_message_id), source_text = VALUES(source_text), updated_at = VALUES(updated_at)`, [ id, userId, safeHour, safeMinute, tz, channel, nextRunAt, sourceChannel, sourceSessionId, sourceMessageId, sourceText, ts, ts, ], ); const [rows] = await pool.query( `SELECT * FROM h5_schedule_digest_subscriptions WHERE user_id = ? AND digest_type = 'todo_day' AND channel = ? LIMIT 1`, [userId, channel], ); return rowToDigest(rows[0]) || { id, userId, hour: safeHour, minute: safeMinute, timezone: tz, channel, status: 'active', nextRunAt, lastRunAt: null, attempts: 0, }; }; const listDueDigestSubscriptions = async ({ now = clock.now(), limit = 50 } = {}) => { const [rows] = await pool.query( `SELECT * FROM h5_schedule_digest_subscriptions 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(rowToDigest); }; const lockDigestSubscription = async (id, { now = clock.now(), lockMs = 120_000 } = {}) => { const lockedUntil = now + lockMs; const [result] = await pool.query( `UPDATE h5_schedule_digest_subscriptions 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_schedule_digest_subscriptions WHERE id = ? LIMIT 1`, [id], ); return rowToDigest(rows[0]); }; const markDigestSent = async (subscription, { now = clock.now() } = {}) => { const nextRunAt = nextDailyRunAt({ hour: subscription.hour, minute: subscription.minute, timezone: subscription.timezone, now: now + 1000, }); await pool.query( `UPDATE h5_schedule_digest_subscriptions SET status = 'active', next_run_at = ?, last_run_at = ?, locked_until = NULL, last_error = NULL, updated_at = ? WHERE id = ?`, [nextRunAt, now, now, subscription.id], ); return { ...subscription, status: 'active', nextRunAt, lastRunAt: now }; }; const markDigestFailed = async (subscription, error, { now = clock.now(), retryMs = 10 * 60 * 1000, maxAttempts = 5 } = {}) => { const attempts = Number(subscription.attempts ?? 0); const status = attempts >= maxAttempts ? 'failed' : 'active'; const nextRunAt = status === 'active' ? now + retryMs : subscription.nextRunAt; await pool.query( `UPDATE h5_schedule_digest_subscriptions SET status = ?, next_run_at = ?, locked_until = NULL, last_error = ?, updated_at = ? WHERE id = ?`, [status, nextRunAt, String(error?.message ?? error ?? '发送失败').slice(0, 500), now, subscription.id], ); }; const logDelivery = async ({ subscriptionId, userId, channel = 'wechat', status, providerMessageId = null, errorCode = null, errorMessage = null }) => { await pool.query( `INSERT INTO h5_schedule_delivery_logs (id, reminder_id, subscription_id, user_id, channel, status, provider_message_id, error_code, error_message, created_at) VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)`, [ crypto.randomUUID(), subscriptionId, userId, channel, status, providerMessageId, errorCode, errorMessage ? String(errorMessage).slice(0, 500) : null, clock.now(), ], ); }; const buildTodoDigestText = async ({ userId, timezone = defaultTimezone, now = clock.now() }) => { const items = await listTodayTodoItems({ userId, timezone, now }); return formatTodoDigest(items, { now, timezone }); }; return { createItem, listItems, listTodayTodoItems, createDailyTodoDigest, listDueDigestSubscriptions, lockDigestSubscription, markDigestSent, markDigestFailed, logDelivery, buildTodoDigestText, }; }