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 <cursoragent@cursor.com>

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}`;
+}

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');
+});