From 3e8cdfdda32e5e46ad66656df52b5d6fd0f6ac4e Mon Sep 17 00:00:00 2001 From: john Date: Sun, 2 Aug 2026 09:21:43 +0800 Subject: [PATCH] fix(wechat): defer 45047/45015 delivery and control customer message budget Persist formal replies when WeChat customer-service quota is exhausted, claim them once on the next passive response, skip progress notices for page generation, and suppress failure notices while a formal reply is pending. Co-authored-by: Cursor --- schema.sql | 14 ++ wechat-mp.mjs | 268 ++++++++++++++-------- wechat-mp.test.mjs | 60 ++++- wechat/customer-service-deferred.mjs | 200 ++++++++++++++++ wechat/customer-service-deferred.test.mjs | 69 ++++++ 5 files changed, 508 insertions(+), 103 deletions(-) create mode 100644 wechat/customer-service-deferred.mjs create mode 100644 wechat/customer-service-deferred.test.mjs diff --git a/schema.sql b/schema.sql index 4287b12..a575b72 100644 --- a/schema.sql +++ b/schema.sql @@ -1010,6 +1010,20 @@ CREATE TABLE IF NOT EXISTS h5_wechat_mp_message_details ( CONSTRAINT fk_wechat_mp_message_details_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS h5_wechat_mp_deferred_delivery ( + app_id VARCHAR(32) NOT NULL, + openid VARCHAR(64) NOT NULL, + priority TINYINT NOT NULL DEFAULT 10, + kind VARCHAR(32) NOT NULL DEFAULT 'text', + content MEDIUMTEXT NOT NULL, + metadata JSON NULL, + source_msg_id VARCHAR(128) NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + PRIMARY KEY (app_id, openid), + KEY idx_wechat_deferred_updated (updated_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS h5_schedule_items ( id CHAR(36) PRIMARY KEY, user_id CHAR(36) NOT NULL, diff --git a/wechat-mp.mjs b/wechat-mp.mjs index c77747d..a6e7e3f 100644 --- a/wechat-mp.mjs +++ b/wechat-mp.mjs @@ -14,6 +14,14 @@ import { } from './wechat-media.mjs'; import { normalizeWechatName, resolveWechatAddressName } from './wechat/user/display-name.mjs'; import { buildAckText } from './wechat/ack/ack-provider.mjs'; +import { + composePassiveReplyWithDeferred, + createWechatCustomerServiceDeferredStore, + DEFERRED_DELIVERY_PRIORITY, + isWechatCustomerServiceQuotaError, + markWechatDeliveryDeferred, + wasWechatDeliveryDeferred, +} from './wechat/customer-service-deferred.mjs'; import { guardScheduleConfirmationReply } from './wechat/handlers/schedule-guard.mjs'; import { handleWechatScheduleIntent } from './wechat/handlers/schedule.mjs'; import { @@ -1554,6 +1562,7 @@ export function createWechatMpService({ repairFreshPageThumbnail, asrTarget: config.asrTarget || DEFAULT_ASR_TARGET, }; + const deferredStore = createWechatCustomerServiceDeferredStore({ mysqlPool, logger }); let accessTokenCache = { token: null, @@ -1784,7 +1793,13 @@ export function createWechatMpService({ openid, content, user = null, - { verifiedHtmlUrls = [], linkExistsForRequest = linkExists } = {}, + { + verifiedHtmlUrls = [], + linkExistsForRequest = linkExists, + deliveryPriority = 'formal_reply', + deferOnQuota = true, + sourceMsgId = null, + } = {}, ) => { const formatted = formatWechatOutboundText(sanitizeWechatAgentOutboundText(content), user); const verifiedUrlSet = new Set( @@ -1822,12 +1837,36 @@ export function createWechatMpService({ }, ), ); - if (Number(payload?.errcode ?? 0) !== 0) { - const errcode = Number(payload?.errcode ?? 0); + const errcode = Number(payload?.errcode ?? 0); + if (errcode !== 0) { const errmsg = String(payload?.errmsg ?? '').trim() || 'unknown_error'; + if (deferOnQuota && isWechatCustomerServiceQuotaError(errcode)) { + if (deliveryPriority === 'progress') { + logger.warn?.('WeChat MP progress reply skipped due to customer-service quota:', errcode); + return { sent: false, skipped: true, errcode }; + } + await deferredStore.upsertDeferred({ + appId: config.appId, + openid, + priority: deliveryPriority, + content: guarded, + metadata: { + verifiedHtmlUrls: [...verifiedUrlSet], + errcode, + }, + sourceMsgId, + }); + logger.warn?.('WeChat MP customer-service delivery deferred:', { + openid, + errcode, + deliveryPriority, + }); + return { sent: false, deferred: true, errcode }; + } throw new Error(`微信客服消息发送失败 errcode=${errcode} errmsg=${errmsg}`); } } + return { sent: true }; }; const sendCustomerServiceImage = async (openid, generatedImage) => { @@ -1861,12 +1900,24 @@ export function createWechatMpService({ return uploaded; }; + const sendWechatFailureNotice = async (openid, text, user, { sourceMsgId = null } = {}) => { + if (await deferredStore.hasFormalReplyPending({ appId: config.appId, openid })) { + return { sent: false, skipped: true, reason: 'formal_reply_pending' }; + } + return sendCustomerServiceText(openid, text, user, { + deliveryPriority: 'failure_notice', + sourceMsgId, + }); + }; + const sendTextToUser = async (userId, content) => { const openid = await userAuth.getWechatOpenidForUser(userId, config.appId); if (!openid) { throw new Error('用户尚未绑定服务号，无法推送提醒'); } - await sendCustomerServiceText(openid, content); + return sendCustomerServiceText(openid, content, null, { + deliveryPriority: 'formal_reply', + }); }; const enforceFreshPageThumbnailDelivery = async ({ @@ -2514,10 +2565,15 @@ export function createWechatMpService({ await rememberWechatUserContext(sessionId, user, { forceBootstrap: route.isNewSession }); let finished = false; let progressTimer = null; - if (config.progressDelayMs > 0 && config.progressText) { + const skipProgressReply = + wechatIntent.kind === 'page.generate' || htmlArtifactDeliveryExpected; + if (config.progressDelayMs > 0 && config.progressText && !skipProgressReply) { progressTimer = setTimeout(() => { if (finished) return; - sendCustomerServiceText(inbound.fromUserName, config.progressText, user).catch((err) => { + sendCustomerServiceText(inbound.fromUserName, config.progressText, user, { + deliveryPriority: 'progress', + deferOnQuota: false, + }).catch((err) => { logger.warn?.('WeChat MP progress reply failed:', err); }); }, config.progressDelayMs); @@ -2678,7 +2734,9 @@ export function createWechatMpService({ } const text = pageOutcome.failureText ?? buildPagePublishFailureText(); try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP page generate failure notice failed:', sendErr); } @@ -2699,7 +2757,9 @@ export function createWechatMpService({ if (htmlGenerationNeedsRetry) { const text = buildHtmlPublishFailureText(); try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP html publish failure notice failed:', sendErr); } @@ -2767,7 +2827,9 @@ export function createWechatMpService({ requestStartedAt, notifyFailure: async (text) => { try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP page data failure notice failed:', sendErr); } @@ -2800,10 +2862,24 @@ export function createWechatMpService({ } } scheduleWechatSessionSnapshotRefresh(sessionId, user.userId); - await sendCustomerServiceText(inbound.fromUserName, await guardScheduleReply(finalizedReply), user, { - verifiedHtmlUrls: publishArtifacts.map((artifact) => artifact.url), - linkExistsForRequest, - }); + const delivery = await sendCustomerServiceText( + inbound.fromUserName, + await guardScheduleReply(finalizedReply), + user, + { + verifiedHtmlUrls: publishArtifacts.map((artifact) => artifact.url), + linkExistsForRequest, + deliveryPriority: 'formal_reply', + sourceMsgId: intent.msgId, + }, + ); + if (delivery?.deferred) { + throw markWechatUserNotified( + markWechatDeliveryDeferred( + new Error(`delivery_deferred errcode=${delivery.errcode ?? 'unknown'}`), + ), + ); + } return { sessionId }; } catch (err) { if ( @@ -2834,7 +2910,9 @@ export function createWechatMpService({ if (!err?.wechatUserNotified) { const text = buildPageDataCollectFailureText(); try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP page data failure notice failed:', sendErr); } @@ -2852,7 +2930,9 @@ export function createWechatMpService({ const text = '我没能可靠确认“做成页面”指的是哪段内容，已经停止沿用旧主题。请把要做成页面的主题或原文再发一次。'; try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP contextual follow-up notice failed:', sendErr); } @@ -2986,7 +3066,9 @@ export function createWechatMpService({ if (pageOutcome.action === 'session_retry' || pageOutcome.action === 'fail') { const text = pageOutcome.failureText ?? buildPagePublishFailureText(); try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP page generate retry failure notice failed:', sendErr); } @@ -3004,7 +3086,9 @@ export function createWechatMpService({ if (htmlGenerationNeedsRetry) { const text = buildHtmlPublishFailureText(); try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP html publish failure notice failed:', sendErr); } @@ -3069,7 +3153,9 @@ export function createWechatMpService({ requestStartedAt: retryStartedAt, notifyFailure: async (text) => { try { - await sendCustomerServiceText(inbound.fromUserName, text, user); + await sendWechatFailureNotice(inbound.fromUserName, text, user, { + sourceMsgId: intent.msgId, + }); } catch (sendErr) { logger.error?.('WeChat MP page data retry failure notice failed:', sendErr); } @@ -3102,10 +3188,24 @@ export function createWechatMpService({ } } scheduleWechatSessionSnapshotRefresh(sessionId, user.userId); - await sendCustomerServiceText(inbound.fromUserName, await guardScheduleReply(finalizedReply), user, { - verifiedHtmlUrls: publishArtifacts.map((artifact) => artifact.url), - linkExistsForRequest, - }); + const retryDelivery = await sendCustomerServiceText( + inbound.fromUserName, + await guardScheduleReply(finalizedReply), + user, + { + verifiedHtmlUrls: publishArtifacts.map((artifact) => artifact.url), + linkExistsForRequest, + deliveryPriority: 'formal_reply', + sourceMsgId: intent.msgId, + }, + ); + if (retryDelivery?.deferred) { + throw markWechatUserNotified( + markWechatDeliveryDeferred( + new Error(`delivery_deferred errcode=${retryDelivery.errcode ?? 'unknown'}`), + ), + ); + } return { sessionId }; } throw err; @@ -3237,17 +3337,38 @@ export function createWechatMpService({ }), }; } + + let claimedDeferredRecord = undefined; + const buildPassiveReplyBody = async (content) => { + if (claimedDeferredRecord === undefined) { + claimedDeferredRecord = await deferredStore.claimDeferred({ + appId: config.appId, + openid: inbound.fromUserName, + }); + } + let merged = String(content ?? ''); + if (claimedDeferredRecord) { + if (claimedDeferredRecord.priority >= DEFERRED_DELIVERY_PRIORITY.formal_reply) { + merged = claimedDeferredRecord.content; + } else { + merged = composePassiveReplyWithDeferred(claimedDeferredRecord.content, merged); + } + } + const [firstChunk] = splitWechatText(merged); + return buildWechatTextReply({ + toUserName: inbound.fromUserName, + fromUserName: inbound.toUserName, + content: firstChunk ?? merged, + }); + }; + if (boundUser.status === 'disabled') { await persistIntentDetail({ intent, userId: boundUser.userId, rawXmlHash }); return { ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '当前账号不可用，请联系管理员处理。', - }), + body: await buildPassiveReplyBody('当前账号不可用，请联系管理员处理。'), }; } @@ -3263,11 +3384,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '当前账号尚未开启服务号文件分析灰度，请先通过 H5 上传文件。', - }), + body: await buildPassiveReplyBody('当前账号尚未开启服务号文件分析灰度，请先通过 H5 上传文件。'), }; } @@ -3289,11 +3406,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: config.unsupportedText, - }), + body: await buildPassiveReplyBody(config.unsupportedText), }; } @@ -3303,11 +3416,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '未识别到语音文字，请再说一次或输入文字。', - }), + body: await buildPassiveReplyBody('未识别到语音文字，请再说一次或输入文字。'), }; } @@ -3346,11 +3455,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '图片处理失败，请稍后重试或直接发送文字说明。', - }), + body: await buildPassiveReplyBody('图片处理失败，请稍后重试或直接发送文字说明。'), }; } } @@ -3397,11 +3502,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: error instanceof Error ? error.message : '文件处理失败，请稍后重试。', - }), + body: await buildPassiveReplyBody(error instanceof Error ? error.message : '文件处理失败，请稍后重试。'), }; } } @@ -3415,11 +3516,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '这次位置字段不完整，请重新发送位置或直接输入地点。', - }), + body: await buildPassiveReplyBody('这次位置字段不完整，请重新发送位置或直接输入地点。'), }; } @@ -3429,11 +3526,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '这次没有拿到完整链接，请重新发送一次。', - }), + body: await buildPassiveReplyBody('这次没有拿到完整链接，请重新发送一次。'), }; } @@ -3443,11 +3536,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: '已收到视频，当前先支持文本、语音、图片、定位和链接。', - }), + body: await buildPassiveReplyBody('已收到视频，当前先支持文本、语音、图片、定位和链接。'), }; } @@ -3463,11 +3552,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: syncReply, - }), + body: await buildPassiveReplyBody(syncReply), }; } } @@ -3484,11 +3569,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: buildStatusText(boundUser, config.statusText), - }), + body: await buildPassiveReplyBody(buildStatusText(boundUser, config.statusText)), }; } } @@ -3529,11 +3610,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: disclosureDecision.responseText, - }), + body: await buildPassiveReplyBody(disclosureDecision.responseText), }; } @@ -3568,11 +3645,7 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: scheduleReply, - }), + body: await buildPassiveReplyBody(scheduleReply), }; } @@ -3607,12 +3680,13 @@ export function createWechatMpService({ }).catch(() => {}); } logger.error?.('WeChat MP background reply failed:', err); - if (wasWechatUserNotified(err)) return; + if (wasWechatUserNotified(err) || wasWechatDeliveryDeferred(err)) return; try { - await sendCustomerServiceText( + await sendWechatFailureNotice( inbound.fromUserName, formatWechatAgentFailureMessage(err), boundUser, + { sourceMsgId: inbound.msgId }, ); } catch (sendErr) { logger.error?.('WeChat MP failure notice failed:', sendErr); @@ -3623,16 +3697,12 @@ export function createWechatMpService({ ok: true, status: 200, contentType: 'application/xml; charset=utf-8', - body: buildWechatTextReply({ - toUserName: inbound.fromUserName, - fromUserName: inbound.toUserName, - content: buildAckText({ - intent, - nickname: resolveWechatAddressName(boundUser), - config, - fallbackText: config.ackText, - }), - }), + body: await buildPassiveReplyBody(buildAckText({ + intent, + nickname: resolveWechatAddressName(boundUser), + config, + fallbackText: config.ackText, + })), task, }; }; diff --git a/wechat-mp.test.mjs b/wechat-mp.test.mjs index f1301d2..ab8994f 100644 --- a/wechat-mp.test.mjs +++ b/wechat-mp.test.mjs @@ -1220,7 +1220,7 @@ test('wechat mp service splits long agent replies into multiple customer message assert.equal(combined, longReply); }); -test('wechat mp service surfaces wechat customer-service errcode and errmsg', async () => { +test('wechat mp service defers 45047 customer-service delivery instead of throwing', async () => { const service = createBoundWechatService({ userAuth: { async getWechatOpenidForUser() { @@ -1241,10 +1241,62 @@ test('wechat mp service surfaces wechat customer-service errcode and errmsg', as }, }); - await assert.rejects( - service.sendTextToUser('user-1', 'hello'), - /errcode=45047.*out of response count limit/, + const result = await service.sendTextToUser('user-1', 'hello'); + assert.equal(result.deferred, true); + assert.equal(result.errcode, 45047); +}); + +test('wechat mp service claims deferred delivery on next passive response once', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + const service = createBoundWechatService({ + config: { + progressDelayMs: 0, + ackText: '已收到', + }, + userAuth: { + async getWechatOpenidForUser() { + return 'openid-1'; + }, + }, + wechatFetch: async (url) => { + if (String(url).includes('/cgi-bin/stable_token')) { + return new Response(JSON.stringify({ access_token: 'access-1', expires_in: 7200 }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + } + return new Response(JSON.stringify({ errcode: 45047, errmsg: 'out of response count limit rid: test-rid' }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }); + }, + }); + + const deferred = await service.sendTextToUser( + 'user-1', + '页面已生成：https://example.com/MindSpace/user-1/public/demo.html', ); + assert.equal(deferred.deferred, true); + assert.equal(deferred.errcode, 45047); + + const claimed = await service.handleInboundMessage(inboundXml({ content: '继续' }), { + timestamp, + nonce, + signature: signatureFor(token, timestamp, nonce), + }); + assert.equal(claimed.status, 200); + assert.match(String(claimed.body), /demo\.html/); + assert.doesNotMatch(String(claimed.body), /已收到/); + + const followUp = await service.handleInboundMessage(inboundXml({ content: '再来一条' }), { + timestamp, + nonce, + signature: signatureFor(token, timestamp, nonce), + }); + assert.doesNotMatch(String(followUp.body), /demo\.html/); + assert.ok(followUp.task); }); test('wechat mp service strips markdown emphasis around outbound links', async () => { diff --git a/wechat/customer-service-deferred.mjs b/wechat/customer-service-deferred.mjs new file mode 100644 index 0000000..914b39a --- /dev/null +++ b/wechat/customer-service-deferred.mjs @@ -0,0 +1,200 @@ +export const WECHAT_CUSTOMER_QUOTA_ERRCODES = new Set([45047, 45015]); + +export const DEFERRED_DELIVERY_PRIORITY = { + progress: 1, + failure_notice: 5, + formal_reply: 10, +}; + +export function resolveDeferredDeliveryPriority(priority) { + if (typeof priority === 'number' && Number.isFinite(priority)) return priority; + const key = String(priority ?? 'formal_reply').trim(); + return DEFERRED_DELIVERY_PRIORITY[key] ?? DEFERRED_DELIVERY_PRIORITY.formal_reply; +} + +export function isWechatCustomerServiceQuotaError(errcode) { + return WECHAT_CUSTOMER_QUOTA_ERRCODES.has(Number(errcode)); +} + +export function parseWechatCustomerServiceErrcode(err) { + const message = err instanceof Error ? err.message : String(err ?? ''); + const match = message.match(/errcode=(\d+)/); + return match ? Number(match[1]) : 0; +} + +function memoryKey(appId, openid) { + return `${String(appId)}:${String(openid)}`; +} + +export function createWechatCustomerServiceDeferredStore({ mysqlPool = null, logger = console } = {}) { + const memory = new Map(); + let schemaReady = false; + + async function ensureSchema() { + if (!mysqlPool?.query || schemaReady) return; + await mysqlPool.query(` + CREATE TABLE IF NOT EXISTS h5_wechat_mp_deferred_delivery ( + app_id VARCHAR(32) NOT NULL, + openid VARCHAR(64) NOT NULL, + priority TINYINT NOT NULL DEFAULT 10, + kind VARCHAR(32) NOT NULL DEFAULT 'text', + content MEDIUMTEXT NOT NULL, + metadata JSON NULL, + source_msg_id VARCHAR(128) NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + PRIMARY KEY (app_id, openid), + KEY idx_wechat_deferred_updated (updated_at) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + `); + schemaReady = true; + } + + async function upsertDeferred({ + appId, + openid, + priority = 'formal_reply', + kind = 'text', + content = '', + metadata = null, + sourceMsgId = null, + }) { + const normalizedContent = String(content ?? '').trim(); + if (!appId || !openid || !normalizedContent) { + return { stored: false, reason: 'invalid_input' }; + } + + const numericPriority = resolveDeferredDeliveryPriority(priority); + const existing = await peekDeferred({ appId, openid }); + if (existing && existing.priority > numericPriority) { + return { stored: false, reason: 'lower_priority', existing }; + } + + const now = Date.now(); + const record = { + appId: String(appId), + openid: String(openid), + priority: numericPriority, + kind: String(kind ?? 'text'), + content: normalizedContent, + metadata: metadata ?? null, + sourceMsgId: sourceMsgId ? String(sourceMsgId) : null, + createdAt: existing?.createdAt ?? now, + updatedAt: now, + }; + memory.set(memoryKey(appId, openid), record); + + if (mysqlPool?.query) { + try { + await ensureSchema(); + await mysqlPool.query( + `INSERT INTO h5_wechat_mp_deferred_delivery + (app_id, openid, priority, kind, content, metadata, source_msg_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + priority = VALUES(priority), + kind = VALUES(kind), + content = VALUES(content), + metadata = VALUES(metadata), + source_msg_id = VALUES(source_msg_id), + updated_at = VALUES(updated_at)`, + [ + record.appId, + record.openid, + record.priority, + record.kind, + record.content, + metadata ? JSON.stringify(metadata) : null, + record.sourceMsgId, + record.createdAt, + record.updatedAt, + ], + ); + } catch (err) { + logger.warn?.('WeChat deferred delivery persist failed:', err); + } + } + + return { stored: true, record }; + } + + async function peekDeferred({ appId, openid }) { + const key = memoryKey(appId, openid); + if (memory.has(key)) return memory.get(key) ?? null; + + if (!mysqlPool?.query) return null; + try { + await ensureSchema(); + const [rows] = await mysqlPool.query( + `SELECT app_id AS appId, openid, priority, kind, content, metadata, + source_msg_id AS sourceMsgId, created_at AS createdAt, updated_at AS updatedAt + FROM h5_wechat_mp_deferred_delivery + WHERE app_id = ? AND openid = ? LIMIT 1`, + [String(appId), String(openid)], + ); + const row = rows?.[0]; + if (!row) return null; + const record = { + ...row, + metadata: + typeof row.metadata === 'string' && row.metadata + ? JSON.parse(row.metadata) + : row.metadata ?? null, + }; + memory.set(key, record); + return record; + } catch (err) { + logger.warn?.('WeChat deferred delivery peek failed:', err); + return null; + } + } + + async function claimDeferred({ appId, openid }) { + const record = await peekDeferred({ appId, openid }); + if (!record) return null; + + memory.delete(memoryKey(appId, openid)); + if (mysqlPool?.query) { + try { + await ensureSchema(); + await mysqlPool.query( + `DELETE FROM h5_wechat_mp_deferred_delivery WHERE app_id = ? AND openid = ?`, + [String(appId), String(openid)], + ); + } catch (err) { + logger.warn?.('WeChat deferred delivery claim delete failed:', err); + } + } + return record; + } + + async function hasFormalReplyPending({ appId, openid }) { + const pending = await peekDeferred({ appId, openid }); + return Boolean(pending && pending.priority >= DEFERRED_DELIVERY_PRIORITY.formal_reply); + } + + return { + ensureSchema, + upsertDeferred, + peekDeferred, + claimDeferred, + hasFormalReplyPending, + }; +} + +export function markWechatDeliveryDeferred(err) { + if (err && typeof err === 'object') err.wechatDeliveryDeferred = true; + return err; +} + +export function wasWechatDeliveryDeferred(err) { + return Boolean(err && typeof err === 'object' && err.wechatDeliveryDeferred); +} + +export function composePassiveReplyWithDeferred(deferredContent, fallbackContent) { + const deferred = String(deferredContent ?? '').trim(); + const fallback = String(fallbackContent ?? '').trim(); + if (!deferred) return fallback; + if (!fallback) return deferred; + return `${deferred}\n\n${fallback}`; +} diff --git a/wechat/customer-service-deferred.test.mjs b/wechat/customer-service-deferred.test.mjs new file mode 100644 index 0000000..3972fa8 --- /dev/null +++ b/wechat/customer-service-deferred.test.mjs @@ -0,0 +1,69 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { + composePassiveReplyWithDeferred, + createWechatCustomerServiceDeferredStore, + DEFERRED_DELIVERY_PRIORITY, + isWechatCustomerServiceQuotaError, + resolveDeferredDeliveryPriority, +} from './customer-service-deferred.mjs'; + +test('customer-service deferred store keeps one formal reply per openid', async () => { + const store = createWechatCustomerServiceDeferredStore(); + const first = await store.upsertDeferred({ + appId: 'wx1', + openid: 'openid-1', + priority: 'formal_reply', + content: '页面链接 https://example.com/page.html', + }); + assert.equal(first.stored, true); + + const lower = await store.upsertDeferred({ + appId: 'wx1', + openid: 'openid-1', + priority: 'failure_notice', + content: '失败通知', + }); + assert.equal(lower.stored, false); + assert.equal(lower.reason, 'lower_priority'); + + const claimed = await store.claimDeferred({ appId: 'wx1', openid: 'openid-1' }); + assert.match(claimed.content, /page\.html/); + assert.equal(claimed.priority, DEFERRED_DELIVERY_PRIORITY.formal_reply); + + const empty = await store.claimDeferred({ appId: 'wx1', openid: 'openid-1' }); + assert.equal(empty, null); +}); + +test('customer-service deferred store replaces with higher priority content', async () => { + const store = createWechatCustomerServiceDeferredStore(); + await store.upsertDeferred({ + appId: 'wx1', + openid: 'openid-2', + priority: 'failure_notice', + content: '旧失败通知', + }); + await store.upsertDeferred({ + appId: 'wx1', + openid: 'openid-2', + priority: 'formal_reply', + content: '正式链接 https://example.com/new.html', + }); + const pending = await store.peekDeferred({ appId: 'wx1', openid: 'openid-2' }); + assert.equal(pending.priority, resolveDeferredDeliveryPriority('formal_reply')); + assert.match(pending.content, /new\.html/); +}); + +test('customer-service quota errcodes include 45047 and 45015', () => { + assert.equal(isWechatCustomerServiceQuotaError(45047), true); + assert.equal(isWechatCustomerServiceQuotaError(45015), true); + assert.equal(isWechatCustomerServiceQuotaError(40001), false); +}); + +test('composePassiveReplyWithDeferred merges fallback content', () => { + assert.equal( + composePassiveReplyWithDeferred('上一条结果', 'ack'), + '上一条结果\n\nack', + ); + assert.equal(composePassiveReplyWithDeferred('', 'ack'), 'ack'); +});