feat(reliability): close messaging gateway P0 gaps
This commit is contained in:
wmq
+365
-34
@@ -18,6 +18,7 @@ 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 { isTimestampOlderThan, webhookEventClaimDecision } from './reliability.js'
|
||||
import {
|
||||
isWeakWhatsAppDisplayName,
|
||||
reusableWhatsAppConversations,
|
||||
@@ -3217,6 +3218,7 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}
|
||||
const customerId = existingCustomer?.id || whatsappCustomerStableId(jid)
|
||||
const identity = options.identity || await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
|
||||
{ signal: AbortSignal.timeout(8_000) },
|
||||
).catch(() => null)
|
||||
const customer = await putResource(tenantId, 'customers', withDefaults('customers', {
|
||||
...(existingCustomer || {}),
|
||||
@@ -3411,6 +3413,9 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
|
||||
for (const [chatIndex, chat] of (chats || []).entries()) {
|
||||
const jid = String(chat.id || '')
|
||||
if (!jid) continue
|
||||
const conversationId = `${channel.id}:${jid}`
|
||||
const currentBeforeSync = await getResource(tenantId, 'conversations', conversationId)
|
||||
const profileDue = options.refreshProfiles === true || !currentBeforeSync?.customerId
|
||||
if (options.full === true) {
|
||||
await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history/sync`,
|
||||
@@ -3420,30 +3425,64 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) {
|
||||
},
|
||||
).catch(() => null)
|
||||
}
|
||||
const identity = await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
|
||||
).catch(() => ({ name: chat.name || jid.split('@')[0] }))
|
||||
let historyCursor = ''
|
||||
const currentCustomer = currentBeforeSync?.customerId
|
||||
? await getResource(tenantId, 'customers', currentBeforeSync.customerId)
|
||||
: null
|
||||
const identity = profileDue || !currentCustomer
|
||||
? await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`,
|
||||
{ signal: AbortSignal.timeout(8_000) },
|
||||
).catch(() => ({ name: chat.name || jid.split('@')[0] }))
|
||||
: {
|
||||
name: currentCustomer.name,
|
||||
phone: currentCustomer.phone,
|
||||
avatarUrl: currentCustomer.avatarUrl,
|
||||
chatType: currentBeforeSync.whatsappChatType,
|
||||
}
|
||||
if (profileDue && currentCustomer) {
|
||||
const customer = await patchResource(tenantId, 'customers', currentCustomer.id, {
|
||||
name: !isWeakWhatsAppDisplayName(currentCustomer.name, jid)
|
||||
? currentCustomer.name
|
||||
: identity?.name || currentCustomer.name,
|
||||
phone: identity?.phone || currentCustomer.phone,
|
||||
avatarUrl: identity?.avatarUrl || currentCustomer.avatarUrl,
|
||||
whatsappChatType: identity?.chatType || currentCustomer.whatsappChatType,
|
||||
profileSyncedAt: new Date().toISOString(),
|
||||
})
|
||||
broadcast(tenantId, { type: 'customer.updated', customerId: currentCustomer.id, customer })
|
||||
}
|
||||
let historyCursor = options.full === true
|
||||
? ''
|
||||
: String(currentBeforeSync?.whatsappReconcileCursor || currentBeforeSync?.lastMessage?.whatsappMessageId || '')
|
||||
let historyPages = 0
|
||||
let latestProcessedMessageId = historyCursor
|
||||
do {
|
||||
const cursorQuery = historyCursor
|
||||
? options.full === true
|
||||
? `&before=${encodeURIComponent(historyCursor)}`
|
||||
: `&after=${encodeURIComponent(historyCursor)}`
|
||||
: ''
|
||||
const history = await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=2000${historyCursor ? `&before=${encodeURIComponent(historyCursor)}` : ''}`,
|
||||
`/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=${options.full === true ? 2000 : 500}${cursorQuery}`,
|
||||
).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++
|
||||
latestProcessedMessageId = String(raw.id || latestProcessedMessageId)
|
||||
}
|
||||
historyPages++
|
||||
historyCursor = options.full === true && history.hasMoreBefore && history.beforeCursor
|
||||
? String(history.beforeCursor)
|
||||
: ''
|
||||
} while (historyCursor && historyPages < 100)
|
||||
historyCursor = options.full === true
|
||||
? history.hasMoreBefore && history.beforeCursor ? String(history.beforeCursor) : ''
|
||||
: history.hasMoreAfter && history.afterCursor ? String(history.afterCursor) : ''
|
||||
} while (historyCursor && historyPages < (options.full === true ? 100 : 20))
|
||||
messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid)
|
||||
const conversationId = `${channel.id}:${jid}`
|
||||
const current = await getResource(tenantId, 'conversations', conversationId)
|
||||
if (current) {
|
||||
await patchResource(tenantId, 'conversations', conversationId, {
|
||||
unreadCount: Number(chat.unreadCount || current.unreadCount || 0),
|
||||
whatsappReconcileCursor: latestProcessedMessageId || current.whatsappReconcileCursor,
|
||||
profileSyncedAt: profileDue ? new Date().toISOString() : current.profileSyncedAt,
|
||||
lastReconciledAt: new Date().toISOString(),
|
||||
historySyncedAt: new Date().toISOString(),
|
||||
})
|
||||
}
|
||||
@@ -3553,17 +3592,46 @@ function scheduleWhatsAppSync(tenantId, channel, options = {}, attempt = 0) {
|
||||
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`)
|
||||
async function syncTelegramTenant(tenantId, channel, options = {}) {
|
||||
const full = options.full === true
|
||||
const refreshProfiles = options.refreshProfiles === true
|
||||
const gateway = full
|
||||
? await gatewayRequest(`/tg-accounts/${channel.gatewayAccountId}/chats/sync`)
|
||||
: { incremental: true }
|
||||
const chats = []
|
||||
for (let offset = 0; offset < 100_000; offset += 200) {
|
||||
const page = await gatewayRequest(
|
||||
`/tg-accounts/${channel.gatewayAccountId}/chats?limit=200&offset=${offset}`,
|
||||
)
|
||||
chats.push(...(page || []))
|
||||
if (!page || page.length < 200) break
|
||||
}
|
||||
let messagesSynced = 0
|
||||
for (const chat of chats || []) {
|
||||
const chatId = String(chat.tgChatId)
|
||||
const conversationId = `${channel.id}:${chatId}`
|
||||
if (await getResource(tenantId, 'conversation-tombstones', conversationId)) continue
|
||||
const existingConversation = await getResource(tenantId, 'conversations', conversationId)
|
||||
const gatewayLastMessageId = String(chat.lastMessageId || '')
|
||||
const profileDue = refreshProfiles || !existingConversation?.customerId
|
||||
if (
|
||||
!full
|
||||
&& existingConversation
|
||||
&& gatewayLastMessageId
|
||||
&& gatewayLastMessageId === String(existingConversation.telegramLastMessageId || '')
|
||||
&& !profileDue
|
||||
) {
|
||||
await patchResource(tenantId, 'conversations', conversationId, {
|
||||
unreadCount: Number(chat.unreadCount || 0),
|
||||
lastReconciledAt: new Date().toISOString(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
const customerId = `tg-${channel.gatewayAccountId}-${chatId}`
|
||||
const existingCustomer = await getResource(tenantId, 'customers', customerId)
|
||||
const identity = await resolveTelegramIdentity(channel, chatId, chat, existingCustomer || {})
|
||||
const identity = profileDue || !existingCustomer
|
||||
? await resolveTelegramIdentity(channel, chatId, chat, existingCustomer || {})
|
||||
: existingCustomer
|
||||
const customer = await putResource(tenantId, 'customers', withDefaults('customers', {
|
||||
...(existingCustomer || {}),
|
||||
...identity,
|
||||
@@ -3572,17 +3640,25 @@ async function syncTelegramTenant(tenantId, channel) {
|
||||
telegramChatType: identity.telegramChatType || chat.type,
|
||||
source: 'telegram',
|
||||
}))
|
||||
const existingConversation = await getResource(tenantId, 'conversations', conversationId)
|
||||
const recent = await gatewayRequest(
|
||||
`/tg-accounts/${channel.gatewayAccountId}/chats/${chatId}/history?limit=100`,
|
||||
).catch(() => ({ messages: [] }))
|
||||
const historyMessages = [...(recent?.messages || [])]
|
||||
let historyComplete = Boolean(existingConversation?.historyComplete) || historyMessages.length < 100
|
||||
let cursor = existingConversation?.historyComplete
|
||||
? undefined
|
||||
: existingConversation?.historyOldestMessageId || historyMessages.at(-1)?.id
|
||||
let historyComplete = Boolean(existingConversation?.historyComplete)
|
||||
let cursor = historyMessages.at(-1)?.id
|
||||
let pages = 1
|
||||
while (!historyComplete && cursor && pages < 50) {
|
||||
const previousLastMessageId = String(existingConversation?.telegramLastMessageId || '')
|
||||
const hasPreviousLastMessage = () => previousLastMessageId
|
||||
&& historyMessages.some((message) => String(message.id) === previousLastMessageId)
|
||||
if (full && existingConversation?.historyOldestMessageId) {
|
||||
cursor = existingConversation.historyOldestMessageId
|
||||
}
|
||||
const maxPages = full ? 50 : 10
|
||||
while (
|
||||
cursor
|
||||
&& pages < maxPages
|
||||
&& (full ? !historyComplete : previousLastMessageId && !hasPreviousLastMessage())
|
||||
) {
|
||||
const page = await gatewayRequest(
|
||||
`/tg-accounts/${channel.gatewayAccountId}/chats/${chatId}/history?limit=100&fromMessageId=${encodeURIComponent(String(cursor))}`,
|
||||
).catch(() => ({ messages: [] }))
|
||||
@@ -3592,11 +3668,12 @@ async function syncTelegramTenant(tenantId, channel) {
|
||||
historyMessages.push(...newRows)
|
||||
pages++
|
||||
if (rows.length < 100 || !newRows.length) {
|
||||
historyComplete = true
|
||||
if (full) historyComplete = true
|
||||
break
|
||||
}
|
||||
cursor = rows.at(-1)?.id
|
||||
}
|
||||
if (full && historyMessages.length < 100) historyComplete = true
|
||||
const history = { messages: historyMessages }
|
||||
let lastMessage
|
||||
for (const raw of [...(history?.messages || [])].reverse()) {
|
||||
@@ -3640,13 +3717,17 @@ async function syncTelegramTenant(tenantId, channel) {
|
||||
historyOldestMessageId: historyMessages.at(-1)?.id
|
||||
? String(historyMessages.at(-1).id)
|
||||
: existingConversation?.historyOldestMessageId,
|
||||
telegramLastMessageId: gatewayLastMessageId
|
||||
|| String(lastMessage?.id || existingConversation?.telegramLastMessageId || ''),
|
||||
profileSyncedAt: profileDue ? new Date().toISOString() : existingConversation?.profileSyncedAt,
|
||||
lastReconciledAt: new Date().toISOString(),
|
||||
historySyncedAt: new Date().toISOString(),
|
||||
createdAt: existingConversation?.createdAt || new Date().toISOString(),
|
||||
})
|
||||
}
|
||||
await patchResource(tenantId, 'channels', channel.id, { lastSyncAt: new Date().toISOString() })
|
||||
broadcast(tenantId, { type: 'sync.completed', channelId: channel.id, chats: chats?.length || 0, messages: messagesSynced })
|
||||
return { gateway, chats: chats?.length || 0, messages: messagesSynced }
|
||||
return { gateway, chats: chats?.length || 0, messages: messagesSynced, full }
|
||||
}
|
||||
|
||||
app.post('/api/channels/:id/sync', async (req, res) => {
|
||||
@@ -3660,7 +3741,7 @@ app.post('/api/channels/:id/sync', async (req, res) => {
|
||||
ok(res, await syncWhatsAppTenant(req.auth.tenantId, channel, { full: req.body.full === true }))
|
||||
return
|
||||
}
|
||||
ok(res, await syncTelegramTenant(req.auth.tenantId, channel))
|
||||
ok(res, await syncTelegramTenant(req.auth.tenantId, channel, { full: req.body.full === true }))
|
||||
})
|
||||
|
||||
app.post('/api/channels/test', async (req, res) => {
|
||||
@@ -4605,6 +4686,42 @@ app.get('/api/search', async (req, res) => {
|
||||
ok(res, result)
|
||||
})
|
||||
|
||||
async function claimWebhookEvent(tenantId, eventId, value) {
|
||||
const existing = await getResource(tenantId, 'webhook-events', eventId)
|
||||
const decision = webhookEventClaimDecision(existing)
|
||||
if (decision === 'duplicate') return { duplicate: true, existing }
|
||||
if (decision === 'busy') {
|
||||
throw fail(409, 'Webhook event is already processing')
|
||||
}
|
||||
const event = await putResource(tenantId, 'webhook-events', {
|
||||
...(existing || {}),
|
||||
...value,
|
||||
id: eventId,
|
||||
status: 'processing',
|
||||
attempts: Number(existing?.attempts || 0) + 1,
|
||||
receivedAt: existing?.receivedAt || new Date().toISOString(),
|
||||
processingAt: new Date().toISOString(),
|
||||
lastError: undefined,
|
||||
})
|
||||
return { duplicate: false, event }
|
||||
}
|
||||
|
||||
async function completeWebhookEvent(tenantId, eventId) {
|
||||
await patchResource(tenantId, 'webhook-events', eventId, {
|
||||
status: 'completed',
|
||||
completedAt: new Date().toISOString(),
|
||||
lastError: undefined,
|
||||
})
|
||||
}
|
||||
|
||||
async function failWebhookEvent(tenantId, eventId, error) {
|
||||
await patchResource(tenantId, 'webhook-events', eventId, {
|
||||
status: 'failed',
|
||||
failedAt: new Date().toISOString(),
|
||||
lastError: String(error?.message || error),
|
||||
}).catch(() => null)
|
||||
}
|
||||
|
||||
app.post('/internal/webhooks/telegram/:tenantId/:channelId', async (req, res) => {
|
||||
const expected = process.env.TELEGRAM_WEBHOOK_SECRET
|
||||
const received = String(req.get('x-webhook-signature') || '').replace(/^sha256=/, '')
|
||||
@@ -4615,14 +4732,21 @@ app.post('/internal/webhooks/telegram/:tenantId/:channelId', async (req, res) =>
|
||||
if (!channel) throw fail(404, 'Webhook 渠道映射不存在')
|
||||
if (channel.status === 'paused') return ok(res, { accepted: true, ignoredPausedChannel: true })
|
||||
const eventId = String(req.body.id || req.body.updateId || crypto.createHash('sha256').update(req.rawBody || Buffer.from(JSON.stringify(req.body))).digest('hex'))
|
||||
if (await getResource(tenantId, 'webhook-events', eventId)) return ok(res, { accepted: true, duplicate: true })
|
||||
await putResource(tenantId, 'webhook-events', { id: eventId, payload: req.body, receivedAt: new Date().toISOString() })
|
||||
const claim = await claimWebhookEvent(tenantId, eventId, {
|
||||
channelId,
|
||||
provider: 'telegram',
|
||||
event: req.body.event,
|
||||
payload: req.body,
|
||||
})
|
||||
if (claim.duplicate) return ok(res, { accepted: true, duplicate: true })
|
||||
try {
|
||||
const telegramMessage = req.body?.data?.message
|
||||
if (['message.received', 'message.pending', 'message.sent', 'message.failed'].includes(req.body.event) && telegramMessage) {
|
||||
const chatId = String(telegramMessage.chat_id)
|
||||
const conversationId = `${channelId}:${chatId}`
|
||||
const tombstone = await getResource(tenantId, 'conversation-tombstones', conversationId)
|
||||
if (tombstone && req.body.event !== 'message.received') {
|
||||
await completeWebhookEvent(tenantId, eventId)
|
||||
return ok(res, { accepted: true, ignoredDeletedConversation: true })
|
||||
}
|
||||
if (tombstone) {
|
||||
@@ -4734,7 +4858,56 @@ app.post('/internal/webhooks/telegram/:tenantId/:channelId', async (req, res) =>
|
||||
enqueueAutomaticProfile(tenantId, conversationId)
|
||||
}
|
||||
}
|
||||
if (req.body.event === 'message.edited') {
|
||||
const update = req.body?.data || {}
|
||||
const chatId = String(update.chat_id || '')
|
||||
const messageId = String(update.message_id || '')
|
||||
if (chatId && messageId) {
|
||||
const conversationId = `${channelId}:${chatId}`
|
||||
const existing = await getResource(tenantId, `messages:${conversationId}`, messageId)
|
||||
if (existing) {
|
||||
const content = update.new_content || existing.telegramContent || {}
|
||||
const originalText = content?.text?.text || content?.caption?.text || existing.originalText
|
||||
const message = await patchResource(tenantId, `messages:${conversationId}`, messageId, {
|
||||
originalText,
|
||||
telegramContent: content,
|
||||
editedAt: new Date().toISOString(),
|
||||
translatedText: undefined,
|
||||
translationCn: undefined,
|
||||
translationStatus: originalText ? 'pending' : existing.translationStatus,
|
||||
})
|
||||
const conversation = await getResource(tenantId, 'conversations', conversationId)
|
||||
if (String(conversation?.lastMessage?.id || '') === messageId) {
|
||||
await patchResource(tenantId, 'conversations', conversationId, { lastMessage: message })
|
||||
}
|
||||
broadcast(tenantId, { type: 'message.updated', conversationId, message })
|
||||
if (message) enqueueAutomaticTranslation(tenantId, conversationId, message)
|
||||
}
|
||||
}
|
||||
} else if (req.body.event === 'message.deleted') {
|
||||
const update = req.body?.data || {}
|
||||
const chatId = String(update.chat_id || '')
|
||||
const conversationId = `${channelId}:${chatId}`
|
||||
for (const rawId of update.message_ids || []) {
|
||||
const messageId = String(rawId)
|
||||
const message = await patchResource(tenantId, `messages:${conversationId}`, messageId, {
|
||||
isDeleted: true,
|
||||
deletedAt: new Date().toISOString(),
|
||||
})
|
||||
if (!message) continue
|
||||
const conversation = await getResource(tenantId, 'conversations', conversationId)
|
||||
if (String(conversation?.lastMessage?.id || '') === messageId) {
|
||||
await patchResource(tenantId, 'conversations', conversationId, { lastMessage: message })
|
||||
}
|
||||
broadcast(tenantId, { type: 'message.updated', conversationId, message })
|
||||
}
|
||||
}
|
||||
await completeWebhookEvent(tenantId, eventId)
|
||||
ok(res, { accepted: true })
|
||||
} catch (error) {
|
||||
await failWebhookEvent(tenantId, eventId, error)
|
||||
throw error
|
||||
}
|
||||
})
|
||||
|
||||
app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) => {
|
||||
@@ -4759,7 +4932,13 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
|
||||
req.body.id
|
||||
|| crypto.createHash('sha256').update(req.rawBody || Buffer.from(JSON.stringify(req.body))).digest('hex'),
|
||||
)
|
||||
if (await getResource(tenantId, 'webhook-events', eventId)) {
|
||||
const claim = await claimWebhookEvent(tenantId, eventId, {
|
||||
channelId,
|
||||
provider: 'whatsapp',
|
||||
event: req.body.event,
|
||||
payload: req.body,
|
||||
})
|
||||
if (claim.duplicate) {
|
||||
if (req.body.event === 'account.connected') {
|
||||
scheduleWhatsAppSync(tenantId, channel, { full: true })
|
||||
} else if (req.body.event === 'history.synced') {
|
||||
@@ -4770,14 +4949,7 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
|
||||
}
|
||||
return ok(res, { accepted: true, duplicate: true })
|
||||
}
|
||||
await putResource(tenantId, 'webhook-events', {
|
||||
id: eventId,
|
||||
channelId,
|
||||
provider: 'whatsapp',
|
||||
event: req.body.event,
|
||||
payload: req.body,
|
||||
receivedAt: new Date().toISOString(),
|
||||
})
|
||||
try {
|
||||
const message = req.body?.data?.message
|
||||
if (['message.received', 'message.sent', 'message.edited'].includes(req.body.event) && message) {
|
||||
await upsertWhatsAppMessage(tenantId, channel, message, {
|
||||
@@ -4944,7 +5116,12 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) =>
|
||||
scheduleWhatsAppSync(tenantId, channel, { full: true })
|
||||
}
|
||||
}
|
||||
await completeWebhookEvent(tenantId, eventId)
|
||||
ok(res, { accepted: true })
|
||||
} catch (error) {
|
||||
await failWebhookEvent(tenantId, eventId, error)
|
||||
throw error
|
||||
}
|
||||
})
|
||||
|
||||
app.get('/health/live', (_req, res) => res.json({ status: 'ok', service: 'lingxi-crossborder-api' }))
|
||||
@@ -5071,6 +5248,160 @@ const webhookReconcileTimer = setInterval(() => {
|
||||
}, 30_000)
|
||||
webhookReconcileTimer.unref?.()
|
||||
|
||||
const activeMessagingReconciliations = new Set()
|
||||
async function reconcileConnectedMessagingChannels({ refreshProfiles = false, full = false } = {}) {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT tenant_id, data FROM resources
|
||||
WHERE kind='channels'
|
||||
AND data->>'status'='connected'
|
||||
AND data->>'type' IN ('telegram','whatsapp')
|
||||
AND data ? 'gatewayAccountId'`,
|
||||
)
|
||||
for (const row of rows) {
|
||||
const tenantId = String(row.tenant_id)
|
||||
const channel = row.data
|
||||
const key = `${tenantId}:${channel.id}`
|
||||
if (activeMessagingReconciliations.has(key)) continue
|
||||
activeMessagingReconciliations.add(key)
|
||||
try {
|
||||
const result = channel.type === 'whatsapp'
|
||||
? await syncWhatsAppTenant(tenantId, channel, { full, refreshProfiles })
|
||||
: await syncTelegramTenant(tenantId, channel, { full, refreshProfiles })
|
||||
await patchResource(tenantId, 'channels', channel.id, {
|
||||
lastReconciledAt: new Date().toISOString(),
|
||||
lastReconcileRecoveredMessages: Number(result?.messages || 0),
|
||||
reconcileError: undefined,
|
||||
...(refreshProfiles ? { lastProfileRefreshAt: new Date().toISOString() } : {}),
|
||||
...(full ? { lastFullReconcileAt: new Date().toISOString() } : {}),
|
||||
})
|
||||
if (Number(result?.messages || 0) > 0) {
|
||||
await putResource(tenantId, 'notifications', {
|
||||
id: `message-reconcile-${channel.id}-${new Date().toISOString().slice(0, 10)}`,
|
||||
type: 'info',
|
||||
title: `${channel.name || channel.type} 已自动补齐消息`,
|
||||
description: `完整性对账恢复了 ${Number(result.messages)} 条此前未入库的消息。`,
|
||||
link: '/channels',
|
||||
read: false,
|
||||
createdAt: new Date().toISOString(),
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
await patchResource(tenantId, 'channels', channel.id, {
|
||||
reconcileError: String(error?.message || error),
|
||||
lastReconcileFailedAt: new Date().toISOString(),
|
||||
}).catch(() => null)
|
||||
await putResource(tenantId, 'notifications', {
|
||||
id: `message-reconcile-failed-${channel.id}-${new Date().toISOString().slice(0, 10)}`,
|
||||
type: 'warning',
|
||||
title: `${channel.name || channel.type} 消息对账失败`,
|
||||
description: '系统将在下一周期自动重试；若持续失败，请检查渠道连接与代理状态。',
|
||||
link: '/channels',
|
||||
read: false,
|
||||
createdAt: new Date().toISOString(),
|
||||
}).catch(() => null)
|
||||
} finally {
|
||||
activeMessagingReconciliations.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function replayMessagingDeadLetters() {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT tenant_id, data FROM resources
|
||||
WHERE kind='channels'
|
||||
AND data->>'status'='connected'
|
||||
AND data->>'type' IN ('telegram','whatsapp')
|
||||
AND data ? 'gatewayAccountId'`,
|
||||
)
|
||||
for (const row of rows) {
|
||||
const tenantId = String(row.tenant_id)
|
||||
const channel = row.data
|
||||
let replayed = 0
|
||||
try {
|
||||
if (channel.type === 'whatsapp') {
|
||||
const deliveries = await whatsappRequest(
|
||||
`/accounts/${channel.gatewayAccountId}/webhook-deliveries?status=dead&limit=100`,
|
||||
)
|
||||
for (const delivery of deliveries || []) {
|
||||
await postWhatsApp(
|
||||
`/accounts/${channel.gatewayAccountId}/webhook-deliveries/${encodeURIComponent(delivery.id)}/replay`,
|
||||
{},
|
||||
)
|
||||
replayed++
|
||||
}
|
||||
} else if (channel.webhookId) {
|
||||
const deliveries = await gatewayRequest(
|
||||
`/tg-accounts/${channel.gatewayAccountId}/webhooks/${channel.webhookId}/deliveries`,
|
||||
)
|
||||
for (const delivery of (deliveries || []).filter((item) => item.status === 'dead_letter')) {
|
||||
await postGateway(
|
||||
`/tg-accounts/${channel.gatewayAccountId}/webhooks/${channel.webhookId}/deliveries/${delivery.id}/replay`,
|
||||
{},
|
||||
)
|
||||
replayed++
|
||||
}
|
||||
}
|
||||
await patchResource(tenantId, 'channels', channel.id, {
|
||||
lastDeadLetterCheckAt: new Date().toISOString(),
|
||||
lastDeadLetterReplayCount: replayed,
|
||||
})
|
||||
if (replayed > 0) {
|
||||
await putResource(tenantId, 'notifications', {
|
||||
id: `dead-letter-replay-${channel.id}-${new Date().toISOString().slice(0, 10)}`,
|
||||
type: 'warning',
|
||||
title: `${channel.name || channel.type} 已恢复失败投递`,
|
||||
description: `系统自动重放了 ${replayed} 条死信，并将在下一轮对账确认消息完整性。`,
|
||||
link: '/channels',
|
||||
read: false,
|
||||
createdAt: new Date().toISOString(),
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
await patchResource(tenantId, 'channels', channel.id, {
|
||||
deadLetterReplayError: String(error?.message || error),
|
||||
lastDeadLetterCheckAt: new Date().toISOString(),
|
||||
}).catch(() => null)
|
||||
await putResource(tenantId, 'notifications', {
|
||||
id: `dead-letter-replay-failed-${channel.id}-${new Date().toISOString().slice(0, 10)}`,
|
||||
type: 'warning',
|
||||
title: `${channel.name || channel.type} 失败投递恢复异常`,
|
||||
description: '死信自动重放未完成，系统将在 5 分钟后重试。',
|
||||
link: '/channels',
|
||||
read: false,
|
||||
createdAt: new Date().toISOString(),
|
||||
}).catch(() => null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function runMessagingIntegrityMaintenance() {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT tenant_id, data FROM resources
|
||||
WHERE kind='channels'
|
||||
AND data->>'status'='connected'
|
||||
AND data->>'type' IN ('telegram','whatsapp')`,
|
||||
)
|
||||
const now = Date.now()
|
||||
const dailyDue = rows.some((row) => (
|
||||
isTimestampOlderThan(row.data.lastProfileRefreshAt, 24 * 60 * 60_000, now)
|
||||
))
|
||||
if (dailyDue) await reconcileConnectedMessagingChannels({ refreshProfiles: true })
|
||||
const weeklyDue = rows.some((row) => (
|
||||
isTimestampOlderThan(row.data.lastFullReconcileAt, 7 * 24 * 60 * 60_000, now)
|
||||
))
|
||||
if (weeklyDue) await reconcileConnectedMessagingChannels({ full: true, refreshProfiles: true })
|
||||
}
|
||||
|
||||
setTimeout(() => void reconcileConnectedMessagingChannels(), 15_000).unref?.()
|
||||
setInterval(
|
||||
() => void reconcileConnectedMessagingChannels(),
|
||||
Math.max(30_000, Number(process.env.MESSAGE_RECONCILE_MS || 60_000)),
|
||||
).unref?.()
|
||||
setTimeout(() => void replayMessagingDeadLetters(), 30_000).unref?.()
|
||||
setInterval(() => void replayMessagingDeadLetters(), 5 * 60_000).unref?.()
|
||||
setTimeout(() => void runMessagingIntegrityMaintenance(), 90_000).unref?.()
|
||||
setInterval(() => void runMessagingIntegrityMaintenance(), 60 * 60_000).unref?.()
|
||||
|
||||
const activeEmailSyncs = new Set()
|
||||
async function pollConnectedEmailChannels() {
|
||||
const { rows } = await pool.query(
|
||||
|
||||
Reference in New Issue
Block a user