wmq/lingxi-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(whatsapp): preserve history across reauthorization

Browse Source
This commit is contained in:
wmq
2026-08-01 23:13:28 +08:00
parent 785a5b196c
commit 00b0ebb775
5 changed files with 210 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/index.js
+139 -11
View File
@@ -18,6 +18,12 @@ import { conversationContext, deepseek } from './ai.js'
import { countBy, dateKey, percent, telegramAttachment } from './business.js'
import { resolveCustomerCountry } from './country.js'
import { knowledgeMetadata, rankKnowledge } from './knowledge.js'
import {
isWeakWhatsAppDisplayName,
reusableWhatsAppConversations,
whatsappConversationActivityAt,
whatsappCustomerStableId,
} from './whatsapp-identity.js'
import {
decryptEmailPassword,
encryptEmailPassword,
@@ -1475,18 +1481,27 @@ app.post('/api/conversations/whatsapp/open', async (req, res) => {
const identity = await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
)
const customerId = `wa-${channel.gatewayAccountId}-${jid}`
const conversationId = `${channel.id}:${jid}`
const existingCustomer = await resolveWhatsAppCustomer(
req.auth.tenantId,
channel,
jid,
conversationId,
)
const customerId = existingCustomer?.id || whatsappCustomerStableId(jid)
const customer = await putResource(req.auth.tenantId, 'customers', withDefaults('customers', {
...(existingCustomer || {}),
id: customerId,
name: identity.name || jid.split('@')[0],
phone: identity.phone || '',
avatarUrl: identity.avatarUrl || undefined,
name: !isWeakWhatsAppDisplayName(existingCustomer?.name, jid)
? existingCustomer.name
: identity.name || jid.split('@')[0],
phone: identity.phone || existingCustomer?.phone || '',
avatarUrl: identity.avatarUrl || existingCustomer?.avatarUrl || undefined,
whatsapp: jid,
whatsappChatType: identity.chatType || 'private',
channel: 'whatsapp',
source: 'whatsapp',
}))
const conversationId = `${channel.id}:${jid}`
await deleteResource(req.auth.tenantId, 'conversation-tombstones', conversationId)
const previous = await getResource(req.auth.tenantId, 'conversations', conversationId)
const conversation = await putResource(req.auth.tenantId, 'conversations', {
@@ -3146,6 +3161,32 @@ function whatsappMessageType(type) {
return 'text'
}
async function resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) {
const current = await getResource(tenantId, 'conversations', conversationId)
if (current?.customerId) {
const customer = await getResource(tenantId, 'customers', current.customerId)
if (customer) return customer
}
const [channels, conversations] = await Promise.all([
listResources(tenantId, 'channels'),
listResources(tenantId, 'conversations'),
])
const activeChannelIds = new Set(channels.map((item) => String(item.id)))
const reusable = reusableWhatsAppConversations(
conversations,
activeChannelIds,
channel.id,
jid,
)
for (const conversation of reusable) {
if (!conversation.customerId) continue
const customer = await getResource(tenantId, 'customers', conversation.customerId)
if (customer) return customer
}
return getResource(tenantId, 'customers', whatsappCustomerStableId(jid))
}
async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}) {
const jid = String(rawMessage.chatId || '')
if (!jid || !rawMessage.id) return null
@@ -3172,15 +3213,19 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
})
return { message, reaction: rawMessage, isNew: false }
}
const customerId = `wa-${channel.gatewayAccountId}-${jid}`
const existingCustomer = await getResource(tenantId, 'customers', customerId)
const existingCustomer = await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId)
const customerId = existingCustomer?.id || whatsappCustomerStableId(jid)
const identity = options.identity || await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
).catch(() => null)
const customer = await putResource(tenantId, 'customers', withDefaults('customers', {
...(existingCustomer || {}),
id: customerId,
name: String(identity?.name || rawMessage.pushName || existingCustomer?.name || jid.split('@')[0]),
name: String(
!isWeakWhatsAppDisplayName(existingCustomer?.name, jid)
? existingCustomer.name
: identity?.name || rawMessage.pushName || existingCustomer?.name || jid.split('@')[0],
),
phone: String(identity?.phone || existingCustomer?.phone || ''),
avatarUrl: identity?.avatarUrl || existingCustomer?.avatarUrl,
whatsapp: jid,
@@ -3267,6 +3312,85 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
return { message, customer, conversation, isNew: !existingMessage }
}
async function mergeLegacyWhatsAppConversations(tenantId, channel, jid) {
const targetId = `${channel.id}:${jid}`
const target = await getResource(tenantId, 'conversations', targetId)
if (!target) return 0
const [channels, conversations] = await Promise.all([
listResources(tenantId, 'channels'),
listResources(tenantId, 'conversations'),
])
const activeChannelIds = new Set(channels.map((item) => String(item.id)))
const legacy = conversations.filter((item) => (
item.channel === 'whatsapp'
&& String(item.whatsappJid || '') === jid
&& String(item.id) !== targetId
&& !activeChannelIds.has(String(item.channelId || String(item.id).split(':')[0]))
))
if (!legacy.length) return 0
let imported = 0
let earliestCreatedAt = target.createdAt
let lastMessage = target.lastMessage
let customerId = target.customerId
let canonicalActivityAt = whatsappConversationActivityAt(target)
const mergedCustomerIds = new Set(target.customerId ? [target.customerId] : [])
const mergedIds = new Set(target.mergedLegacyConversationIds || [])
for (const source of legacy) {
if (source.customerId) mergedCustomerIds.add(source.customerId)
const sourceActivityAt = whatsappConversationActivityAt(source)
if (source.customerId && sourceActivityAt >= canonicalActivityAt) {
customerId = source.customerId
canonicalActivityAt = sourceActivityAt
}
const messages = await listResources(tenantId, `messages:${source.id}`)
for (const message of messages) {
const exists = await getResource(tenantId, `messages:${targetId}`, message.id)
await putResource(tenantId, `messages:${targetId}`, {
...message,
conversationId: targetId,
})
if (!exists) imported++
if (!lastMessage || Date.parse(message.createdAt || '') > Date.parse(lastMessage.createdAt || '')) {
lastMessage = { ...message, conversationId: targetId }
}
}
if (!earliestCreatedAt || Date.parse(source.createdAt || '') < Date.parse(earliestCreatedAt)) {
earliestCreatedAt = source.createdAt
}
mergedIds.add(source.id)
for (const message of messages) {
await deleteResource(tenantId, `messages:${source.id}`, message.id)
}
await deleteResource(tenantId, 'conversations', source.id)
}
await patchResource(tenantId, 'conversations', targetId, {
customerId,
createdAt: earliestCreatedAt || target.createdAt,
lastMessage,
mergedLegacyConversationIds: [...mergedIds],
historySyncedAt: new Date().toISOString(),
})
for (const duplicateCustomerId of mergedCustomerIds) {
if (!duplicateCustomerId || duplicateCustomerId === customerId) continue
for (const kind of ['opportunities', 'tasks', 'risks']) {
const resources = await listResources(tenantId, kind)
for (const resource of resources.filter((item) => item.customerId === duplicateCustomerId)) {
await patchResource(tenantId, kind, resource.id, { customerId })
}
}
const remainingConversations = (await listResources(tenantId, 'conversations'))
.some((item) => item.customerId === duplicateCustomerId)
const remainingBusinessReferences = (await Promise.all(
['opportunities', 'tasks', 'risks'].map((kind) => listResources(tenantId, kind)),
)).some((items) => items.some((item) => item.customerId === duplicateCustomerId))
if (!remainingConversations && !remainingBusinessReferences) {
await deleteResource(tenantId, 'customers', duplicateCustomerId)
}
}
return imported
}
async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
const state = await whatsappRequest(`/accounts/${channel.gatewayAccountId}/auth/state`)
if (state.status !== 'connected' || !['open', 'ready'].includes(state.connectionState)) {
@@ -3314,6 +3438,7 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
? String(history.beforeCursor)
: ''
} while (historyCursor && historyPages < 100)
messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid)
const conversationId = `${channel.id}:${jid}`
const current = await getResource(tenantId, 'conversations', conversationId)
if (current) {
@@ -4757,14 +4882,17 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
for (const contact of req.body?.data?.contacts || []) {
const jid = String(contact.id || '')
if (!jid) continue
const customerId = `wa-${channel.gatewayAccountId}-${jid}`
const existing = await getResource(tenantId, 'customers', customerId)
const conversationId = `${channel.id}:${jid}`
const existing = await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId)
if (!existing) continue
const customerId = existing.id
const identity = await whatsappRequest(
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
).catch(() => null)
const customer = await patchResource(tenantId, 'customers', customerId, {
name: identity?.name || contact.name || contact.notify || contact.verifiedName || existing.name,
name: !isWeakWhatsAppDisplayName(existing.name, jid)
? existing.name
: identity?.name || contact.name || contact.notify || contact.verifiedName || existing.name,
phone: identity?.phone || existing.phone,
avatarUrl: identity?.avatarUrl || existing.avatarUrl,
whatsappVerifiedName: contact.verifiedName || existing.whatsappVerifiedName,