memind/schedule-service.mjs

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 rowToBalanceAlert(row) {
  if (!row) return null;
  return {
    id: row.id,
    userId: row.user_id,
    thresholdCents: Number(row.threshold_cents),
    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),
    lastNotifiedBalanceCents:
      row.last_notified_balance_cents == null ? null : Number(row.last_notified_balance_cents),
    attempts: Number(row.attempts ?? 0),
    lastError: row.last_error ?? null,
    lockedUntil: row.locked_until == null ? null : Number(row.locked_until),
    sourceChannel: row.source_channel ?? null,
    sourceSessionId: row.source_session_id ?? null,
    sourceMessageId: row.source_message_id ?? null,
    sourceText: row.source_text ?? null,
  };
}

function rowToReminder(row) {
  if (!row) return null;
  return {
    id: row.id,
    userId: row.user_id,
    itemId: row.item_id,
    remindAt: Number(row.remind_at),
    offsetMinutes: row.offset_minutes == null ? null : Number(row.offset_minutes),
    channel: row.channel || 'wechat',
    status: row.status,
    attempts: Number(row.attempts ?? 0),
    lastError: row.last_error ?? null,
    lockedUntil: row.locked_until == null ? null : Number(row.locked_until),
    sentAt: row.sent_at == null ? null : Number(row.sent_at),
    createdAt: Number(row.created_at),
    updatedAt: Number(row.updated_at),
  };
}

