fix(whatsapp): complete history synchronization

2026-08-01 20:23:17 +08:00
parent cf25d71fcf
commit 785a5b196c
5 changed files with 271 additions and 18 deletions
@@ -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, {