From ad126606f9965d6030e9df49a02f6740a4d28688 Mon Sep 17 00:00:00 2001 From: wmq Date: Sat, 1 Aug 2026 23:55:41 +0800 Subject: [PATCH] feat(reliability): close messaging gateway P0 gaps --- .env.example | 3 + CHANGELOG.md | 2 + server/package.json | 2 +- server/src/gateway.js | 3 + server/src/index.js | 399 ++++++++++++++++-- server/src/reliability.js | 15 + server/src/whatsapp.js | 3 + server/test/reliability.test.js | 18 + telegram-gateway/DEPLOYMENT.md | 5 + .../src/database/entities/tg-chat.entity.ts | 2 + .../1700000003000-TgChatIdentityUnique.ts | 36 ++ .../src/modules/tg-chats/tg-chats.service.ts | 29 +- .../modules/tg-updates/tg-updates.module.ts | 3 +- .../modules/tg-updates/tg-updates.service.ts | 61 +++ .../test/business-services.spec.ts | 4 +- 15 files changed, 544 insertions(+), 41 deletions(-) create mode 100644 server/src/reliability.js create mode 100644 server/test/reliability.test.js create mode 100644 telegram-gateway/src/database/migrations/1700000003000-TgChatIdentityUnique.ts diff --git a/.env.example b/.env.example index 2474423..ea3febb 100644 --- a/.env.example +++ b/.env.example @@ -41,6 +41,9 @@ SMTP_FROM= # 建议使用 32 字节以上独立随机值；未配置时回退到 LX_JWT_SECRET 派生密钥。 EMAIL_CREDENTIAL_KEY= EMAIL_POLL_MS=30000 +MESSAGE_RECONCILE_MS=60000 +WHATSAPP_GATEWAY_TIMEOUT_MS=30000 +TELEGRAM_GATEWAY_TIMEOUT_MS=30000 # 真实 Stripe Checkout STRIPE_SECRET_KEY= diff --git a/CHANGELOG.md b/CHANGELOG.md index 58edef3..09393cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ## [Unreleased] +- 完成双消息网关 P0 可靠性闭环：Webhook 采用 processing/completed/failed 可恢复状态机，Telegram 编辑与撤回落库，60 秒增量游标对账、5 分钟死信自动重放、每日资料刷新和每周受控全量对账。 +- Telegram 会话发现改为 TDLib 持续加载并持久化全部会话，平台按 offset 分页读取；网关请求增加超时与任务并发合并，避免单个资料接口阻塞全部同步。 - 修复 WhatsApp 账号重新授权后因网关账号 ID 与联系人名称变化导致历史会话分叉的问题；客户按稳定 JID 复用，旧渠道消息、任务、风险和商机自动归并，并保留最近有效的联系人资料。 ### 新增 diff --git a/server/package.json b/server/package.json index d52f7d9..29f295e 100644 --- a/server/package.json +++ b/server/package.json @@ -5,7 +5,7 @@ "type": "module", "scripts": { "start": "node src/index.js", - "test": "node --test test/business.test.js test/whatsapp-identity.test.js", + "test": "node --test test/business.test.js test/reliability.test.js test/whatsapp-identity.test.js", "test:email-protocol": "node --test test/email-protocol.test.js", "test:email-api": "node --test test/email-api-e2e.test.js" }, diff --git a/server/src/gateway.js b/server/src/gateway.js index 93feb25..f5550dc 100644 --- a/server/src/gateway.js +++ b/server/src/gateway.js @@ -45,6 +45,9 @@ export async function gatewayRequest(path, options = {}) { if (!gatewayToken || tokenExpiresSoon(gatewayToken)) await refreshGatewayToken() const send = () => fetch(`${base()}${path}`, { ...options, + signal: options.signal || AbortSignal.timeout( + Math.max(5_000, Number(process.env.TELEGRAM_GATEWAY_TIMEOUT_MS || 30_000)), + ), headers: { ...(options.body instanceof FormData ? {} : { 'content-type': 'application/json' }), authorization: `Bearer ${gatewayToken}`, diff --git a/server/src/index.js b/server/src/index.js index 79386f2..35c89bf 100644 --- a/server/src/index.js +++ b/server/src/index.js @@ -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( diff --git a/server/src/reliability.js b/server/src/reliability.js new file mode 100644 index 0000000..164afe2 --- /dev/null +++ b/server/src/reliability.js @@ -0,0 +1,15 @@ +export function webhookEventClaimDecision(existing, now = Date.now(), leaseMs = 2 * 60_000) { + if (existing?.status === 'completed') return 'duplicate' + const processingAt = Date.parse(existing?.processingAt || '') + if ( + existing?.status === 'processing' + && Number.isFinite(processingAt) + && processingAt > now - leaseMs + ) return 'busy' + return 'claim' +} + +export function isTimestampOlderThan(value, ageMs, now = Date.now()) { + const parsed = Date.parse(value || '') + return !Number.isFinite(parsed) || parsed < now - ageMs +} diff --git a/server/src/whatsapp.js b/server/src/whatsapp.js index 1ba246d..f7fcfc9 100644 --- a/server/src/whatsapp.js +++ b/server/src/whatsapp.js @@ -7,6 +7,9 @@ export async function whatsappRequest(path, options = {}) { } const response = await fetch(`${base()}${path}`, { ...options, + signal: options.signal || AbortSignal.timeout( + Math.max(5_000, Number(process.env.WHATSAPP_GATEWAY_TIMEOUT_MS || 30_000)), + ), headers: { ...(options.body instanceof FormData ? {} : { 'content-type': 'application/json' }), authorization: `Bearer ${token}`, diff --git a/server/test/reliability.test.js b/server/test/reliability.test.js new file mode 100644 index 0000000..90876d2 --- /dev/null +++ b/server/test/reliability.test.js @@ -0,0 +1,18 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { isTimestampOlderThan, webhookEventClaimDecision } from '../src/reliability.js' + +test('completed webhooks deduplicate while failed and expired processing events can resume', () => { + const now = Date.parse('2026-08-01T12:00:00.000Z') + assert.equal(webhookEventClaimDecision({ status: 'completed' }, now), 'duplicate') + assert.equal(webhookEventClaimDecision({ status: 'failed' }, now), 'claim') + assert.equal(webhookEventClaimDecision({ status: 'processing', processingAt: '2026-08-01T11:59:30.000Z' }, now), 'busy') + assert.equal(webhookEventClaimDecision({ status: 'processing', processingAt: '2026-08-01T11:57:00.000Z' }, now), 'claim') +}) + +test('missing timestamps are due and recent maintenance timestamps are not', () => { + const now = Date.parse('2026-08-01T12:00:00.000Z') + assert.equal(isTimestampOlderThan(undefined, 60_000, now), true) + assert.equal(isTimestampOlderThan('2026-08-01T11:58:00.000Z', 60_000, now), true) + assert.equal(isTimestampOlderThan('2026-08-01T11:59:30.000Z', 60_000, now), false) +}) diff --git a/telegram-gateway/DEPLOYMENT.md b/telegram-gateway/DEPLOYMENT.md index 3433645..8ff3f43 100644 --- a/telegram-gateway/DEPLOYMENT.md +++ b/telegram-gateway/DEPLOYMENT.md @@ -38,6 +38,11 @@ CORS_ORIGINS=https://your-control-plane.example Production uses migrations and keeps TypeORM synchronization disabled. +If Docker build containers cannot reach the default Debian repositories, set +`DEBIAN_MIRROR` and `DEBIAN_SECURITY_MIRROR` to reachable mirrors. HTTP mirrors +can bootstrap older base images that do not yet contain CA certificates; keep +all application and credential traffic on HTTPS. + ## Start From the project root: diff --git a/telegram-gateway/src/database/entities/tg-chat.entity.ts b/telegram-gateway/src/database/entities/tg-chat.entity.ts index 4f3a39e..8f36abb 100644 --- a/telegram-gateway/src/database/entities/tg-chat.entity.ts +++ b/telegram-gateway/src/database/entities/tg-chat.entity.ts @@ -6,6 +6,7 @@ import { UpdateDateColumn, ManyToOne, JoinColumn, + Index, } from 'typeorm'; import { TgAccount } from './tg-account.entity'; @@ -18,6 +19,7 @@ export enum ChatType { } @Entity('tg_chats') +@Index('IDX_tg_chat_account_identity', ['accountId', 'tgChatId'], { unique: true }) export class TgChat { @PrimaryGeneratedColumn('uuid') id: string; diff --git a/telegram-gateway/src/database/migrations/1700000003000-TgChatIdentityUnique.ts b/telegram-gateway/src/database/migrations/1700000003000-TgChatIdentityUnique.ts new file mode 100644 index 0000000..cf009ed --- /dev/null +++ b/telegram-gateway/src/database/migrations/1700000003000-TgChatIdentityUnique.ts @@ -0,0 +1,36 @@ +import { MigrationInterface, QueryRunner, TableIndex } from 'typeorm'; + +export class TgChatIdentityUnique1700000003000 implements MigrationInterface { + name = 'TgChatIdentityUnique1700000003000'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + DELETE FROM tg_chats older + USING tg_chats newer + WHERE older."accountId" = newer."accountId" + AND older."tgChatId" = newer."tgChatId" + AND ( + older."updatedAt" < newer."updatedAt" + OR (older."updatedAt" = newer."updatedAt" AND older.id < newer.id) + ) + `); + const table = await queryRunner.getTable('tg_chats'); + if (!table?.indices.some((index) => index.name === 'IDX_tg_chat_account_identity')) { + await queryRunner.createIndex( + 'tg_chats', + new TableIndex({ + name: 'IDX_tg_chat_account_identity', + columnNames: ['accountId', 'tgChatId'], + isUnique: true, + }), + ); + } + } + + public async down(queryRunner: QueryRunner): Promise { + const table = await queryRunner.getTable('tg_chats'); + if (table?.indices.some((index) => index.name === 'IDX_tg_chat_account_identity')) { + await queryRunner.dropIndex('tg_chats', 'IDX_tg_chat_account_identity'); + } + } +} diff --git a/telegram-gateway/src/modules/tg-chats/tg-chats.service.ts b/telegram-gateway/src/modules/tg-chats/tg-chats.service.ts index 77882e4..be91831 100644 --- a/telegram-gateway/src/modules/tg-chats/tg-chats.service.ts +++ b/telegram-gateway/src/modules/tg-chats/tg-chats.service.ts @@ -19,8 +19,8 @@ export class TgChatsService { return this.chatRepo.find({ where: { accountId }, order: { lastMessageDate: 'DESC', updatedAt: 'DESC' }, - take: limit, - skip: offset, + take: Math.min(200, Math.max(1, Number(limit) || 50)), + skip: Math.max(0, Number(offset) || 0), }); } @@ -106,6 +106,21 @@ export class TgChatsService { async syncChatsFromTdlib(tenantId: string, accountId: string): Promise { await this.tgAccountsService.findById(tenantId, accountId); + let loadedPages = 0; + for (; loadedPages < 100; loadedPages++) { + try { + await this.tdlibClientManager.sendRequest(accountId, { + '@type': 'loadChats', + chat_list: { '@type': 'chatListMain' }, + limit: 100, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (/404|all chats.*loaded|not found/i.test(message)) break; + throw error; + } + } + await new Promise((resolve) => setTimeout(resolve, 250)); const result = await this.tdlibClientManager.sendRequest(accountId, { '@type': 'getChats', limit: 100, @@ -123,7 +138,8 @@ export class TgChatsService { // A deleted/inaccessible chat must not abort the remaining first sync. } } - return { total: chatIds.length, synced: synced.length, chatIds }; + const total = await this.chatRepo.count({ where: { accountId } }); + return { total, synced: synced.length, loadedPages, chatIds }; } private async persistTdlibChat(accountId: string, chat: Record): Promise { @@ -331,6 +347,11 @@ export class TgChatsService { chat = this.chatRepo.create({ accountId, ...data }); } - return this.chatRepo.save(chat); + const { account: _account, ...persisted } = chat; + // TypeORM's DeepPartial type does not understand the JSON transformer fields. + await this.chatRepo.upsert(persisted as any, ['accountId', 'tgChatId']); + return this.chatRepo.findOneOrFail({ + where: { accountId, tgChatId: data.tgChatId }, + }); } } diff --git a/telegram-gateway/src/modules/tg-updates/tg-updates.module.ts b/telegram-gateway/src/modules/tg-updates/tg-updates.module.ts index b4991c1..4f884af 100644 --- a/telegram-gateway/src/modules/tg-updates/tg-updates.module.ts +++ b/telegram-gateway/src/modules/tg-updates/tg-updates.module.ts @@ -5,10 +5,11 @@ import { TdlibModule } from '../tdlib/tdlib.module'; import { TgWebhooksModule } from '../tg-webhooks/tg-webhooks.module'; import { TgUpdatesService } from './tg-updates.service'; import { TgAccount } from '../../database/entities/tg-account.entity'; +import { TgChat } from '../../database/entities/tg-chat.entity'; @Module({ imports: [ - TypeOrmModule.forFeature([TgRawUpdate, TgAccount]), + TypeOrmModule.forFeature([TgRawUpdate, TgAccount, TgChat]), TdlibModule, TgWebhooksModule, ], diff --git a/telegram-gateway/src/modules/tg-updates/tg-updates.service.ts b/telegram-gateway/src/modules/tg-updates/tg-updates.service.ts index d2048e5..aaf53e7 100644 --- a/telegram-gateway/src/modules/tg-updates/tg-updates.service.ts +++ b/telegram-gateway/src/modules/tg-updates/tg-updates.service.ts @@ -9,6 +9,7 @@ import { TgAccount, TgAccountStatus, } from '../../database/entities/tg-account.entity'; +import { ChatType, TgChat } from '../../database/entities/tg-chat.entity'; @Injectable() export class TgUpdatesService implements OnModuleInit, OnModuleDestroy { @@ -20,6 +21,8 @@ export class TgUpdatesService implements OnModuleInit, OnModuleDestroy { private readonly rawUpdateRepo: Repository, @InjectRepository(TgAccount) private readonly accountRepo: Repository, + @InjectRepository(TgChat) + private readonly chatRepo: Repository, private readonly tdlib: TdlibClientManagerService, private readonly webhooks: TgWebhooksService, ) {} @@ -50,6 +53,11 @@ export class TgUpdatesService implements OnModuleInit, OnModuleDestroy { if (updateType === 'updateAuthorizationState') { await this.updateAccountAuthorization(update.accountId, update.data); } + if (updateType === 'updateNewChat') { + await this.persistChat(update.accountId, update.data.chat as Record); + } else if (updateType === 'updateChatLastMessage') { + await this.persistChatLastMessage(update.accountId, update.data); + } const event = this.mapWebhookEvent(updateType, update.data); if (event) { @@ -60,6 +68,59 @@ export class TgUpdatesService implements OnModuleInit, OnModuleDestroy { await this.rawUpdateRepo.save(saved); } + private messageText(content: Record = {}): string { + return String( + content.text?.text + || content.caption?.text + || (content['@type'] ? `[${content['@type']}]` : ''), + ); + } + + private async persistChat(accountId: string, chat: Record | undefined) { + if (!chat?.id) return; + const rawType = String(chat.type?.['@type'] || ''); + const typeMap: Record = { + chatTypePrivate: ChatType.PRIVATE, + chatTypeBasicGroup: ChatType.GROUP, + chatTypeSupergroup: chat.type?.is_channel ? ChatType.CHANNEL : ChatType.SUPERGROUP, + chatTypeSecret: ChatType.SECRET, + }; + const existing = await this.chatRepo.findOne({ + where: { accountId, tgChatId: Number(chat.id) }, + }); + const value = existing || this.chatRepo.create({ accountId, tgChatId: Number(chat.id) }); + Object.assign(value, { + type: typeMap[rawType] || existing?.type || ChatType.PRIVATE, + title: String(chat.title || existing?.title || `Telegram ${chat.id}`), + lastMessage: this.messageText(chat.last_message?.content) || existing?.lastMessage || '', + lastMessageId: Number(chat.last_message?.id || existing?.lastMessageId || 0), + lastMessageDate: chat.last_message?.date + ? new Date(Number(chat.last_message.date) * 1000) + : existing?.lastMessageDate, + unreadCount: Number(chat.unread_count ?? existing?.unreadCount ?? 0), + photo: chat.photo || existing?.photo, + raw: { ...(existing?.raw || {}), ...chat }, + }); + const { account: _account, ...persisted } = value; + // TypeORM's DeepPartial type does not understand the JSON transformer fields. + await this.chatRepo.upsert(persisted as any, ['accountId', 'tgChatId']); + } + + private async persistChatLastMessage(accountId: string, update: Record) { + const chatId = Number(update.chat_id || 0); + if (!chatId) return; + const existing = await this.chatRepo.findOne({ where: { accountId, tgChatId: chatId } }); + if (!existing) return; + const message = update.last_message || {}; + existing.lastMessage = this.messageText(message.content) || existing.lastMessage; + existing.lastMessageId = Number(message.id || existing.lastMessageId || 0); + existing.lastMessageDate = message.date + ? new Date(Number(message.date) * 1000) + : existing.lastMessageDate; + existing.raw = { ...(existing.raw || {}), last_message: message }; + await this.chatRepo.save(existing); + } + private async updateAccountAuthorization( accountId: string, data: Record, diff --git a/telegram-gateway/test/business-services.spec.ts b/telegram-gateway/test/business-services.spec.ts index 0b2a850..741dc80 100644 --- a/telegram-gateway/test/business-services.spec.ts +++ b/telegram-gateway/test/business-services.spec.ts @@ -159,10 +159,12 @@ describe('Business service behavior', () => { }); it('persists Telegram broadcast channels as channel instead of supergroup', async () => { + let stored: any; const repo = { findOne: jest.fn().mockResolvedValue(null), create: jest.fn((value) => value), - save: jest.fn(async (value) => value), + upsert: jest.fn(async (value) => { stored = value; }), + findOneOrFail: jest.fn(async () => stored), }; const service = new TgChatsService(repo as any, tdlib as any, accounts as any); const saved = await (service as any).persistTdlibChat('account-1', {