From d820247aa1c99ea8e314a0c6dec3b2c1ce2fe67f Mon Sep 17 00:00:00 2001 From: john Date: Wed, 16 Sep 2026 21:30:53 +0800 Subject: [PATCH] fix(schedule): retry reminders when WeChat delivery is deferred Treat deferred/skipped customer-service sends as failures so reminder worker does not mark reminders sent, and add a 103 read-only health check script. Co-authored-by: Cursor --- notification-dispatcher.mjs | 17 +- notification-dispatcher.test.mjs | 16 +- package.json | 1 + schedule-reminder-worker.mjs | 17 +- schedule-reminder-worker.test.mjs | 61 +++++ scheduled-task-worker.mjs | 16 +- scheduled-task-worker.test.mjs | 111 ++++++++ .../check-schedule-reminder-health-103.mjs | 252 ++++++++++++++++++ 8 files changed, 485 insertions(+), 6 deletions(-) create mode 100644 scripts/check-schedule-reminder-health-103.mjs diff --git a/notification-dispatcher.mjs b/notification-dispatcher.mjs index b4da67c..65cd1d6 100644 --- a/notification-dispatcher.mjs +++ b/notification-dispatcher.mjs @@ -1,10 +1,25 @@ -function resolveWechatDispatchSent(result) { +export const WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT = + 'WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT'; + +export function resolveWechatDispatchSent(result) { if (result && typeof result === 'object') { return result.sent !== false && !result.deferred && !result.skipped; } return result !== false; } +export async function deliverWechatScheduleNotification( + sendScheduleNotification, + payload, +) { + const result = await sendScheduleNotification(payload); + if (!resolveWechatDispatchSent(result)) { + const err = new Error('微信提醒发送未完成（deferred、skipped 或未绑定）'); + err.code = WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT; + throw err; + } +} + export function createNotificationDispatcher({ sendWechatTextToUser, logger = console } = {}) { const sendWechat = async (userId, text, options = {}) => { if (typeof sendWechatTextToUser !== 'function') return false; diff --git a/notification-dispatcher.test.mjs b/notification-dispatcher.test.mjs index 76dd5e6..70ddf97 100644 --- a/notification-dispatcher.test.mjs +++ b/notification-dispatcher.test.mjs @@ -1,6 +1,10 @@ import assert from 'node:assert/strict'; import test from 'node:test'; -import { createNotificationDispatcher } from './notification-dispatcher.mjs'; +import { + createNotificationDispatcher, + deliverWechatScheduleNotification, + WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT, +} from './notification-dispatcher.mjs'; test('notification dispatcher forwards recharge success text unchanged', async () => { const sent = []; @@ -144,3 +148,13 @@ test('notification dispatcher returns false when wechat sender is unavailable', }, ]); }); + +test('deliverWechatScheduleNotification throws when sender defers', async () => { + await assert.rejects( + () => deliverWechatScheduleNotification( + async () => ({ sent: false, deferred: true, errcode: 45015 }), + { userId: 'user-5', text: '提醒' }, + ), + (err) => err.code === WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT, + ); +}); diff --git a/package.json b/package.json index 258c4fe..0849a99 100644 --- a/package.json +++ b/package.json @@ -139,6 +139,7 @@ "verify:tang-itl-readiness-103": "node scripts/verify-tang-itl-readiness-103.mjs", "simulate:tang-wechat-itl": "node scripts/simulate-tang-wechat-itl-flow.mjs", "check:itl-rollout-config": "node scripts/check-itl-rollout-config.mjs", + "check:schedule-reminder-health-103": "node scripts/check-schedule-reminder-health-103.mjs", "test:mindspace-e2e": "node scripts/mindspace-e2e.mjs", "test:mindspace-pages-e2e": "node scripts/mindspace-pages-e2e.mjs", "test:mindspace-publications-e2e": "node scripts/mindspace-publications-e2e.mjs", diff --git a/schedule-reminder-worker.mjs b/schedule-reminder-worker.mjs index 76462cc..331eb60 100644 --- a/schedule-reminder-worker.mjs +++ b/schedule-reminder-worker.mjs @@ -1,3 +1,5 @@ +import { deliverWechatScheduleNotification } from './notification-dispatcher.mjs'; + export function startScheduleReminderWorker({ scheduleService, sendWechatTextToUser, @@ -45,7 +47,10 @@ export function startScheduleReminderWorker({ }, }); if (reminder.channel === 'wechat') { - await sendScheduleNotification({ userId: reminder.userId, text }); + await deliverWechatScheduleNotification(sendScheduleNotification, { + userId: reminder.userId, + text, + }); } await scheduleService.logDelivery({ reminderId: reminder.id, @@ -88,7 +93,10 @@ export function startScheduleReminderWorker({ timezone: subscription.timezone, }, }); - await sendScheduleNotification({ userId: subscription.userId, text }); + await deliverWechatScheduleNotification(sendScheduleNotification, { + userId: subscription.userId, + text, + }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, @@ -135,7 +143,10 @@ export function startScheduleReminderWorker({ balanceCents, }, }); - await sendScheduleNotification({ userId: subscription.userId, text }); + await deliverWechatScheduleNotification(sendScheduleNotification, { + userId: subscription.userId, + text, + }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, diff --git a/schedule-reminder-worker.test.mjs b/schedule-reminder-worker.test.mjs index 408970b..75ef95e 100644 --- a/schedule-reminder-worker.test.mjs +++ b/schedule-reminder-worker.test.mjs @@ -264,3 +264,64 @@ test('schedule reminder worker skips wechat for in_app reminders', async () => { assert.deepEqual(sent, []); }); + +test('schedule reminder worker retries when wechat delivery is deferred', async () => { + const calls = []; + const reminder = { + id: 'rem-deferred', + userId: 'user-1', + itemId: 'item-1', + remindAt: Date.now() - 1000, + channel: 'wechat', + attempts: 1, + }; + const worker = startScheduleReminderWorker({ + intervalMs: 60_000, + scheduleService: { + async listDueReminders() { + return [reminder]; + }, + async lockReminder() { + return reminder; + }, + async buildReminderText() { + return '【待办提醒】开会'; + }, + async createUserNotification() {}, + async logDelivery(input) { + calls.push(`log:${input.status}`); + }, + async markReminderSent() { + calls.push('sent'); + }, + async markReminderFailed() { + calls.push('failed'); + }, + async markReminderCancelled() {}, + async listDueDigestSubscriptions() { + return []; + }, + async lockDigestSubscription() { + return null; + }, + async listDueBalanceAlerts() { + return []; + }, + async lockBalanceAlert() { + return null; + }, + }, + notificationDispatcher: { + async sendScheduleNotification() { + return false; + }, + }, + logger: { warn() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + assert.deepEqual(calls, ['log:failed', 'failed']); +}); diff --git a/scheduled-task-worker.mjs b/scheduled-task-worker.mjs index 21400be..702292a 100644 --- a/scheduled-task-worker.mjs +++ b/scheduled-task-worker.mjs @@ -1,3 +1,7 @@ +import { + resolveWechatDispatchSent, + WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT, +} from './notification-dispatcher.mjs'; import { buildScheduledTaskVerifiedHtmlUrls, deliveryTextPromisesPublicHtml, @@ -72,7 +76,7 @@ export function startScheduledTaskWorker({ userId: task.userId, }); } else { - const sent = await sendScheduleNotification({ + const sendResult = await sendScheduleNotification({ userId: task.userId, text, verifiedHtmlUrls, @@ -80,6 +84,7 @@ export function startScheduledTaskWorker({ logger.warn?.('Scheduled task wechat notification failed:', err); return false; }); + const sent = resolveWechatDispatchSent(sendResult); if (sent) { wechatDelivery = { sentAt: Date.now(), @@ -87,6 +92,15 @@ export function startScheduledTaskWorker({ source: 'scheduled_task_worker', textOnly: verifiedHtmlUrls.length === 0, }; + } else if (notifyChannel === 'wechat') { + const err = new Error('定时任务微信发送未完成（deferred、skipped 或未绑定）'); + err.code = WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT; + throw err; + } else { + logger.warn?.('[ScheduledTask] wechat delivery incomplete; web notification kept', { + taskId: task.id, + userId: task.userId, + }); } } } diff --git a/scheduled-task-worker.test.mjs b/scheduled-task-worker.test.mjs index 0f0efbd..3e07267 100644 --- a/scheduled-task-worker.test.mjs +++ b/scheduled-task-worker.test.mjs @@ -289,3 +289,114 @@ test('scheduled task worker marks failure when page link is not deliverable yet' assert.deepEqual(calls, ['failed:SCHEDULED_TASK_NON_DELIVERY', 'notify:scheduled_task_failed']); }); + +test('scheduled task worker fails wechat-only task when delivery is deferred', async () => { + const calls = []; + const task = { + id: 'task-wechat-deferred', + userId: 'user-6', + title: '仅微信通知', + recurrence: 'once', + notifyChannel: 'wechat', + attempts: 1, + }; + const worker = startScheduledTaskWorker({ + intervalMs: 60_000, + userAuth: { id: 'user-auth' }, + tkmindProxy: { id: 'proxy' }, + scheduledTaskService: { + async listDueTasks() { + return [task]; + }, + async lockTask() { + return task; + }, + async markTaskRunning(input) { + return input; + }, + async markTaskSucceeded() { + calls.push('success'); + }, + async markTaskFailed(input, err) { + calls.push(`failed:${err.code}`); + return { ...input, status: 'failed', lastError: err.message }; + }, + }, + notificationDispatcher: { + async sendScheduleNotification() { + return false; + }, + }, + executeTask: async () => ({ + sessionId: 'session-6', + requestId: 'req-6', + deliveryText: '任务摘要已完成，详细内容已写入 MindSpace 工作区。', + readyPaths: [], + }), + logger: { warn() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + assert.deepEqual(calls, ['failed:WECHAT_SCHEDULE_NOTIFICATION_NOT_SENT']); +}); + +test('scheduled task worker keeps both-channel task when wechat is deferred but web succeeds', async () => { + const calls = []; + const task = { + id: 'task-both-deferred', + userId: 'user-7', + title: '双通道通知', + recurrence: 'once', + notifyChannel: 'both', + attempts: 1, + }; + const worker = startScheduledTaskWorker({ + intervalMs: 60_000, + userAuth: { id: 'user-auth' }, + tkmindProxy: { id: 'proxy' }, + scheduledTaskService: { + async listDueTasks() { + return [task]; + }, + async lockTask() { + return task; + }, + async markTaskRunning(input) { + return input; + }, + async markTaskSucceeded(input, payload) { + calls.push(`success:${payload.result.wechatDelivery ? 'wechat' : 'web-only'}`); + return input; + }, + async markTaskFailed() { + calls.push('failed'); + }, + }, + scheduleService: { + async createUserNotification() { + calls.push('notify:web'); + }, + }, + notificationDispatcher: { + async sendScheduleNotification() { + return false; + }, + }, + executeTask: async () => ({ + sessionId: 'session-7', + requestId: 'req-7', + deliveryText: '任务摘要已完成，详细内容已写入 MindSpace 工作区。', + readyPaths: [], + }), + logger: { warn() {} }, + runOnStart: false, + }); + + await worker.runOnce(); + worker.stop(); + + assert.deepEqual(calls, ['notify:web', 'success:web-only']); +}); diff --git a/scripts/check-schedule-reminder-health-103.mjs b/scripts/check-schedule-reminder-health-103.mjs new file mode 100644 index 0000000..fe96a9f --- /dev/null +++ b/scripts/check-schedule-reminder-health-103.mjs @@ -0,0 +1,252 @@ +#!/usr/bin/env node +/** + * 103 服务号定时提醒健康巡检（只读） + * + * 用法: + * node scripts/check-schedule-reminder-health-103.mjs + * DATABASE_URL=... node scripts/check-schedule-reminder-health-103.mjs + * + * 在 103 上可配合: + * cd /Users/john/Project/Memind && node scripts/check-schedule-reminder-health-103.mjs + */ +import process from 'node:process'; +import mysql from 'mysql2/promise'; +import { isScheduledTaskWorkerEnabled } from '../scheduled-task-worker-config.mjs'; +import { isWechatNewsMorningDraftWorkerEnabled } from '../wechat-news-morning-draft-worker-config.mjs'; +import { loadH5Environment } from './load-env.mjs'; + +loadH5Environment(import.meta.dirname); + +const env = process.env; +const now = Date.now(); +const dayMs = 24 * 60 * 60 * 1000; +const hourMs = 60 * 60 * 1000; + +let passed = 0; +let failed = 0; +let warned = 0; + +function pass(label, detail = '') { + passed += 1; + console.log(`✔ ${label}${detail ? `: ${detail}` : ''}`); +} + +function fail(label, detail = '') { + failed += 1; + console.error(`✘ ${label}${detail ? `: ${detail}` : ''}`); +} + +function warn(label, detail = '') { + warned += 1; + console.warn(`△ ${label}${detail ? `: ${detail}` : ''}`); +} + +function envFlag(name) { + return String(env[name] ?? '').trim(); +} + +function envEnabled(name) { + return envFlag(name) === '1'; +} + +async function scalar(pool, sql, params = []) { + const [rows] = await pool.query(sql, params); + return Number(rows?.[0]?.c ?? rows?.[0]?.count ?? 0); +} + +async function checkEnvironment() { + console.log('\n=== 环境变量 ===\n'); + + if (envEnabled('H5_WECHAT_MP_ENABLED')) pass('H5_WECHAT_MP_ENABLED=1'); + else fail('H5_WECHAT_MP_ENABLED=1'); + + if (envEnabled('H5_SCHEDULE_ENABLED')) pass('H5_SCHEDULE_ENABLED=1'); + else fail('H5_SCHEDULE_ENABLED=1'); + + if (envEnabled('H5_REMINDER_WORKER_ENABLED')) pass('H5_REMINDER_WORKER_ENABLED=1'); + else fail('H5_REMINDER_WORKER_ENABLED=1'); + + if (isScheduledTaskWorkerEnabled(env)) { + pass('Scheduled task worker enabled'); + } else { + warn('Scheduled task worker disabled', 'H5_SCHEDULED_TASK_WORKER_ENABLED=0 且 H5_REMINDER_WORKER_ENABLED≠1'); + } + + if (isWechatNewsMorningDraftWorkerEnabled(env)) { + pass('News morning draft worker enabled'); + } else { + warn('News morning draft worker disabled'); + } + + if (envFlag('H5_DEFAULT_TIMEZONE')) { + pass('H5_DEFAULT_TIMEZONE', envFlag('H5_DEFAULT_TIMEZONE')); + } else { + warn('H5_DEFAULT_TIMEZONE 未设置', '默认 Asia/Shanghai'); + } + + const passiveCandidate = + envFlag('MEMIND_PORTAL_RUNTIME_ROLE') === 'candidate' + && envFlag('MEMIND_CANARY_PASSIVE_RUNTIME') !== '0'; + if (passiveCandidate) { + fail('Passive canary runtime', 'worker 会被禁用，不应承载定时提醒'); + } else { + pass('非 passive canary runtime'); + } +} + +async function checkDatabase(pool) { + console.log('\n=== 数据库指标（只读）===\n'); + + const overduePending = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_reminders + WHERE status = 'pending' AND remind_at < ?`, + [now - hourMs], + ); + if (overduePending === 0) pass('无 overdue pending 提醒'); + else fail('overdue pending 提醒', String(overduePending)); + + const stuckLocked = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_reminders + WHERE status = 'locked' AND locked_until IS NOT NULL AND locked_until < ?`, + [now], + ); + if (stuckLocked === 0) pass('无 stuck locked 提醒'); + else fail('stuck locked 提醒', String(stuckLocked)); + + const failedReminders24h = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_reminders + WHERE status = 'failed' AND updated_at >= ?`, + [now - dayMs], + ); + if (failedReminders24h === 0) pass('24h 内无 failed 提醒'); + else warn('24h 内 failed 提醒', String(failedReminders24h)); + + const deliveryFailed24h = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_delivery_logs + WHERE status = 'failed' AND created_at >= ?`, + [now - dayMs], + ); + if (deliveryFailed24h === 0) pass('24h 内无 failed delivery log'); + else warn('24h 内 failed delivery log', String(deliveryFailed24h)); + + const deliverySuccess24h = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_delivery_logs + WHERE status = 'success' AND created_at >= ?`, + [now - dayMs], + ); + pass('24h delivery success 计数', String(deliverySuccess24h)); + + const deferredQueue = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_wechat_mp_deferred_delivery`, + ); + if (deferredQueue === 0) pass('deferred 队列为空'); + else warn('deferred 队列积压', String(deferredQueue)); + + const activeDigests = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_digest_subscriptions WHERE status = 'active'`, + ); + pass('active 待办摘要订阅', String(activeDigests)); + + const activeScheduledTasks = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_scheduled_tasks WHERE status = 'active'`, + ); + pass('active 定时自动任务', String(activeScheduledTasks)); + + const failedTasks24h = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_scheduled_tasks + WHERE status = 'failed' AND updated_at >= ?`, + [now - dayMs], + ); + if (failedTasks24h === 0) pass('24h 内无 failed 定时任务'); + else warn('24h 内 failed 定时任务', String(failedTasks24h)); + + const morningReminders = await scalar( + pool, + `SELECT COUNT(*) AS c FROM h5_schedule_items + WHERE status = 'active' + AND deleted_at IS NULL + AND JSON_UNQUOTE(JSON_EXTRACT(metadata_json, '$.source')) = 'subscribe_morning_reminder'`, + ); + pass('active 早安提醒订阅', String(morningReminders)); + + const [recentFailedDeliveries] = await pool.query( + `SELECT d.created_at, d.error_message, u.username, d.channel + FROM h5_schedule_delivery_logs d + JOIN h5_users u ON u.id = d.user_id + WHERE d.status = 'failed' AND d.created_at >= ? + ORDER BY d.created_at DESC + LIMIT 5`, + [now - dayMs], + ); + if (recentFailedDeliveries.length > 0) { + console.log('\n--- 最近失败投递（最多 5 条）---'); + for (const row of recentFailedDeliveries) { + console.log( + ` ${new Date(Number(row.created_at)).toISOString()} ${row.username} ${row.channel} ${row.error_message ?? ''}`, + ); + } + } + + const [overdueSamples] = await pool.query( + `SELECT r.id, r.remind_at, r.attempts, r.last_error, u.username, i.title + FROM h5_schedule_reminders r + JOIN h5_schedule_items i ON i.id = r.item_id + JOIN h5_users u ON u.id = r.user_id + WHERE r.status = 'pending' AND r.remind_at < ? + ORDER BY r.remind_at ASC + LIMIT 5`, + [now - hourMs], + ); + if (overdueSamples.length > 0) { + console.log('\n--- overdue pending 样本（最多 5 条）---'); + for (const row of overdueSamples) { + console.log( + ` ${row.username} "${row.title}" attempts=${row.attempts} remind_at=${new Date(Number(row.remind_at)).toISOString()} ${row.last_error ?? ''}`, + ); + } + } +} + +async function main() { + console.log('=== 103 服务号定时提醒健康巡检 ==='); + console.log(`时间: ${new Date(now).toISOString()}`); + + await checkEnvironment(); + + if (!env.DATABASE_URL) { + fail('DATABASE_URL', '未配置，跳过数据库检查'); + summarize(); + process.exit(1); + } + + const pool = mysql.createPool({ uri: env.DATABASE_URL, connectionLimit: 2 }); + try { + await checkDatabase(pool); + } finally { + await pool.end(); + } + + summarize(); + process.exit(failed > 0 ? 1 : 0); +} + +function summarize() { + console.log('\n=== 汇总 ==='); + console.log(`通过: ${passed}`); + console.log(`警告: ${warned}`); + console.log(`失败: ${failed}`); +} + +main().catch((error) => { + console.error(error); + process.exit(1); +});