From 785a5b196cad0879b98674bc864480844eb59110 Mon Sep 17 00:00:00 2001 From: wmq Date: Sat, 1 Aug 2026 20:23:17 +0800 Subject: [PATCH] fix(whatsapp): complete history synchronization --- CHANGELOG.md | 4 + server/src/index.js | 157 ++++++++++++++++++++++++++--- whatsapp-gateway/package-lock.json | 12 ++- whatsapp-gateway/package.json | 3 +- whatsapp-gateway/src/index.js | 113 ++++++++++++++++++++- 5 files changed, 271 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8dcd3c..9785c40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ - 将 Telegram Gateway 首次启动所需的 PostgreSQL 初始化脚本纳入版本控制，避免空数据库因挂载目录而启动失败。 - Telegram Gateway 本机调试端口改为 `127.0.0.1:3010`，避免与 Gitea 的 `3000` 端口冲突并阻止公网直连管理接口。 +- 修复 WhatsApp `FULL` 历史通知被 Baileys 默认过滤的问题；首次连接和历史阶段完成后自动执行逐会话分页补齐。 +- WhatsApp 历史同步增加并发合并、进度持久化、失败状态和平台数据库游标分页，避免重复同步及单次 2000 条截断。 +- 修复 WhatsApp 先于平台 API 连通时首次同步被静默丢弃的问题，增加延迟启动、指数重试和重复 Webhook 恢复。 +- WhatsApp 媒体与历史文件下载接入 Undici 代理 Dispatcher，并可恢复处理旧版本曾忽略但仍保存在缓存中的历史通知。 ## [1.1.0] - 2026-08-01 diff --git a/server/src/index.js b/server/src/index.js index dec2793..d76b8e7 100644 --- a/server/src/index.js +++ b/server/src/index.js @@ -56,6 +56,8 @@ function broadcast(tenantId, event) { const automaticTranslationQueues = { high: [], low: [] } const automaticTranslationKeys = new Set() const automaticTranslationCache = new Map() +const whatsappSyncStates = new Map() +const whatsappSyncSchedules = new Map() let activeAutomaticTranslations = 0 let activeLowPriorityTranslations = 0 let lowPriorityResumeAt = 0 @@ -3265,14 +3267,24 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} return { message, customer, conversation, isNew: !existingMessage } } -async function syncWhatsAppTenant(tenantId, channel, options = {}) { +async function performWhatsAppTenantSync(tenantId, channel, options = {}) { const state = await whatsappRequest(`/accounts/${channel.gatewayAccountId}/auth/state`) if (state.status !== 'connected' || !['open', 'ready'].includes(state.connectionState)) { throw fail(409, 'WhatsApp 账号尚未完成关联设备配对') } const chats = await whatsappRequest(`/accounts/${channel.gatewayAccountId}/chats`) let messagesSynced = 0 - for (const chat of chats || []) { + if (options.full === true) { + await patchResource(tenantId, 'channels', channel.id, { + historySyncStatus: 'running', + historySyncStartedAt: new Date().toISOString(), + historySyncChatsCompleted: 0, + historySyncChatsTotal: chats?.length || 0, + historySyncMessagesSynced: 0, + historySyncError: undefined, + }) + } + for (const [chatIndex, chat] of (chats || []).entries()) { const jid = String(chat.id || '') if (!jid) continue if (options.full === true) { @@ -3280,20 +3292,28 @@ async function syncWhatsAppTenant(tenantId, channel, options = {}) { `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history/sync`, { method: 'POST', - body: JSON.stringify({ pages: 20, count: 50 }), + body: JSON.stringify({ pages: 50, count: 50 }), }, ).catch(() => null) } const identity = await whatsappRequest( `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`, ).catch(() => ({ name: chat.name || jid.split('@')[0] })) - const history = await whatsappRequest( - `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=2000`, - ).catch(() => ({ messages: [] })) - for (const raw of history.messages || []) { - const result = await upsertWhatsAppMessage(tenantId, channel, raw, { identity, notify: false }) - if (result?.isNew) messagesSynced++ - } + let historyCursor = '' + let historyPages = 0 + do { + const history = await whatsappRequest( + `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=2000${historyCursor ? `&before=${encodeURIComponent(historyCursor)}` : ''}`, + ).catch(() => ({ messages: [], beforeCursor: null, hasMoreBefore: false })) + for (const raw of history.messages || []) { + const result = await upsertWhatsAppMessage(tenantId, channel, raw, { identity, notify: false }) + if (result?.isNew) messagesSynced++ + } + historyPages++ + historyCursor = options.full === true && history.hasMoreBefore && history.beforeCursor + ? String(history.beforeCursor) + : '' + } while (historyCursor && historyPages < 100) const conversationId = `${channel.id}:${jid}` const current = await getResource(tenantId, 'conversations', conversationId) if (current) { @@ -3302,12 +3322,27 @@ async function syncWhatsAppTenant(tenantId, channel, options = {}) { historySyncedAt: new Date().toISOString(), }) } + if (options.full === true) { + await patchResource(tenantId, 'channels', channel.id, { + historySyncChatsCompleted: chatIndex + 1, + historySyncChatsTotal: chats.length, + historySyncMessagesSynced: messagesSynced, + historySyncUpdatedAt: new Date().toISOString(), + }) + } } await patchResource(tenantId, 'channels', channel.id, { lastSyncAt: new Date().toISOString(), status: 'connected', authState: 'connected', errorMessage: undefined, + ...(options.full === true ? { + historySyncStatus: 'completed', + historySyncCompletedAt: new Date().toISOString(), + historySyncChatsCompleted: chats?.length || 0, + historySyncChatsTotal: chats?.length || 0, + historySyncMessagesSynced: messagesSynced, + } : {}), }) await appendChannelLog( tenantId, @@ -3323,6 +3358,76 @@ async function syncWhatsAppTenant(tenantId, channel, options = {}) { return { chats: chats?.length || 0, messages: messagesSynced } } +async function syncWhatsAppTenant(tenantId, channel, options = {}) { + const key = `${tenantId}:${channel.id}` + const existing = whatsappSyncStates.get(key) + if (existing) { + if (options.full === true && !existing.runningFull) existing.fullRequested = true + return existing.promise + } + + const syncState = { + fullRequested: options.full === true, + runningFull: false, + promise: null, + } + syncState.promise = (async () => { + let result = { chats: 0, messages: 0 } + try { + do { + const full = syncState.fullRequested + syncState.fullRequested = false + syncState.runningFull = full + result = await performWhatsAppTenantSync(tenantId, channel, { full }) + syncState.runningFull = false + } while (syncState.fullRequested) + return result + } catch (error) { + if (syncState.runningFull) { + await patchResource(tenantId, 'channels', channel.id, { + historySyncStatus: 'failed', + historySyncError: String(error?.message || error), + historySyncUpdatedAt: new Date().toISOString(), + }).catch(() => null) + } + throw error + } finally { + whatsappSyncStates.delete(key) + } + })() + whatsappSyncStates.set(key, syncState) + return syncState.promise +} + +function scheduleWhatsAppSync(tenantId, channel, options = {}, attempt = 0) { + const key = `${tenantId}:${channel.id}` + const scheduled = whatsappSyncSchedules.get(key) + if (scheduled) { + if (options.full === true) scheduled.full = true + return + } + const state = { full: options.full === true, timer: null } + const delay = attempt === 0 ? 2_000 : Math.min(60_000, 2_000 * (2 ** attempt)) + state.timer = setTimeout(async () => { + whatsappSyncSchedules.delete(key) + try { + await syncWhatsAppTenant(tenantId, channel, { full: state.full }) + } catch (error) { + if (attempt < 6) { + scheduleWhatsAppSync(tenantId, channel, { full: state.full }, attempt + 1) + } else { + await patchResource(tenantId, 'channels', channel.id, { + historySyncStatus: 'failed', + historySyncError: String(error?.message || error), + historySyncUpdatedAt: new Date().toISOString(), + }).catch(() => null) + } + } + }, delay) + state.timer.unref?.() + whatsappSyncSchedules.set(key, state) +} + async function syncTelegramTenant(tenantId, channel) { const gateway = await gatewayRequest(`/tg-accounts/${channel.gatewayAccountId}/chats/sync`) const chats = await gatewayRequest(`/tg-accounts/${channel.gatewayAccountId}/chats?limit=100`) @@ -4530,6 +4635,14 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) => || crypto.createHash('sha256').update(req.rawBody || Buffer.from(JSON.stringify(req.body))).digest('hex'), ) if (await getResource(tenantId, 'webhook-events', eventId)) { + if (req.body.event === 'account.connected') { + scheduleWhatsAppSync(tenantId, channel, { full: true }) + } else if (req.body.event === 'history.synced') { + const historyReady = req.body?.data?.isLatest === true || Number(req.body?.data?.progress) === 100 + scheduleWhatsAppSync(tenantId, channel, { full: historyReady }) + } else if (req.body.event === 'history.status' && req.body?.data?.status === 'complete') { + scheduleWhatsAppSync(tenantId, channel, { full: true }) + } return ok(res, { accepted: true, duplicate: true }) } await putResource(tenantId, 'webhook-events', { @@ -4683,9 +4796,25 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) => createdAt: new Date().toISOString(), }) } - if (connected) void syncWhatsAppTenant(tenantId, channel).catch(() => null) + if (connected) scheduleWhatsAppSync(tenantId, channel, { full: true }) } else if (req.body.event === 'history.synced') { - void syncWhatsAppTenant(tenantId, channel).catch(() => null) + const historyReady = req.body?.data?.isLatest === true || Number(req.body?.data?.progress) === 100 + scheduleWhatsAppSync(tenantId, channel, { full: historyReady }) + } else if (req.body.event === 'history.status') { + await patchResource(tenantId, 'channels', channelId, { + gatewayHistoryStatus: req.body?.data?.status, + gatewayHistoryProgress: req.body?.data?.progress, + gatewayHistorySyncType: req.body?.data?.syncType, + gatewayHistoryUpdatedAt: req.body?.data?.updatedAt || new Date().toISOString(), + }) + broadcast(tenantId, { + type: 'channel.history.updated', + channelId, + history: req.body.data, + }) + if (req.body?.data?.status === 'complete') { + scheduleWhatsAppSync(tenantId, channel, { full: true }) + } } ok(res, { accepted: true }) }) @@ -4779,7 +4908,9 @@ for (const row of existingWhatsAppChannels.rows) { const tenantId = String(row.tenant_id) await ensureWhatsAppWebhook(tenantId, row.data, row.data.status !== 'paused').then(async (updatedChannel) => { if (row.data.status !== 'paused') { - void syncWhatsAppTenant(tenantId, updatedChannel).catch(() => null) + scheduleWhatsAppSync(tenantId, updatedChannel, { + full: row.data.historySyncStatus !== 'completed', + }) } }).catch(async (error) => { await patchResource(tenantId, 'channels', row.data.id, { diff --git a/whatsapp-gateway/package-lock.json b/whatsapp-gateway/package-lock.json index a729ffb..09fefba 100644 --- a/whatsapp-gateway/package-lock.json +++ b/whatsapp-gateway/package-lock.json @@ -15,7 +15,8 @@ "https-proxy-agent": "^7.0.6", "multer": "2.2.0", "pino": "9.6.0", - "qrcode": "1.5.4" + "qrcode": "1.5.4", + "undici": "^6.21.3" } }, "node_modules/@borewit/text-codec": { @@ -2624,6 +2625,15 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/undici": { + "version": "6.21.3", + "resolved": "https://registry.npmmirror.com/undici/-/undici-6.21.3.tgz", + "integrity": "sha512-gBLkYIlEnSp8pFbT64yFgGE6UIB9tAkhukC23PmMDCe5Nd+cRqKxSjw5y54MK2AZMgZfJWMaNE4nYUHgi1XEOw==", + "license": "MIT", + "engines": { + "node": ">=18.17" + } + }, "node_modules/undici-types": { "version": "8.3.0", "resolved": "https://registry.npmmirror.com/undici-types/-/undici-types-8.3.0.tgz", diff --git a/whatsapp-gateway/package.json b/whatsapp-gateway/package.json index 7468bb4..1659dc9 100644 --- a/whatsapp-gateway/package.json +++ b/whatsapp-gateway/package.json @@ -17,6 +17,7 @@ "https-proxy-agent": "^7.0.6", "multer": "2.2.0", "pino": "9.6.0", - "qrcode": "1.5.4" + "qrcode": "1.5.4", + "undici": "^6.21.3" } } diff --git a/whatsapp-gateway/src/index.js b/whatsapp-gateway/src/index.js index 3bd4aac..0494298 100644 --- a/whatsapp-gateway/src/index.js +++ b/whatsapp-gateway/src/index.js @@ -9,10 +9,12 @@ import path from 'node:path' import https from 'node:https' import { Boom } from '@hapi/boom' import { HttpsProxyAgent } from 'https-proxy-agent' +import { ProxyAgent as UndiciProxyAgent } from 'undici' import makeWASocket, { Browsers, BufferJSON, DisconnectReason, + downloadAndProcessHistorySyncNotification, downloadMediaMessage, fetchLatestBaileysVersion, getContentType, @@ -308,6 +310,61 @@ function upsertMessage(accountId, raw) { return message } +async function replayStoredHistoryNotifications( + accountId, account, store, logger, processedIds = new Set(), downloadOptions = {}, +) { + const replayed = new Set(account.replayedHistoryNotificationIds || []) + const candidates = [] + for (const messages of store.messages.values()) { + for (const item of messages || []) { + const notification = item.raw?.message?.protocolMessage?.historySyncNotification + if (!notification || !item.id || replayed.has(item.id) || processedIds.has(item.id)) continue + candidates.push({ id: item.id, notification }) + } + } + for (const candidate of candidates) { + try { + const data = await downloadAndProcessHistorySyncNotification( + candidate.notification, downloadOptions, logger, + ) + for (const chat of data.chats || []) { + const id = String(chat.id || '') + if (id) store.chats.set(id, { ...(store.chats.get(id) || {}), ...jsonSafe(chat), id }) + } + for (const contact of data.contacts || []) { + const id = String(contact.id || '') + if (id) store.contacts.set(id, { ...(store.contacts.get(id) || {}), ...jsonSafe(contact), id }) + } + for (const message of data.messages || []) upsertMessage(accountId, message) + replayed.add(candidate.id) + account.replayedHistoryNotificationIds = [...replayed].slice(-100) + account.historySync = { + status: 'processing', + syncType: data.syncType ?? candidate.notification.syncType ?? null, + progress: data.progress ?? candidate.notification.progress ?? null, + chats: data.chats?.length || 0, + contacts: data.contacts?.length || 0, + messages: data.messages?.length || 0, + recovered: true, + updatedAt: new Date().toISOString(), + } + await saveAccounts() + scheduleStoreSave(accountId) + void deliverWebhook(account, 'history.synced', { + chats: data.chats?.length || 0, + contacts: data.contacts?.length || 0, + messages: data.messages?.length || 0, + progress: data.progress ?? candidate.notification.progress ?? null, + syncType: data.syncType ?? candidate.notification.syncType ?? null, + recovered: true, + isLatest: true, + }) + } catch (error) { + logger.warn({ err: error, messageId: candidate.id }, 'stored history notification replay failed') + } + } +} + function storedMessage(accountId, jid, messageId) { const store = stores.get(accountId) return (store?.messages.get(jid) || []).find((item) => item.id === messageId) @@ -672,10 +729,13 @@ async function connectAccount(accountId) { const authPath = path.join(authDir, accountId) await fs.mkdir(authPath, { recursive: true, mode: 0o700 }) const { state, saveCreds } = await useMultiFileAuthState(authPath) - const { version } = await fetchLatestBaileysVersion() const logger = log.child({ accountId }) const proxyUrl = accountProxyUrl(account) const proxyAgent = proxyUrl ? new HttpsProxyAgent(proxyUrl) : undefined + const fetchDispatcher = proxyUrl ? new UndiciProxyAgent(proxyUrl) : undefined + const { version } = await fetchLatestBaileysVersion( + fetchDispatcher ? { dispatcher: fetchDispatcher } : {}, + ) const sock = makeWASocket({ version, logger, @@ -685,7 +745,11 @@ async function connectAccount(accountId) { }, browser: Browsers.ubuntu('Lingxi Crossborder'), ...(proxyAgent ? { agent: proxyAgent, fetchAgent: proxyAgent } : {}), + ...(fetchDispatcher ? { options: { dispatcher: fetchDispatcher } } : {}), syncFullHistory: true, + // Baileys 7 excludes FULL history notifications by default even when + // syncFullHistory is enabled. Accept every server-provided history phase. + shouldSyncHistoryMessage: () => true, markOnlineOnConnect: false, generateHighQualityLinkPreview: true, getMessage: async (key) => { @@ -694,7 +758,7 @@ async function connectAccount(accountId) { }, cachedGroupMetadata: async (jid) => store.groups.get(jid), }) - sessions.set(accountId, { sock, connecting: true }) + sessions.set(accountId, { sock, connecting: true, fetchDispatcher }) account.status = state.creds.registered ? 'connecting' : 'waiting_pairing' account.updatedAt = new Date().toISOString() await saveAccounts() @@ -722,6 +786,18 @@ async function connectAccount(accountId) { const session = sessions.get(accountId) if (session) session.connecting = false await saveAccounts() + const processedIds = new Set( + (state.creds.processedHistoryMessages || []).map((item) => String(item?.key?.id || '')), + ) + void replayStoredHistoryNotifications( + accountId, + account, + store, + logger, + processedIds, + fetchDispatcher ? { dispatcher: fetchDispatcher } : {}, + ) + .catch((error) => logger.warn({ err: error }, 'stored history recovery failed')) for (const job of outboundJobs.values()) { if (job.accountId === accountId && ['queued', 'retrying', 'sending'].includes(job.status)) { scheduleOutboundJob(job.id) @@ -733,7 +809,9 @@ async function connectAccount(accountId) { if (sessions.get(accountId)?.sock !== sock) return const statusCode = new Boom(update.lastDisconnect?.error).output?.statusCode const loggedOut = statusCode === DisconnectReason.loggedOut + const session = sessions.get(accountId) sessions.delete(accountId) + if (session?.fetchDispatcher) void session.fetchDispatcher.close().catch(() => undefined) account.status = loggedOut ? 'logged_out' : 'disconnected' account.connectionState = loggedOut ? 'logged_out' : 'reconnecting' account.lastError = update.lastDisconnect?.error?.message || `连接关闭 (${statusCode || 'unknown'})` @@ -753,7 +831,9 @@ async function connectAccount(accountId) { } } }) - sock.ev.on('messaging-history.set', ({ chats = [], contacts = [], messages = [] }) => { + sock.ev.on('messaging-history.set', ({ + chats = [], contacts = [], messages = [], isLatest, progress, syncType, chunkOrder, + }) => { for (const chat of chats) { const id = String(chat.id || '') if (id) store.chats.set(id, { ...(store.chats.get(id) || {}), ...jsonSafe(chat), id }) @@ -763,13 +843,40 @@ async function connectAccount(accountId) { if (id) store.contacts.set(id, { ...(store.contacts.get(id) || {}), ...jsonSafe(contact), id }) } for (const item of messages) upsertMessage(accountId, item) + account.historySync = { + status: isLatest || progress === 100 ? 'processing' : 'syncing', + syncType: syncType ?? null, + progress: progress ?? null, + chunkOrder: chunkOrder ?? null, + chats: chats.length, + contacts: contacts.length, + messages: messages.length, + updatedAt: new Date().toISOString(), + } + void saveAccounts() scheduleStoreSave(accountId) void deliverWebhook(account, 'history.synced', { chats: chats.length, contacts: contacts.length, messages: messages.length, + isLatest: Boolean(isLatest), + progress: progress ?? null, + syncType: syncType ?? null, + chunkOrder: chunkOrder ?? null, }) }) + sock.ev.on('messaging-history.status', ({ syncType, status, explicit }) => { + account.historySync = { + ...(account.historySync || {}), + status, + syncType, + explicit: Boolean(explicit), + updatedAt: new Date().toISOString(), + ...(status === 'complete' ? { completedAt: new Date().toISOString() } : {}), + } + void saveAccounts() + void deliverWebhook(account, 'history.status', account.historySync) + }) sock.ev.on('chats.upsert', (items) => { for (const chat of items || []) { const id = String(chat.id || '')