Add smart ACK provider for WeChat MP replies

Replace fixed ackText with a rule-based AckProvider that picks response templates by message type and intent (translate, summary, rewrite, poster, ppt, mindmap, code, search, schedule). Pure sync, zero I/O, auto-falls back to config.ackText on any error. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-26 15:19:03 +08:00
parent 9ed4fd48d7
commit 9b4a25799f
162 changed files with 17276 additions and 2054 deletions
@@ -50,6 +50,73 @@ function rowToDigest(row) {
  };
 }

+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 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) {
@@ -163,12 +230,128 @@ export function createScheduleService(pool, options = {}) {
    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 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,
@@ -251,6 +434,107 @@ export function createScheduleService(pool, options = {}) {
    };
  };

+  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 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 *
@@ -334,16 +618,136 @@ export function createScheduleService(pool, options = {}) {
    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,
+    createReminder,
    listItems,
+    listItemsBySourceMessage,
    listTodayTodoItems,
+    listDigestSubscriptions,
    createDailyTodoDigest,
+    createBalanceLowAlert,
    listDueDigestSubscriptions,
    lockDigestSubscription,
    markDigestSent,
    markDigestFailed,
+    listDueBalanceAlerts,
+    lockBalanceAlert,
+    markBalanceAlertSent,
+    markBalanceAlertFailed,
    logDelivery,
    buildTodoDigestText,
+    getUserWalletSnapshot,
+    createUserNotification,
+    listUserNotifications,
+    markUserNotificationRead,
+    markAllUserNotificationsRead,
+    deleteUserNotification,
+    clearUserNotifications,
  };
 }