function rowToReminderWithItem(row) {
  const reminder = rowToReminder(row);
  if (!reminder) return null;
  return {
    ...reminder,
    itemTitle: String(row.item_title ?? '').trim(),
    itemKind: row.item_kind === 'event' ? 'event' : 'task',
    itemTimezone: row.item_timezone || DEFAULT_TIMEZONE,
    itemStartAt: row.item_start_at == null ? null : Number(row.item_start_at),
    itemEndAt: row.item_end_at == null ? null : Number(row.item_end_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;
}

function rowToUserNotification(row) {
  if (!row) return null;
  return {
    id: row.id,
    userId: row.user_id,
    channel: row.channel,
    notificationType: row.notification_type,
    title: row.title,
    body: row.body,
    data: parseJsonColumn(row.data_json),
    status: row.status,
    readAt: row.read_at == null ? null : Number(row.read_at),
    createdAt: Number(row.created_at),
    updatedAt: Number(row.updated_at),
  };
}

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 listItemsBySourceMessage = async ({ userId, sourceMessageId, limit = 20 } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const cleanSourceMessageId = String(sourceMessageId ?? '').trim();
    if (!cleanSourceMessageId) return [];
    const [rows] = await pool.query(
      `SELECT *
       FROM h5_schedule_items
       WHERE user_id = ?
         AND source_message_id = ?
         AND deleted_at IS NULL
       ORDER BY created_at DESC
       LIMIT ?`,
      [userId, cleanSourceMessageId, Math.max(1, Math.min(100, Number(limit) || 20))],
    );
    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 listUpcomingItems = async ({
    userId,
    timezone = defaultTimezone,
    days = 7,
    now = clock.now(),
  } = {}) => {
    const safeDays = Math.max(1, Math.min(30, Number(days) || 7));
    const start = startOfLocalDay(now, timezone);
    const end = addLocalDays(start, safeDays, timezone);
    return listItems({ userId, from: start, to: end, status: 'active', limit: 200 });
  };

  const listUpcomingReminders = async ({
    userId,
    timezone = defaultTimezone,
    days = 7,
    now = clock.now(),
    limit = 100,
  } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const safeDays = Math.max(1, Math.min(30, Number(days) || 7));
    const start = startOfLocalDay(now, timezone);
    const end = addLocalDays(start, safeDays, timezone);
    const [rows] = await pool.query(
      `SELECT r.*,
              i.title AS item_title,
              i.kind AS item_kind,
              i.timezone AS item_timezone,
              i.start_at AS item_start_at,
              i.end_at AS item_end_at
       FROM h5_schedule_reminders r
       INNER JOIN h5_schedule_items i ON i.id = r.item_id
       WHERE r.user_id = ?
         AND r.status IN ('pending', 'locked', 'sent')
         AND r.remind_at >= ?
         AND r.remind_at < ?
         AND i.deleted_at IS NULL
         AND i.status = 'active'
       ORDER BY r.remind_at ASC
       LIMIT ?`,
      [
        userId,
        start,
        end,
        Math.max(1, Math.min(200, Number(limit) || 100)),
      ],
    );
    return rows.map(rowToReminderWithItem);
  };

  const getItem = async ({ userId, itemId }) => {
    if (!userId || !itemId) throw new Error('缺少事项参数');
    const [rows] = await pool.query(
      `SELECT *
       FROM h5_schedule_items
       WHERE id = ? AND user_id = ? AND deleted_at IS NULL
       LIMIT 1`,
      [itemId, userId],
    );
    return rowToItem(rows[0]);
  };

  const getReminder = async ({ userId, reminderId }) => {
    if (!userId || !reminderId) throw new Error('缺少提醒参数');
    const [rows] = await pool.query(
      `SELECT r.*,
              i.title AS item_title,
              i.kind AS item_kind,
              i.timezone AS item_timezone,
              i.start_at AS item_start_at,
              i.end_at AS item_end_at
       FROM h5_schedule_reminders r
       INNER JOIN h5_schedule_items i ON i.id = r.item_id
       WHERE r.id = ? AND r.user_id = ? AND i.deleted_at IS NULL
       LIMIT 1`,
      [reminderId, userId],
    );
    return rowToReminderWithItem(rows[0]);
  };

  const cancelReminder = async ({ userId, reminderId, reason = '用户忽略' } = {}) => {
    const reminder = await getReminder({ userId, reminderId });
    if (!reminder) throw new Error('提醒不存在或无权访问');
    if (reminder.status === 'cancelled') return reminder;
    if (reminder.status === 'sent') throw new Error('已通知的提醒不能忽略');
    return markReminderCancelled(reminder, reason);
  };

  const deleteReminder = async ({ userId, reminderId }) => {
    if (!userId || !reminderId) throw new Error('缺少提醒参数');
    const [result] = await pool.query(
      `DELETE FROM h5_schedule_reminders WHERE id = ? AND user_id = ?`,
      [reminderId, userId],
    );
    return Number(result?.affectedRows ?? 0) > 0;
  };

  const deleteReminders = async ({ userId, reminderIds }) => {
    if (!userId) throw new Error('缺少用户');
    const ids = [...new Set((reminderIds ?? []).map((id) => String(id ?? '').trim()).filter(Boolean))];
    if (ids.length === 0) return 0;
    const [result] = await pool.query(
      `DELETE FROM h5_schedule_reminders WHERE user_id = ? AND id IN (${ids.map(() => '?').join(', ')})`,
      [userId, ...ids],
    );
    return Number(result?.affectedRows ?? 0);
  };

  const createReminder = async ({
    userId,
    itemId,
    remindAt,
    offsetMinutes = null,
    channel = 'wechat',
  }) => {
    if (!userId) throw new Error('缺少用户');
    if (!itemId) throw new Error('缺少事项');
    const safeRemindAt = Number(remindAt);
    if (!Number.isFinite(safeRemindAt) || safeRemindAt <= 0) {
      throw new Error('提醒时间无效');
    }
    const safeOffsetMinutes =
      offsetMinutes == null || offsetMinutes === ''
        ? null
        : Number(offsetMinutes);
    if (safeOffsetMinutes != null && !Number.isInteger(safeOffsetMinutes)) {
      throw new Error('提醒偏移分钟无效');
    }
    const safeChannel = channel === 'in_app' ? 'in_app' : 'wechat';
    const [itemRows] = await pool.query(
      `SELECT id
       FROM h5_schedule_items
       WHERE id = ? AND user_id = ? AND deleted_at IS NULL
       LIMIT 1`,
      [itemId, userId],
    );
    if (!itemRows[0]) throw new Error('事项不存在或无权访问');
    const id = crypto.randomUUID();
    const ts = clock.now();
    await pool.query(
      `INSERT INTO h5_schedule_reminders
       (id, user_id, item_id, remind_at, offset_minutes, channel, status, attempts,
        last_error, locked_until, sent_at, created_at, updated_at)
       VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, NULL, NULL, NULL, ?, ?)
       ON DUPLICATE KEY UPDATE
         offset_minutes = VALUES(offset_minutes),
         status = 'pending',
         last_error = NULL,
         locked_until = NULL,
         updated_at = VALUES(updated_at)`,
      [id, userId, itemId, safeRemindAt, safeOffsetMinutes, safeChannel, ts, ts],
    );
    const [rows] = await pool.query(
      `SELECT *
       FROM h5_schedule_reminders
       WHERE item_id = ? AND remind_at = ? AND channel = ?
       LIMIT 1`,
      [itemId, safeRemindAt, safeChannel],
    );
    return rowToReminder(rows[0]) || {
      id,
      userId,
      itemId,
      remindAt: safeRemindAt,
      offsetMinutes: safeOffsetMinutes,
      channel: safeChannel,
      status: 'pending',
      attempts: 0,
      lastError: null,
      lockedUntil: null,
      sentAt: null,
      createdAt: ts,
      updatedAt: ts,
    };
  };

  const listDigestSubscriptions = async ({
    userId,
    digestType = 'todo_day',
    channel = null,
    status = null,
    limit = 20,
  } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const clauses = ['user_id = ?', 'digest_type = ?'];
    const params = [userId, digestType];
    if (channel) {
      clauses.push('channel = ?');
      params.push(channel);
    }
    if (status) {
      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_schedule_digest_subscriptions
       WHERE ${clauses.join(' AND ')}
       ORDER BY next_run_at ASC, created_at DESC
       LIMIT ?`,
      params,
    );
    return rows.map(rowToDigest);
  };

  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 createBalanceLowAlert = async ({
    userId,
    thresholdCents,
    channel = 'wechat',
    sourceChannel = 'agent',
    sourceSessionId = null,
    sourceMessageId = null,
    sourceText = null,
  } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const safeThreshold = Number(thresholdCents);
    if (!Number.isFinite(safeThreshold) || safeThreshold < 0) {
      throw new Error('余额阈值无效');
    }
    const now = clock.now();
    const id = crypto.randomUUID();
    await pool.query(
      `INSERT INTO h5_balance_alert_subscriptions
       (id, user_id, threshold_cents, channel, status, next_run_at, last_run_at,
        last_notified_balance_cents, attempts, locked_until, last_error,
        source_channel, source_session_id, source_message_id, source_text, created_at, updated_at)
       VALUES (?, ?, ?, ?, 'active', ?, NULL, NULL, 0, NULL, NULL, ?, ?, ?, ?, ?, ?)
       ON DUPLICATE KEY UPDATE
         threshold_cents = VALUES(threshold_cents),
         channel = VALUES(channel),
         status = 'active',
         next_run_at = VALUES(next_run_at),
         last_error = NULL,
         locked_until = NULL,
         updated_at = VALUES(updated_at)`,
      [id, userId, safeThreshold, channel, now, sourceChannel, sourceSessionId, sourceMessageId, sourceText, now, now],
    );
    const [rows] = await pool.query(
      `SELECT * FROM h5_balance_alert_subscriptions WHERE user_id = ? AND channel = ? LIMIT 1`,
      [userId, channel],
    );
    return rowToBalanceAlert(rows[0]) || {
      id,
      userId,
      thresholdCents: safeThreshold,
      channel,
      status: 'active',
      nextRunAt: now,
      lastRunAt: null,
      lastNotifiedBalanceCents: null,
      attempts: 0,
    };
  };

  const listDueReminders = async ({ now = clock.now(), limit = 50 } = {}) => {
    const [rows] = await pool.query(
      `SELECT r.*
       FROM h5_schedule_reminders r
       INNER JOIN h5_schedule_items i ON i.id = r.item_id
       WHERE r.status = 'pending'
         AND r.remind_at <= ?
         AND (r.locked_until IS NULL OR r.locked_until <= ?)
         AND i.deleted_at IS NULL
         AND i.status = 'active'
       ORDER BY r.remind_at ASC
       LIMIT ?`,
      [now, now, Math.max(1, Math.min(200, Number(limit) || 50))],
    );
    return rows.map(rowToReminder);
  };

  const lockReminder = async (id, { now = clock.now(), lockMs = 120_000 } = {}) => {
    const lockedUntil = now + lockMs;
    const [result] = await pool.query(
      `UPDATE h5_schedule_reminders
       SET status = 'locked', locked_until = ?, attempts = attempts + 1, updated_at = ?
       WHERE id = ? AND status = 'pending' AND remind_at <= ?`,
      [lockedUntil, now, id, now],
    );
    if (Number(result?.affectedRows ?? 0) !== 1) return null;
    const [rows] = await pool.query(
      `SELECT * FROM h5_schedule_reminders WHERE id = ? LIMIT 1`,
      [id],
    );
    return rowToReminder(rows[0]);
  };

  const markReminderSent = async (reminder, { now = clock.now() } = {}) => {
    await pool.query(
      `UPDATE h5_schedule_reminders
       SET status = 'sent', sent_at = ?, locked_until = NULL, last_error = NULL, updated_at = ?
       WHERE id = ?`,
      [now, now, reminder.id],
    );
    return { ...reminder, status: 'sent', sentAt: now, lockedUntil: null };
  };

  const markReminderCancelled = async (reminder, reason, { now = clock.now() } = {}) => {
    const lastError = String(reason ?? '已取消').slice(0, 500);
    await pool.query(
      `UPDATE h5_schedule_reminders
       SET status = 'cancelled', locked_until = NULL, last_error = ?, updated_at = ?
       WHERE id = ?`,
      [lastError, now, reminder.id],
    );
    return { ...reminder, status: 'cancelled', lastError };
  };

  const markReminderFailed = async (
    reminder,
    error,
    { now = clock.now(), retryMs = 10 * 60 * 1000, maxAttempts = 5 } = {},
  ) => {
    const attempts = Number(reminder.attempts ?? 0);
    const status = attempts >= maxAttempts ? 'failed' : 'pending';
    const lockedUntil = status === 'pending' ? now + retryMs : null;
    await pool.query(
      `UPDATE h5_schedule_reminders
       SET status = ?, locked_until = ?, last_error = ?, updated_at = ?
       WHERE id = ?`,
      [
        status,
        lockedUntil,
        String(error?.message ?? error ?? '发送失败').slice(0, 500),
        now,
        reminder.id,
      ],
    );
  };

  const buildReminderText = async (reminder) => {
    const item = await getItem({ userId: reminder.userId, itemId: reminder.itemId });
    if (!item || item.status !== 'active') return null;
    const timezone = item.timezone || defaultTimezone;
    const remindLabel = formatLocalTime(reminder.remindAt, timezone);
    const eventAt = item.startAt ?? item.dueAt ?? null;
    const eventLabel = eventAt ? formatLocalTime(eventAt, timezone) : null;
    const lines = [`【待办提醒】${item.title}`];
    if (eventLabel && eventLabel !== remindLabel) {
      lines.push(`事项时间：${eventLabel}`);
    }
    lines.push(`提醒时间：${remindLabel}`);
    if (item.location) lines.push(`地点：${item.location}`);
    return lines.join('\n');
  };

  const listDueBalanceAlerts = async ({ now = clock.now(), limit = 50 } = {}) => {
    const [rows] = await pool.query(
      `SELECT *
       FROM h5_balance_alert_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(rowToBalanceAlert);
  };

  const lockBalanceAlert = async (id, { now = clock.now(), lockMs = 120_000 } = {}) => {
    const lockedUntil = now + lockMs;
    const [result] = await pool.query(
      `UPDATE h5_balance_alert_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_balance_alert_subscriptions WHERE id = ? LIMIT 1`,
      [id],
    );
    return rowToBalanceAlert(rows[0]);
  };

  const markBalanceAlertSent = async (subscription, { now = clock.now() } = {}) => {
    await pool.query(
      `UPDATE h5_balance_alert_subscriptions
       SET status = 'active', next_run_at = ?, last_run_at = ?, last_notified_balance_cents = ?,
           locked_until = NULL, last_error = NULL, updated_at = ?
       WHERE id = ?`,
      [now + 5 * 60 * 1000, now, subscription.lastNotifiedBalanceCents ?? null, now, subscription.id],
    );
    return { ...subscription, status: 'active', lastRunAt: now };
  };

  const markBalanceAlertFailed = 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_balance_alert_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 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 ({
    reminderId = null,
    subscriptionId = null,
    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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [
        crypto.randomUUID(),
        reminderId,
        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 });
  };

  const getUserWalletSnapshot = async (userId) => {
    if (!userId) throw new Error('缺少用户');
    const [rows] = await pool.query(
      `SELECT balance_cents AS balanceCents, tokens_used AS tokensUsed
       FROM h5_user_wallets
       WHERE user_id = ?
       LIMIT 1`,
      [userId],
    );
    return rows[0] ?? null;
  };

  const createUserNotification = async ({
    userId,
    channel = 'web',
    notificationType,
    title,
    body,
    data = null,
  }) => {
    if (!userId) throw new Error('缺少用户');
    if (!notificationType) throw new Error('缺少通知类型');
    if (!title) throw new Error('缺少通知标题');
    if (!body) throw new Error('缺少通知内容');
    const now = clock.now();
    const id = crypto.randomUUID();
    await pool.query(
      `INSERT INTO h5_user_notifications
       (id, user_id, channel, notification_type, title, body, data_json, status, read_at, created_at, updated_at)
       VALUES (?, ?, ?, ?, ?, ?, ?, 'unread', NULL, ?, ?)`,
      [id, userId, channel, notificationType, title, body, data ? JSON.stringify(data) : null, now, now],
    );
    return { id, userId, channel, notificationType, title, body, data, status: 'unread', readAt: null, createdAt: now, updatedAt: now };
  };

  const listUserNotifications = async ({ userId, status = 'unread', limit = 20 } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const clauses = ['user_id = ?'];
    const params = [userId];
    if (status && status !== 'all') {
      clauses.push('status = ?');
      params.push(status);
    }
    params.push(Math.max(1, Math.min(100, Number(limit) || 20)));
    const [rows] = await pool.query(
      `SELECT *
       FROM h5_user_notifications
       WHERE ${clauses.join(' AND ')}
       ORDER BY created_at DESC
       LIMIT ?`,
      params,
    );
    return rows.map(rowToUserNotification);
  };

  const markUserNotificationRead = async ({ userId, notificationId }) => {
    if (!userId || !notificationId) throw new Error('缺少通知参数');
    const now = clock.now();
    const [result] = await pool.query(
      `UPDATE h5_user_notifications
       SET status = 'read', read_at = ?, updated_at = ?
       WHERE id = ? AND user_id = ? AND status <> 'read'`,
      [now, now, notificationId, userId],
    );
    return Number(result?.affectedRows ?? 0) > 0;
  };

  const markAllUserNotificationsRead = async ({ userId } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const now = clock.now();
    const [result] = await pool.query(
      `UPDATE h5_user_notifications
       SET status = 'read', read_at = ?, updated_at = ?
       WHERE user_id = ? AND status <> 'read'`,
      [now, now, userId],
    );
    return Number(result?.affectedRows ?? 0);
  };

  const deleteUserNotification = async ({ userId, notificationId } = {}) => {
    if (!userId || !notificationId) throw new Error('缺少通知参数');
    const [result] = await pool.query(
      `DELETE FROM h5_user_notifications
       WHERE id = ? AND user_id = ?`,
      [notificationId, userId],
    );
    return Number(result?.affectedRows ?? 0) > 0;
  };

  const clearUserNotifications = async ({ userId, status = 'all' } = {}) => {
    if (!userId) throw new Error('缺少用户');
    const clauses = ['user_id = ?'];
    const params = [userId];
    if (status && status !== 'all') {
      clauses.push('status = ?');
      params.push(status);
    }
    const [result] = await pool.query(
      `DELETE FROM h5_user_notifications
       WHERE ${clauses.join(' AND ')}`,
      params,
    );
    return Number(result?.affectedRows ?? 0);
  };

  return {
    createItem,
    getItem,
    createReminder,
    listItems,
    listItemsBySourceMessage,
    listTodayTodoItems,
    listUpcomingItems,
    listUpcomingReminders,
    getReminder,
    cancelReminder,
    deleteReminder,
    deleteReminders,
    listDigestSubscriptions,
    createDailyTodoDigest,
    createBalanceLowAlert,
    listDueReminders,
    lockReminder,
    markReminderSent,
    markReminderCancelled,
    markReminderFailed,
    buildReminderText,
    listDueDigestSubscriptions,
    lockDigestSubscription,
    markDigestSent,
    markDigestFailed,
    listDueBalanceAlerts,
    lockBalanceAlert,
    markBalanceAlertSent,
    markBalanceAlertFailed,
    logDelivery,
    buildTodoDigestText,
    getUserWalletSnapshot,
    createUserNotification,
    listUserNotifications,
    markUserNotificationRead,
    markAllUserNotificationsRead,
    deleteUserNotification,
    clearUserNotifications,
  };
}