From 6e4e0e2937c5abc2356bfc00257be5bdf9a8f09f Mon Sep 17 00:00:00 2001 From: john Date: Tue, 30 Jun 2026 21:26:46 +0800 Subject: [PATCH] refactor: centralize wechat notification dispatch --- admin-bootstrap.mjs | 13 +++- notification-dispatcher.mjs | 30 ++++++++ notification-dispatcher.test.mjs | 126 +++++++++++++++++++++++++++++++ schedule-reminder-worker.mjs | 14 +++- server.mjs | 14 +++- user-auth.mjs | 1 + 6 files changed, 186 insertions(+), 12 deletions(-) create mode 100644 notification-dispatcher.mjs create mode 100644 notification-dispatcher.test.mjs diff --git a/admin-bootstrap.mjs b/admin-bootstrap.mjs index e22d7e3..5f393d4 100644 --- a/admin-bootstrap.mjs +++ b/admin-bootstrap.mjs @@ -21,6 +21,7 @@ import { createPlazaInteractionService } from './plaza-interactions.mjs'; import { createPlazaOpsService } from './plaza-ops.mjs'; import { createNoopPlazaRedis } from './plaza-redis.mjs'; import { ensureAlgorithmConfig, loadAlgorithmConfig } from './plaza-algorithm.mjs'; +import { createNotificationDispatcher } from './notification-dispatcher.mjs'; import { createWechatAdminService } from './wechat-admin.mjs'; import { createWechatMpService, loadWechatMpConfig } from './wechat-mp.mjs'; @@ -98,16 +99,20 @@ export async function createAdminServices(env = {}) { throw new Error('admin bootstrap does not proxy WeChat chat sessions'); }, }); - userAuth.setRechargeNotifier(async ({ userId, title, body }) => { - if (!wechatMpService?.enabled) return; - await wechatMpService.sendTextToUser(userId, `${title}\n${body}`.trim()); + const notificationDispatcher = createNotificationDispatcher({ + sendWechatTextToUser: wechatMpService?.enabled + ? (userId, text) => wechatMpService.sendTextToUser(userId, text) + : null, + }); + userAuth.setRechargeNotifier(async ({ userId, title, body, dedupeKey }) => { + await notificationDispatcher.sendRechargeSuccess({ userId, title, body, dedupeKey }); }); const wechatAdmin = createWechatAdminService(pool, { config: wechatMpConfig, scheduleEnabled: process.env.H5_SCHEDULE_ENABLED === '1', reminderWorkerEnabled: process.env.H5_REMINDER_WORKER_ENABLED === '1', sendWechatTextToUser: wechatMpService?.enabled - ? (userId, text) => wechatMpService.sendTextToUser(userId, text) + ? (userId, text) => notificationDispatcher.sendScheduleNotification({ userId, text }) : null, }); diff --git a/notification-dispatcher.mjs b/notification-dispatcher.mjs new file mode 100644 index 0000000..7cb32ab --- /dev/null +++ b/notification-dispatcher.mjs @@ -0,0 +1,30 @@ +export function createNotificationDispatcher({ sendWechatTextToUser, logger = console } = {}) { + const sendWechat = async (userId, text) => { + if (typeof sendWechatTextToUser !== 'function') return false; + await sendWechatTextToUser(userId, text); + return true; + }; + + return { + async sendRechargeSuccess({ userId, title, body, dedupeKey = null }) { + const sent = await sendWechat(userId, `${title}\n${body}`.trim()); + logger.info?.('Notification dispatch:', { + type: 'recharge_success', + userId, + dedupeKey, + sent, + }); + return sent; + }, + async sendScheduleNotification({ userId, text }) { + const sent = await sendWechat(userId, text); + logger.info?.('Notification dispatch:', { + type: 'schedule_notification', + userId, + dedupeKey: null, + sent, + }); + return sent; + }, + }; +} diff --git a/notification-dispatcher.test.mjs b/notification-dispatcher.test.mjs new file mode 100644 index 0000000..94139b9 --- /dev/null +++ b/notification-dispatcher.test.mjs @@ -0,0 +1,126 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createNotificationDispatcher } from './notification-dispatcher.mjs'; + +test('notification dispatcher forwards recharge success text unchanged', async () => { + const sent = []; + const dedupeKeys = []; + const logs = []; + const dispatcher = createNotificationDispatcher({ + async sendWechatTextToUser(userId, text) { + sent.push({ userId, text }); + }, + logger: { + info(message, payload) { + logs.push({ message, payload }); + }, + }, + }); + + const sentOk = await dispatcher.sendRechargeSuccess({ + userId: 'user-1', + title: '充值成功', + body: '你已成功充值 ¥25.00，余额已更新。', + dedupeKey: (dedupeKeys.push('recharge:order-1'), 'recharge:order-1'), + }); + + assert.equal(sentOk, true); + assert.deepEqual(dedupeKeys, ['recharge:order-1']); + assert.deepEqual(sent, [ + { userId: 'user-1', text: '充值成功\n你已成功充值 ¥25.00，余额已更新。' }, + ]); + assert.deepEqual(logs, [ + { + message: 'Notification dispatch:', + payload: { + type: 'recharge_success', + userId: 'user-1', + dedupeKey: 'recharge:order-1', + sent: true, + }, + }, + ]); +}); + +test('notification dispatcher forwards schedule notification text unchanged', async () => { + const sent = []; + const logs = []; + const dispatcher = createNotificationDispatcher({ + async sendWechatTextToUser(userId, text) { + sent.push({ userId, text }); + }, + logger: { + info(message, payload) { + logs.push({ message, payload }); + }, + }, + }); + + const sentOk = await dispatcher.sendScheduleNotification({ + userId: 'user-2', + text: '【待办提醒】开会\n提醒时间：09:00', + }); + + assert.equal(sentOk, true); + assert.deepEqual(sent, [ + { userId: 'user-2', text: '【待办提醒】开会\n提醒时间：09:00' }, + ]); + assert.deepEqual(logs, [ + { + message: 'Notification dispatch:', + payload: { + type: 'schedule_notification', + userId: 'user-2', + dedupeKey: null, + sent: true, + }, + }, + ]); +}); + +test('notification dispatcher returns false when wechat sender is unavailable', async () => { + const logs = []; + const dispatcher = createNotificationDispatcher({ + logger: { + info(message, payload) { + logs.push({ message, payload }); + }, + }, + }); + + assert.equal( + await dispatcher.sendRechargeSuccess({ + userId: 'user-3', + title: '充值成功', + body: '余额已更新。', + }), + false, + ); + assert.equal( + await dispatcher.sendScheduleNotification({ + userId: 'user-3', + text: '【待办提醒】开会', + }), + false, + ); + assert.deepEqual(logs, [ + { + message: 'Notification dispatch:', + payload: { + type: 'recharge_success', + userId: 'user-3', + dedupeKey: null, + sent: false, + }, + }, + { + message: 'Notification dispatch:', + payload: { + type: 'schedule_notification', + userId: 'user-3', + dedupeKey: null, + sent: false, + }, + }, + ]); +}); diff --git a/schedule-reminder-worker.mjs b/schedule-reminder-worker.mjs index 97e6d40..0911c10 100644 --- a/schedule-reminder-worker.mjs +++ b/schedule-reminder-worker.mjs @@ -1,12 +1,18 @@ export function startScheduleReminderWorker({ scheduleService, sendWechatTextToUser, + notificationDispatcher, logger = console, intervalMs = Number(process.env.H5_REMINDER_SCAN_INTERVAL_MS ?? 30_000), maxAttempts = Number(process.env.H5_REMINDER_MAX_ATTEMPTS ?? 5), runOnStart = true, } = {}) { - if (!scheduleService || typeof sendWechatTextToUser !== 'function') { + const sendScheduleNotification = + notificationDispatcher?.sendScheduleNotification?.bind(notificationDispatcher) ?? + (typeof sendWechatTextToUser === 'function' + ? async ({ userId, text }) => sendWechatTextToUser(userId, text) + : null); + if (!scheduleService || typeof sendScheduleNotification !== 'function') { return { stop() {} }; } @@ -39,7 +45,7 @@ export function startScheduleReminderWorker({ }, }); if (reminder.channel === 'wechat') { - await sendWechatTextToUser(reminder.userId, text); + await sendScheduleNotification({ userId: reminder.userId, text }); } await scheduleService.logDelivery({ reminderId: reminder.id, @@ -81,7 +87,7 @@ export function startScheduleReminderWorker({ timezone: subscription.timezone, }, }); - await sendWechatTextToUser(subscription.userId, text); + await sendScheduleNotification({ userId: subscription.userId, text }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, @@ -128,7 +134,7 @@ export function startScheduleReminderWorker({ balanceCents, }, }); - await sendWechatTextToUser(subscription.userId, text); + await sendScheduleNotification({ userId: subscription.userId, text }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, diff --git a/server.mjs b/server.mjs index 915c205..ec69d6b 100644 --- a/server.mjs +++ b/server.mjs @@ -34,6 +34,7 @@ import { PUBLISH_ROOT_DIR, PUBLISH_KEY_UUID, PUBLIC_ZONE_DIR, resolvePublishDir, import { ensureWorkspaceHtmlThumbnail, startWorkspaceThumbnailWatcher, workspaceThumbnailRelativePath } from './mindspace-workspace-thumbnails.mjs'; import { startWorkspaceAssetSyncWatcher } from './mindspace-workspace-sync.mjs'; import { attachRequestId, sendData, sendError } from './api-response.mjs'; +import { createNotificationDispatcher } from './notification-dispatcher.mjs'; import { createMindSpaceAuditWriter } from './mindspace-audit.mjs'; import { assertMindSpaceRoute, mindspaceFlags } from './mindspace-flags.mjs'; import { createMindSpaceService, DEFAULT_MAX_FILE_BYTES } from './mindspace.mjs'; @@ -258,6 +259,7 @@ let subscriptionService = null; let wechatPayClient = null; let wechatOAuthService = null; let wechatMpService = null; +let notificationDispatcher = null; let scheduleService = null; let feedbackService = null; let scheduleReminderWorker = null; @@ -592,9 +594,13 @@ async function bootstrapUserAuth() { }) : null, }); - userAuth.setRechargeNotifier(async ({ userId, title, body }) => { - if (!wechatMpService?.enabled) return; - await wechatMpService.sendTextToUser(userId, `${title}\n${body}`.trim()); + notificationDispatcher = createNotificationDispatcher({ + sendWechatTextToUser: wechatMpService?.enabled + ? (userId, text) => wechatMpService.sendTextToUser(userId, text) + : null, + }); + userAuth.setRechargeNotifier(async ({ userId, title, body, dedupeKey }) => { + await notificationDispatcher.sendRechargeSuccess({ userId, title, body, dedupeKey }); }); if ( process.env.H5_REMINDER_WORKER_ENABLED === '1' && @@ -603,7 +609,7 @@ async function bootstrapUserAuth() { ) { scheduleReminderWorker = startScheduleReminderWorker({ scheduleService, - sendWechatTextToUser: (userId, text) => wechatMpService.sendTextToUser(userId, text), + notificationDispatcher, }); console.log('Schedule reminder worker enabled'); } diff --git a/user-auth.mjs b/user-auth.mjs index 979f231..819b7a6 100644 --- a/user-auth.mjs +++ b/user-auth.mjs @@ -1215,6 +1215,7 @@ export function createUserAuth(pool, options = {}) { amountCents: amount, operatorId: operatorId ?? null, paymentOrderId, + dedupeKey: paymentOrderId ? `recharge:${paymentOrderId}` : null, notificationType: rechargeType, title, body,