wmq
2065 lines
81 KiB
JavaScript
2065 lines
81 KiB
JavaScript
import express from 'express'
|
|
import 'express-async-errors'
|
|
import multer from 'multer'
|
|
import pino from 'pino'
|
|
import QRCode from 'qrcode'
|
|
import crypto from 'node:crypto'
|
|
import fs from 'node:fs/promises'
|
|
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,
|
|
jidNormalizedUser,
|
|
makeCacheableSignalKeyStore,
|
|
useMultiFileAuthState,
|
|
} from 'baileys'
|
|
|
|
const port = Number(process.env.PORT || 3200)
|
|
const dataDir = process.env.DATA_DIR || '/data'
|
|
const authDir = path.join(dataDir, 'auth')
|
|
const storeDir = path.join(dataDir, 'store')
|
|
const outboundMediaDir = path.join(dataDir, 'outbound-media')
|
|
const mediaCacheDir = path.join(dataDir, 'media-cache')
|
|
const accountsFile = path.join(dataDir, 'accounts.json')
|
|
const webhookDeliveriesFile = path.join(dataDir, 'webhook-deliveries.json')
|
|
const outboundJobsFile = path.join(dataDir, 'outbound-jobs.json')
|
|
const gatewayToken = String(process.env.WHATSAPP_GATEWAY_TOKEN || '')
|
|
const defaultProxyUrl = String(process.env.WHATSAPP_DEFAULT_PROXY_URL || '').trim()
|
|
const maxMessagesPerChat = Math.max(2_000, Number(process.env.MAX_MESSAGES_PER_CHAT || 20_000))
|
|
const log = pino({ level: process.env.LOG_LEVEL || 'info' })
|
|
|
|
if (gatewayToken.length < 24) {
|
|
throw new Error('WHATSAPP_GATEWAY_TOKEN 必须至少 24 个字符')
|
|
}
|
|
|
|
await fs.mkdir(authDir, { recursive: true })
|
|
await fs.mkdir(storeDir, { recursive: true })
|
|
await fs.mkdir(outboundMediaDir, { recursive: true })
|
|
await fs.mkdir(mediaCacheDir, { recursive: true })
|
|
|
|
const accounts = new Map()
|
|
const sessions = new Map()
|
|
const stores = new Map()
|
|
const reconnectTimers = new Map()
|
|
const persistTimers = new Map()
|
|
const webhookDeliveries = new Map()
|
|
const webhookDeliveryTimers = new Map()
|
|
const webhookDeliveriesProcessing = new Set()
|
|
const outboundJobs = new Map()
|
|
const outboundJobTimers = new Map()
|
|
const outboundJobsProcessing = new Set()
|
|
const credentialKey = crypto.createHash('sha256').update(gatewayToken).digest()
|
|
|
|
function encryptCredential(value) {
|
|
if (!value) return ''
|
|
const iv = crypto.randomBytes(12)
|
|
const cipher = crypto.createCipheriv('aes-256-gcm', credentialKey, iv)
|
|
const encrypted = Buffer.concat([cipher.update(String(value), 'utf8'), cipher.final()])
|
|
return [
|
|
'v1',
|
|
iv.toString('base64url'),
|
|
cipher.getAuthTag().toString('base64url'),
|
|
encrypted.toString('base64url'),
|
|
].join('.')
|
|
}
|
|
|
|
function decryptCredential(value) {
|
|
if (!value) return ''
|
|
const parts = String(value).split('.')
|
|
if (parts.length !== 4 || parts[0] !== 'v1') return String(value)
|
|
const decipher = crypto.createDecipheriv(
|
|
'aes-256-gcm',
|
|
credentialKey,
|
|
Buffer.from(parts[1], 'base64url'),
|
|
)
|
|
decipher.setAuthTag(Buffer.from(parts[2], 'base64url'))
|
|
return Buffer.concat([
|
|
decipher.update(Buffer.from(parts[3], 'base64url')),
|
|
decipher.final(),
|
|
]).toString('utf8')
|
|
}
|
|
|
|
function accountWebhookSecret(account) {
|
|
return decryptCredential(account?.webhookSecretEncrypted || account?.webhookSecret || '')
|
|
}
|
|
|
|
function accountProxyUrl(account) {
|
|
return decryptCredential(account?.proxyUrlEncrypted || '') || defaultProxyUrl
|
|
}
|
|
|
|
function jsonSafe(value) {
|
|
return JSON.parse(JSON.stringify(value, BufferJSON.replacer), BufferJSON.reviver)
|
|
}
|
|
|
|
async function readJson(file, fallback) {
|
|
try {
|
|
return JSON.parse(await fs.readFile(file, 'utf8'), BufferJSON.reviver)
|
|
} catch {
|
|
return fallback
|
|
}
|
|
}
|
|
|
|
async function atomicJson(file, value) {
|
|
const temp = `${file}.${crypto.randomUUID()}.tmp`
|
|
await fs.writeFile(temp, JSON.stringify(value, BufferJSON.replacer))
|
|
await fs.rename(temp, file)
|
|
}
|
|
|
|
for (const delivery of await readJson(webhookDeliveriesFile, [])) {
|
|
if (delivery?.id) webhookDeliveries.set(delivery.id, delivery)
|
|
}
|
|
for (const job of await readJson(outboundJobsFile, [])) {
|
|
if (job?.id) outboundJobs.set(job.id, job)
|
|
}
|
|
|
|
async function saveWebhookDeliveries() {
|
|
const sorted = [...webhookDeliveries.values()]
|
|
.sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || ''))
|
|
const active = sorted.filter((item) => !['delivered', 'dead'].includes(item.status))
|
|
const terminal = sorted.filter((item) => ['delivered', 'dead'].includes(item.status)).slice(0, 2_000)
|
|
const values = [...active, ...terminal]
|
|
const keep = new Set(values.map((item) => item.id))
|
|
for (const id of webhookDeliveries.keys()) {
|
|
if (!keep.has(id)) webhookDeliveries.delete(id)
|
|
}
|
|
await atomicJson(webhookDeliveriesFile, values)
|
|
}
|
|
|
|
let webhookDeliverySaveChain = Promise.resolve()
|
|
function queueWebhookDeliveriesSave() {
|
|
webhookDeliverySaveChain = webhookDeliverySaveChain
|
|
.catch(() => undefined)
|
|
.then(() => saveWebhookDeliveries())
|
|
.catch((error) => log.error({ err: error }, 'webhook delivery persistence failed'))
|
|
return webhookDeliverySaveChain
|
|
}
|
|
|
|
async function saveOutboundJobs() {
|
|
const sorted = [...outboundJobs.values()]
|
|
.sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || ''))
|
|
const active = sorted.filter((item) => !['sent', 'dead'].includes(item.status))
|
|
const terminal = sorted.filter((item) => ['sent', 'dead'].includes(item.status)).slice(0, 2_000)
|
|
const values = [...active, ...terminal]
|
|
const keep = new Set(values.map((item) => item.id))
|
|
for (const id of outboundJobs.keys()) {
|
|
if (!keep.has(id)) outboundJobs.delete(id)
|
|
}
|
|
await atomicJson(outboundJobsFile, values)
|
|
}
|
|
|
|
let outboundJobsSaveChain = Promise.resolve()
|
|
function queueOutboundJobsSave() {
|
|
outboundJobsSaveChain = outboundJobsSaveChain
|
|
.catch(() => undefined)
|
|
.then(() => saveOutboundJobs())
|
|
.catch((error) => log.error({ err: error }, 'outbound job persistence failed'))
|
|
return outboundJobsSaveChain
|
|
}
|
|
|
|
async function saveAccounts() {
|
|
const values = [...accounts.values()].map(({ qrDataUrl, pairingCode, ...item }) => item)
|
|
await atomicJson(accountsFile, values)
|
|
}
|
|
|
|
function createStore() {
|
|
return {
|
|
chats: new Map(),
|
|
contacts: new Map(),
|
|
messages: new Map(),
|
|
groups: new Map(),
|
|
}
|
|
}
|
|
|
|
async function loadStore(accountId) {
|
|
if (stores.has(accountId)) return stores.get(accountId)
|
|
const raw = await readJson(path.join(storeDir, `${accountId}.json`), {})
|
|
const store = createStore()
|
|
for (const item of raw.chats || []) store.chats.set(item.id, item)
|
|
for (const item of raw.contacts || []) store.contacts.set(item.id, item)
|
|
for (const [jid, messages] of raw.messages || []) store.messages.set(jid, messages)
|
|
for (const item of raw.groups || []) store.groups.set(item.id, item)
|
|
stores.set(accountId, store)
|
|
return store
|
|
}
|
|
|
|
async function readAuthFiles(accountId) {
|
|
const directory = path.join(authDir, accountId)
|
|
const names = await fs.readdir(directory).catch(() => [])
|
|
const files = {}
|
|
for (const name of names) {
|
|
if (!/^[a-zA-Z0-9_.-]+\.json$/.test(name)) continue
|
|
files[name] = await fs.readFile(path.join(directory, name), 'utf8')
|
|
}
|
|
return files
|
|
}
|
|
|
|
function scheduleStoreSave(accountId) {
|
|
if (persistTimers.has(accountId)) return
|
|
const timer = setTimeout(async () => {
|
|
persistTimers.delete(accountId)
|
|
const store = stores.get(accountId)
|
|
if (!store) return
|
|
await atomicJson(path.join(storeDir, `${accountId}.json`), {
|
|
chats: [...store.chats.values()],
|
|
contacts: [...store.contacts.values()],
|
|
messages: [...store.messages.entries()],
|
|
groups: [...store.groups.values()],
|
|
}).catch((error) => log.error({ error, accountId }, 'store persistence failed'))
|
|
}, 350)
|
|
timer.unref?.()
|
|
persistTimers.set(accountId, timer)
|
|
}
|
|
|
|
function unwrapMessage(message = {}) {
|
|
return message.ephemeralMessage?.message
|
|
|| message.viewOnceMessage?.message
|
|
|| message.viewOnceMessageV2?.message
|
|
|| message.documentWithCaptionMessage?.message
|
|
|| message
|
|
}
|
|
|
|
function messageContent(raw) {
|
|
const content = unwrapMessage(raw?.message || {})
|
|
const contentType = getContentType(content)
|
|
const value = contentType ? content[contentType] : null
|
|
const mapping = {
|
|
imageMessage: 'image',
|
|
videoMessage: 'video',
|
|
audioMessage: 'audio',
|
|
documentMessage: 'file',
|
|
stickerMessage: 'sticker',
|
|
contactMessage: 'contact',
|
|
contactsArrayMessage: 'contact',
|
|
locationMessage: 'location',
|
|
liveLocationMessage: 'location',
|
|
pollCreationMessage: 'poll',
|
|
reactionMessage: 'reaction',
|
|
}
|
|
const text = content.conversation
|
|
|| content.extendedTextMessage?.text
|
|
|| (contentType === 'reactionMessage' ? value?.text : '')
|
|
|| value?.caption
|
|
|| value?.fileName
|
|
|| value?.displayName
|
|
|| ''
|
|
return {
|
|
type: mapping[contentType] || 'text',
|
|
contentType: contentType || 'unknown',
|
|
text,
|
|
caption: value?.caption || '',
|
|
mimetype: value?.mimetype || '',
|
|
fileName: value?.fileName || '',
|
|
fileLength: Number(value?.fileLength || 0),
|
|
quotedMessageId: value?.key?.id || value?.contextInfo?.stanzaId || undefined,
|
|
}
|
|
}
|
|
|
|
function normalizedMessage(accountId, raw) {
|
|
const protocol = raw?.message?.protocolMessage
|
|
const editedMessage = protocol?.editedMessage
|
|
const key = editedMessage && protocol?.key ? protocol.key : raw.key || {}
|
|
const jid = String(key.remoteJid || '')
|
|
const timestamp = Number(raw.messageTimestamp || Date.now() / 1000)
|
|
const effectiveRaw = editedMessage
|
|
? { ...raw, key, message: editedMessage }
|
|
: raw
|
|
return {
|
|
id: String(key.id || crypto.randomUUID()),
|
|
accountId,
|
|
chatId: jid,
|
|
senderId: String(key.participant || jid),
|
|
fromMe: Boolean(key.fromMe),
|
|
pushName: raw.pushName || '',
|
|
timestamp,
|
|
createdAt: new Date(timestamp * 1000).toISOString(),
|
|
...messageContent(effectiveRaw),
|
|
edited: Boolean(editedMessage),
|
|
raw: jsonSafe(effectiveRaw),
|
|
}
|
|
}
|
|
|
|
function revokedMessageKey(raw) {
|
|
const protocol = raw?.message?.protocolMessage
|
|
const type = String(protocol?.type ?? '').toUpperCase()
|
|
if (!protocol?.key || !(protocol.type === 0 || type.includes('REVOKE'))) return null
|
|
return protocol.key
|
|
}
|
|
|
|
function messageDeliveryStatus(status, fromMe = true) {
|
|
const value = Number(status)
|
|
if (!Number.isFinite(value)) return fromMe ? 'sent' : 'delivered'
|
|
if (value <= 0) return 'failed'
|
|
if (value === 1) return 'sending'
|
|
if (value === 2) return 'sent'
|
|
if (value === 3) return 'delivered'
|
|
return 'read'
|
|
}
|
|
|
|
function upsertMessage(accountId, raw) {
|
|
const store = stores.get(accountId)
|
|
if (!store || !raw?.key?.remoteJid || !raw?.message) return null
|
|
const message = normalizedMessage(accountId, raw)
|
|
const list = store.messages.get(message.chatId) || []
|
|
const index = list.findIndex((item) => item.id === message.id)
|
|
if (index >= 0) list[index] = message
|
|
else list.push(message)
|
|
list.sort((a, b) => a.timestamp - b.timestamp)
|
|
if (list.length > maxMessagesPerChat) list.splice(0, list.length - maxMessagesPerChat)
|
|
store.messages.set(message.chatId, list)
|
|
const existing = store.chats.get(message.chatId) || { id: message.chatId }
|
|
store.chats.set(message.chatId, {
|
|
...existing,
|
|
id: message.chatId,
|
|
name: existing.name || message.pushName || message.chatId.split('@')[0],
|
|
timestamp: message.timestamp,
|
|
unreadCount: message.fromMe ? Number(existing.unreadCount || 0) : Number(existing.unreadCount || 0) + 1,
|
|
lastMessage: { id: message.id, text: message.text, type: message.type, createdAt: message.createdAt },
|
|
})
|
|
scheduleStoreSave(accountId)
|
|
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)
|
|
}
|
|
|
|
function accountPublic(account) {
|
|
if (!account) return account
|
|
const {
|
|
webhookSecret: _webhookSecret,
|
|
webhookSecretEncrypted: _webhookSecretEncrypted,
|
|
proxyUrlEncrypted: _proxyUrlEncrypted,
|
|
...safe
|
|
} = account
|
|
return {
|
|
...safe,
|
|
webhookSecretConfigured: Boolean(accountWebhookSecret(account)),
|
|
proxyConfigured: Boolean(accountProxyUrl(account)),
|
|
}
|
|
}
|
|
|
|
function accountSocket(accountId) {
|
|
const session = sessions.get(accountId)
|
|
if (!session?.sock) {
|
|
const error = new Error('WhatsApp 账号尚未启动')
|
|
error.status = 409
|
|
throw error
|
|
}
|
|
return session.sock
|
|
}
|
|
|
|
function disconnectStatus(error) {
|
|
return Number(error?.output?.statusCode || new Boom(error).output?.statusCode || 500)
|
|
}
|
|
|
|
function isTransientPairingError(error) {
|
|
const status = disconnectStatus(error)
|
|
return [408, 428, 500, 503, 515].includes(status)
|
|
|| /connection (?:closed|failure|terminated)|timed out/i.test(String(error?.message || ''))
|
|
}
|
|
|
|
async function resetPairingSession(accountId) {
|
|
const account = accounts.get(accountId)
|
|
if (!account) throw Object.assign(new Error('账号不存在'), { status: 404 })
|
|
const timer = reconnectTimers.get(accountId)
|
|
if (timer) clearTimeout(timer)
|
|
reconnectTimers.delete(accountId)
|
|
const session = sessions.get(accountId)
|
|
sessions.delete(accountId)
|
|
if (session?.sock) {
|
|
try {
|
|
session.sock.end(new Error('Pairing session reset'))
|
|
} catch {
|
|
// Socket may already be closed.
|
|
}
|
|
}
|
|
await fs.rm(path.join(authDir, accountId), { recursive: true, force: true })
|
|
await fs.mkdir(path.join(authDir, accountId), { recursive: true, mode: 0o700 })
|
|
delete account.user
|
|
delete account.qrDataUrl
|
|
delete account.pairingCode
|
|
delete account.lastError
|
|
account.status = 'created'
|
|
account.connectionState = 'idle'
|
|
account.updatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
}
|
|
|
|
async function waitForPairingReady(accountId, sock, timeoutMs = 20_000) {
|
|
const account = accounts.get(accountId)
|
|
if (account?.qrDataUrl) return
|
|
try {
|
|
await sock.waitForConnectionUpdate(async (update) => Boolean(update.qr || update.connection === 'open'), timeoutMs)
|
|
} catch (error) {
|
|
if (isTransientPairingError(error)) {
|
|
throw Object.assign(new Error('WhatsApp 连接尚未就绪'), {
|
|
status: 503,
|
|
cause: error,
|
|
})
|
|
}
|
|
throw error
|
|
}
|
|
if (sessions.get(accountId)?.sock !== sock) {
|
|
throw Object.assign(new Error('WhatsApp 连接已重置，请重试'), { status: 503 })
|
|
}
|
|
}
|
|
|
|
function scheduleWebhookDelivery(deliveryId, delayMs = 0) {
|
|
const existing = webhookDeliveryTimers.get(deliveryId)
|
|
if (existing) clearTimeout(existing)
|
|
const timer = setTimeout(() => {
|
|
webhookDeliveryTimers.delete(deliveryId)
|
|
void processWebhookDelivery(deliveryId)
|
|
}, Math.max(0, delayMs))
|
|
timer.unref?.()
|
|
webhookDeliveryTimers.set(deliveryId, timer)
|
|
}
|
|
|
|
async function processWebhookDelivery(deliveryId) {
|
|
if (webhookDeliveriesProcessing.has(deliveryId)) return
|
|
const delivery = webhookDeliveries.get(deliveryId)
|
|
if (!delivery || ['delivered', 'dead'].includes(delivery.status)) return
|
|
const account = accounts.get(delivery.accountId)
|
|
const webhookSecret = accountWebhookSecret(account)
|
|
if (!account?.webhookUrl || !account.webhookActive || !webhookSecret) {
|
|
delivery.status = 'dead'
|
|
delivery.lastError = 'Webhook is not configured or inactive'
|
|
delivery.updatedAt = new Date().toISOString()
|
|
await saveWebhookDeliveries()
|
|
return
|
|
}
|
|
const waitMs = Math.max(0, Date.parse(delivery.nextAttemptAt || '') - Date.now())
|
|
if (waitMs > 0) {
|
|
scheduleWebhookDelivery(deliveryId, waitMs)
|
|
return
|
|
}
|
|
webhookDeliveriesProcessing.add(deliveryId)
|
|
delivery.status = 'delivering'
|
|
delivery.attempts = Number(delivery.attempts || 0) + 1
|
|
delivery.updatedAt = new Date().toISOString()
|
|
void queueWebhookDeliveriesSave()
|
|
try {
|
|
const signature = crypto.createHmac('sha256', webhookSecret)
|
|
.update(delivery.payload)
|
|
.digest('hex')
|
|
const response = await fetch(account.webhookUrl, {
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'x-webhook-signature': `sha256=${signature}`,
|
|
'x-webhook-attempt': String(delivery.attempts),
|
|
'x-webhook-delivery-id': delivery.id,
|
|
},
|
|
body: delivery.payload,
|
|
signal: AbortSignal.timeout(15_000),
|
|
})
|
|
const responseBody = await response.text().catch(() => '')
|
|
delivery.responseStatus = response.status
|
|
delivery.responseBody = responseBody.slice(0, 500)
|
|
if (!response.ok) throw new Error(`Webhook HTTP ${response.status}${responseBody ? `: ${responseBody.slice(0, 300)}` : ''}`)
|
|
delivery.status = 'delivered'
|
|
delivery.deliveredAt = new Date().toISOString()
|
|
delete delivery.lastError
|
|
delete delivery.nextAttemptAt
|
|
} catch (error) {
|
|
delivery.lastError = error?.cause?.message || error?.message || 'Webhook request failed'
|
|
if (delivery.attempts >= Number(delivery.maxAttempts || 8)) {
|
|
delivery.status = 'dead'
|
|
delivery.deadAt = new Date().toISOString()
|
|
log.error({ accountId: account.id, event: delivery.event, deliveryId, err: error }, 'webhook delivery exhausted')
|
|
} else {
|
|
const delay = Math.min(5 * 60_000, 1000 * (2 ** Math.max(0, delivery.attempts - 1)))
|
|
delivery.status = 'retrying'
|
|
delivery.nextAttemptAt = new Date(Date.now() + delay).toISOString()
|
|
scheduleWebhookDelivery(deliveryId, delay)
|
|
}
|
|
} finally {
|
|
delivery.updatedAt = new Date().toISOString()
|
|
webhookDeliveriesProcessing.delete(deliveryId)
|
|
await queueWebhookDeliveriesSave()
|
|
}
|
|
}
|
|
|
|
async function deliverWebhook(account, event, data) {
|
|
if (!account.webhookUrl || !account.webhookActive || !accountWebhookSecret(account)) return
|
|
const messageId = ['message.received', 'message.sent', 'message.edited'].includes(event)
|
|
? String(data?.message?.id || '')
|
|
: ''
|
|
const id = messageId
|
|
? `wa-${crypto.createHash('sha256').update(`${account.id}:${event}:${messageId}`).digest('hex').slice(0, 32)}`
|
|
: crypto.randomUUID()
|
|
if (webhookDeliveries.has(id)) return
|
|
const payload = JSON.stringify({
|
|
id,
|
|
event,
|
|
occurredAt: new Date().toISOString(),
|
|
data: { accountId: account.id, ...data },
|
|
})
|
|
webhookDeliveries.set(id, {
|
|
id,
|
|
accountId: account.id,
|
|
event,
|
|
payload,
|
|
status: 'pending',
|
|
attempts: 0,
|
|
maxAttempts: 8,
|
|
createdAt: new Date().toISOString(),
|
|
updatedAt: new Date().toISOString(),
|
|
})
|
|
scheduleWebhookDelivery(id)
|
|
await queueWebhookDeliveriesSave()
|
|
}
|
|
|
|
function queuedOutboundMessage(job) {
|
|
const timestamp = Math.floor(Date.parse(job.createdAt) / 1000)
|
|
return {
|
|
id: job.clientMessageId,
|
|
accountId: job.accountId,
|
|
chatId: job.jid,
|
|
senderId: job.jid,
|
|
fromMe: true,
|
|
timestamp,
|
|
createdAt: job.createdAt,
|
|
type: job.type || 'text',
|
|
contentType: job.type === 'text' || !job.type ? 'conversation' : 'queuedMedia',
|
|
text: job.text || job.caption || job.fileName || '',
|
|
caption: job.caption || '',
|
|
mimetype: job.mimetype || '',
|
|
fileName: job.fileName || '',
|
|
deliveryStatus: job.status === 'dead' ? 'failed' : 'sending',
|
|
queued: job.status !== 'sent',
|
|
}
|
|
}
|
|
|
|
function scheduleOutboundJob(jobId, delayMs = 0) {
|
|
const existing = outboundJobTimers.get(jobId)
|
|
if (existing) clearTimeout(existing)
|
|
const timer = setTimeout(() => {
|
|
outboundJobTimers.delete(jobId)
|
|
void processOutboundJob(jobId)
|
|
}, Math.max(0, delayMs))
|
|
timer.unref?.()
|
|
outboundJobTimers.set(jobId, timer)
|
|
}
|
|
|
|
function outboundThrottleDelay(job) {
|
|
const now = Date.now()
|
|
const sent = [...outboundJobs.values()]
|
|
.filter((item) => item.accountId === job.accountId && item.status === 'sent' && item.sentAt)
|
|
const accountWindow = sent
|
|
.map((item) => Date.parse(item.sentAt))
|
|
.filter((time) => time > now - 60_000)
|
|
.sort((a, b) => a - b)
|
|
if (accountWindow.length >= 30) return Math.max(250, accountWindow[0] + 60_000 - now)
|
|
const chatWindow = sent
|
|
.filter((item) => item.jid === job.jid)
|
|
.map((item) => Date.parse(item.sentAt))
|
|
.filter((time) => time > now - 10_000)
|
|
.sort((a, b) => a - b)
|
|
if (chatWindow.length >= 5) return Math.max(250, chatWindow[0] + 10_000 - now)
|
|
return 0
|
|
}
|
|
|
|
async function processOutboundJob(jobId) {
|
|
if (outboundJobsProcessing.has(jobId)) return outboundJobs.get(jobId)?.result
|
|
const job = outboundJobs.get(jobId)
|
|
if (!job || ['sent', 'dead'].includes(job.status)) return job?.result
|
|
const account = accounts.get(job.accountId)
|
|
const sock = sessions.get(job.accountId)?.sock
|
|
if (!account || !sock || account.status !== 'connected') {
|
|
job.status = 'queued'
|
|
job.nextAttemptAt = new Date(Date.now() + 5_000).toISOString()
|
|
job.updatedAt = new Date().toISOString()
|
|
await queueOutboundJobsSave()
|
|
scheduleOutboundJob(job.id, 5_000)
|
|
return queuedOutboundMessage(job)
|
|
}
|
|
const throttleDelay = outboundThrottleDelay(job)
|
|
if (throttleDelay > 0) {
|
|
job.status = 'queued'
|
|
job.throttled = true
|
|
job.nextAttemptAt = new Date(Date.now() + throttleDelay).toISOString()
|
|
job.updatedAt = new Date().toISOString()
|
|
await queueOutboundJobsSave()
|
|
scheduleOutboundJob(job.id, throttleDelay)
|
|
return queuedOutboundMessage(job)
|
|
}
|
|
outboundJobsProcessing.add(job.id)
|
|
job.status = 'sending'
|
|
job.attempts = Number(job.attempts || 0) + 1
|
|
job.updatedAt = new Date().toISOString()
|
|
void queueOutboundJobsSave()
|
|
try {
|
|
const store = await loadStore(job.accountId)
|
|
const quoted = job.quotedMessageId
|
|
? (store.messages.get(job.jid) || []).find((item) => item.id === job.quotedMessageId)?.raw
|
|
: undefined
|
|
let content = { text: job.text }
|
|
if (job.type && job.type !== 'text') {
|
|
const buffer = await fs.readFile(job.filePath)
|
|
if (job.type === 'sticker') content = { sticker: buffer, mimetype: job.mimetype }
|
|
else if (job.type === 'image') content = {
|
|
image: buffer,
|
|
caption: job.caption || '',
|
|
mimetype: job.mimetype,
|
|
}
|
|
else if (job.type === 'video') content = {
|
|
video: buffer,
|
|
caption: job.caption || '',
|
|
mimetype: job.mimetype,
|
|
}
|
|
else if (job.type === 'audio') content = {
|
|
audio: buffer,
|
|
mimetype: job.mimetype,
|
|
ptt: Boolean(job.ptt),
|
|
}
|
|
else content = {
|
|
document: buffer,
|
|
mimetype: job.mimetype,
|
|
fileName: job.fileName,
|
|
caption: job.caption || '',
|
|
}
|
|
}
|
|
const result = await sock.sendMessage(
|
|
job.jid,
|
|
content,
|
|
{
|
|
...(quoted ? { quoted } : {}),
|
|
messageId: job.clientMessageId,
|
|
},
|
|
)
|
|
if (!result) throw new Error('WhatsApp 未返回发送结果')
|
|
const message = upsertMessage(job.accountId, result)
|
|
job.status = 'sent'
|
|
job.result = { ...message, deliveryStatus: 'sent', queued: false }
|
|
job.sentAt = new Date().toISOString()
|
|
delete job.throttled
|
|
delete job.lastError
|
|
delete job.nextAttemptAt
|
|
if (job.filePath) {
|
|
await fs.rm(job.filePath, { force: true }).catch(() => undefined)
|
|
delete job.filePath
|
|
}
|
|
await deliverWebhook(account, 'message.sent', {
|
|
message: job.result,
|
|
idempotencyKey: job.idempotencyKey,
|
|
})
|
|
return job.result
|
|
} catch (error) {
|
|
job.lastError = error?.message || 'WhatsApp 发送失败'
|
|
if (job.attempts >= 8) {
|
|
job.status = 'dead'
|
|
job.deadAt = new Date().toISOString()
|
|
await deliverWebhook(account, 'message.updated', {
|
|
updates: [{
|
|
id: job.clientMessageId,
|
|
chatId: job.jid,
|
|
status: 'failed',
|
|
}],
|
|
})
|
|
} else {
|
|
const riskLimited = /(?:429|rate.?limit|too many|spam|flood|temporar(?:y|ily) unavailable)/i
|
|
.test(String(job.lastError))
|
|
const delay = riskLimited
|
|
? Math.min(30 * 60_000, 60_000 * (2 ** Math.max(0, job.attempts - 1)))
|
|
: Math.min(60_000, 1000 * (2 ** Math.max(0, job.attempts - 1)))
|
|
job.status = 'retrying'
|
|
job.riskBackoff = riskLimited
|
|
job.nextAttemptAt = new Date(Date.now() + delay).toISOString()
|
|
scheduleOutboundJob(job.id, delay)
|
|
}
|
|
return queuedOutboundMessage(job)
|
|
} finally {
|
|
job.updatedAt = new Date().toISOString()
|
|
outboundJobsProcessing.delete(job.id)
|
|
await queueOutboundJobsSave()
|
|
}
|
|
}
|
|
|
|
async function connectAccount(accountId) {
|
|
const account = accounts.get(accountId)
|
|
if (!account) throw Object.assign(new Error('账号不存在'), { status: 404 })
|
|
if (sessions.get(accountId)?.sock) return accountPublic(account)
|
|
if (reconnectTimers.has(accountId)) {
|
|
clearTimeout(reconnectTimers.get(accountId))
|
|
reconnectTimers.delete(accountId)
|
|
}
|
|
const store = await loadStore(accountId)
|
|
const authPath = path.join(authDir, accountId)
|
|
await fs.mkdir(authPath, { recursive: true, mode: 0o700 })
|
|
const { state, saveCreds } = await useMultiFileAuthState(authPath)
|
|
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,
|
|
auth: {
|
|
creds: state.creds,
|
|
keys: makeCacheableSignalKeyStore(state.keys, logger),
|
|
},
|
|
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) => {
|
|
const list = store.messages.get(String(key.remoteJid || '')) || []
|
|
return list.find((item) => item.id === key.id)?.raw?.message
|
|
},
|
|
cachedGroupMetadata: async (jid) => store.groups.get(jid),
|
|
})
|
|
sessions.set(accountId, { sock, connecting: true, fetchDispatcher })
|
|
account.status = state.creds.registered ? 'connecting' : 'waiting_pairing'
|
|
account.updatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
|
|
sock.ev.on('creds.update', saveCreds)
|
|
sock.ev.on('connection.update', async (update) => {
|
|
if (update.qr) {
|
|
account.qrDataUrl = await QRCode.toDataURL(update.qr, { margin: 1, width: 320 })
|
|
account.status = 'waiting_pairing'
|
|
account.connectionState = 'pairing'
|
|
delete account.lastError
|
|
account.updatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
}
|
|
if (update.connection === 'open') {
|
|
account.status = 'connected'
|
|
account.connectionState = 'ready'
|
|
account.lastConnectedAt = new Date().toISOString()
|
|
account.user = sock.user ? {
|
|
id: jidNormalizedUser(sock.user.id),
|
|
name: sock.user.name || account.name || account.phoneNumber,
|
|
} : account.user
|
|
delete account.lastError
|
|
delete account.qrDataUrl
|
|
delete account.pairingCode
|
|
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)
|
|
}
|
|
}
|
|
void deliverWebhook(account, 'account.connected', { account: accountPublic(account) })
|
|
}
|
|
if (update.connection === 'close') {
|
|
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'})`
|
|
log.warn({ accountId, statusCode, loggedOut, error: update.lastDisconnect?.error }, 'WhatsApp connection closed')
|
|
await saveAccounts()
|
|
void deliverWebhook(account, 'account.disconnected', {
|
|
loggedOut,
|
|
reason: account.lastError,
|
|
})
|
|
if (!loggedOut) {
|
|
const timer = setTimeout(() => {
|
|
reconnectTimers.delete(accountId)
|
|
void connectAccount(accountId).catch((error) => log.error({ error, accountId }, 'reconnect failed'))
|
|
}, 5000)
|
|
timer.unref?.()
|
|
reconnectTimers.set(accountId, timer)
|
|
}
|
|
}
|
|
})
|
|
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 })
|
|
}
|
|
for (const contact of contacts) {
|
|
const id = String(contact.id || '')
|
|
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 || '')
|
|
if (id) store.chats.set(id, { ...(store.chats.get(id) || {}), ...jsonSafe(chat), id })
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
})
|
|
sock.ev.on('chats.update', (items) => {
|
|
for (const patch of items || []) {
|
|
const id = String(patch.id || '')
|
|
if (id) store.chats.set(id, { ...(store.chats.get(id) || {}), ...jsonSafe(patch), id })
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
})
|
|
sock.ev.on('contacts.upsert', (items) => {
|
|
for (const contact of items || []) {
|
|
const id = String(contact.id || '')
|
|
if (id) store.contacts.set(id, { ...(store.contacts.get(id) || {}), ...jsonSafe(contact), id })
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
void deliverWebhook(account, 'contacts.updated', { contacts: jsonSafe(items || []) })
|
|
})
|
|
sock.ev.on('contacts.update', (items) => {
|
|
for (const contact of items || []) {
|
|
const id = String(contact.id || '')
|
|
if (id) store.contacts.set(id, { ...(store.contacts.get(id) || {}), ...jsonSafe(contact), id })
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
void deliverWebhook(account, 'contacts.updated', { contacts: jsonSafe(items || []) })
|
|
})
|
|
sock.ev.on('groups.upsert', (items) => {
|
|
for (const group of items || []) store.groups.set(group.id, jsonSafe(group))
|
|
scheduleStoreSave(accountId)
|
|
})
|
|
sock.ev.on('groups.update', async (items) => {
|
|
for (const group of items || []) {
|
|
if (!group.id) continue
|
|
const metadata = await sock.groupMetadata(group.id).catch(() => null)
|
|
if (metadata) store.groups.set(group.id, jsonSafe(metadata))
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
})
|
|
sock.ev.on('messages.upsert', ({ messages = [], type }) => {
|
|
for (const raw of messages) {
|
|
const revokedKey = revokedMessageKey(raw)
|
|
if (revokedKey) {
|
|
const jid = String(revokedKey.remoteJid || raw.key?.remoteJid || '')
|
|
const list = store.messages.get(jid) || []
|
|
const index = list.findIndex((item) => item.id === String(revokedKey.id || ''))
|
|
if (index >= 0) {
|
|
list[index] = { ...list[index], deleted: true, deletedAt: new Date().toISOString() }
|
|
scheduleStoreSave(accountId)
|
|
}
|
|
void deliverWebhook(account, 'message.deleted', { keys: [jsonSafe(revokedKey)], all: false })
|
|
continue
|
|
}
|
|
const message = upsertMessage(accountId, raw)
|
|
if (!message) continue
|
|
const event = message.edited ? 'message.edited' : message.fromMe ? 'message.sent' : 'message.received'
|
|
void deliverWebhook(account, event, { message, upsertType: type })
|
|
}
|
|
})
|
|
sock.ev.on('messages.update', (updates) => {
|
|
const normalizedUpdates = []
|
|
for (const item of updates || []) {
|
|
const key = item?.key || {}
|
|
const revokedKey = revokedMessageKey({ key, message: item.update?.message })
|
|
if (revokedKey) {
|
|
const revokedJid = String(revokedKey.remoteJid || key.remoteJid || '')
|
|
const revokedId = String(revokedKey.id || '')
|
|
const revokedList = store.messages.get(revokedJid) || []
|
|
const revokedIndex = revokedList.findIndex((message) => message.id === revokedId)
|
|
if (revokedIndex >= 0) {
|
|
revokedList[revokedIndex] = {
|
|
...revokedList[revokedIndex],
|
|
deleted: true,
|
|
deletedAt: new Date().toISOString(),
|
|
}
|
|
}
|
|
void deliverWebhook(account, 'message.deleted', {
|
|
keys: [jsonSafe({ ...revokedKey, remoteJid: revokedJid })],
|
|
all: false,
|
|
})
|
|
continue
|
|
}
|
|
const jid = String(key.remoteJid || '')
|
|
const id = String(key.id || '')
|
|
if (!jid || !id) continue
|
|
const list = store.messages.get(jid) || []
|
|
const index = list.findIndex((message) => message.id === id)
|
|
const existing = index >= 0 ? list[index] : null
|
|
let message = existing
|
|
if (item.update?.message && existing?.raw) {
|
|
message = normalizedMessage(accountId, {
|
|
...existing.raw,
|
|
...item.update,
|
|
key,
|
|
message: item.update.message,
|
|
})
|
|
} else if (existing) {
|
|
message = {
|
|
...existing,
|
|
...(item.update?.status == null
|
|
? {}
|
|
: { deliveryStatus: messageDeliveryStatus(item.update.status, existing.fromMe) }),
|
|
raw: jsonSafe({ ...existing.raw, ...item.update, key }),
|
|
}
|
|
}
|
|
if (message && index >= 0) list[index] = message
|
|
normalizedUpdates.push({
|
|
id,
|
|
chatId: jid,
|
|
status: item.update?.status == null
|
|
? undefined
|
|
: messageDeliveryStatus(item.update.status, key.fromMe),
|
|
message,
|
|
update: jsonSafe(item.update || {}),
|
|
})
|
|
}
|
|
scheduleStoreSave(accountId)
|
|
void deliverWebhook(account, 'message.updated', { updates: normalizedUpdates })
|
|
})
|
|
sock.ev.on('messages.delete', (value) => {
|
|
const keys = Array.isArray(value?.keys) ? value.keys : []
|
|
for (const key of keys) {
|
|
const jid = String(key.remoteJid || '')
|
|
const list = store.messages.get(jid) || []
|
|
const index = list.findIndex((message) => message.id === String(key.id || ''))
|
|
if (index >= 0) list[index] = { ...list[index], deleted: true, deletedAt: new Date().toISOString() }
|
|
}
|
|
if (value?.all && value?.jid) store.messages.set(String(value.jid), [])
|
|
scheduleStoreSave(accountId)
|
|
void deliverWebhook(account, 'message.deleted', {
|
|
keys: jsonSafe(keys),
|
|
jid: value?.jid ? String(value.jid) : undefined,
|
|
all: Boolean(value?.all),
|
|
})
|
|
})
|
|
sock.ev.on('presence.update', (value) => {
|
|
void deliverWebhook(account, 'presence.updated', {
|
|
chatId: String(value?.id || ''),
|
|
presences: jsonSafe(value?.presences || {}),
|
|
})
|
|
})
|
|
return accountPublic(account)
|
|
}
|
|
|
|
const app = express()
|
|
app.set('trust proxy', 1)
|
|
app.use(express.json({ limit: '12mb' }))
|
|
app.use((req, res, next) => {
|
|
res.setHeader('x-request-id', req.get('x-request-id') || crypto.randomUUID())
|
|
next()
|
|
})
|
|
app.get('/health/live', (_req, res) => res.json({ status: 'ok', service: 'lingxi-whatsapp-gateway' }))
|
|
app.get('/health/ready', (_req, res) => res.json({
|
|
status: 'ready',
|
|
accounts: accounts.size,
|
|
connected: [...accounts.values()].filter((item) => item.status === 'connected').length,
|
|
}))
|
|
app.use((req, res, next) => {
|
|
const token = String(req.get('authorization') || '').replace(/^Bearer\s+/i, '')
|
|
|| String(req.get('x-gateway-token') || '')
|
|
if (token.length !== gatewayToken.length || !crypto.timingSafeEqual(Buffer.from(token), Buffer.from(gatewayToken))) {
|
|
return res.status(401).json({ code: 401, message: '网关令牌无效' })
|
|
}
|
|
next()
|
|
})
|
|
|
|
app.get('/metrics', async (_req, res) => {
|
|
const jobs = [...outboundJobs.values()]
|
|
const deliveries = [...webhookDeliveries.values()]
|
|
const sentLatencies = jobs
|
|
.filter((item) => item.sentAt && item.createdAt)
|
|
.map((item) => Math.max(0, Date.parse(item.sentAt) - Date.parse(item.createdAt)))
|
|
let chats = 0
|
|
let cachedMessages = 0
|
|
for (const store of stores.values()) {
|
|
chats += store.chats.size
|
|
for (const items of store.messages.values()) cachedMessages += items.length
|
|
}
|
|
const alerts = [
|
|
...[...accounts.values()]
|
|
.filter((item) => item.status !== 'connected')
|
|
.map((item) => ({
|
|
severity: 'warning',
|
|
code: 'account_disconnected',
|
|
accountId: item.id,
|
|
message: `WhatsApp 账号 ${item.name || item.phoneNumber} 未连接`,
|
|
})),
|
|
...jobs
|
|
.filter((item) => item.status === 'dead')
|
|
.slice(-20)
|
|
.map((item) => ({
|
|
severity: 'critical',
|
|
code: 'outbound_dead',
|
|
accountId: item.accountId,
|
|
jobId: item.id,
|
|
message: item.lastError || 'WhatsApp 消息发送任务进入死信',
|
|
})),
|
|
...deliveries
|
|
.filter((item) => item.status === 'dead')
|
|
.slice(-20)
|
|
.map((item) => ({
|
|
severity: 'critical',
|
|
code: 'webhook_dead',
|
|
accountId: item.accountId,
|
|
deliveryId: item.id,
|
|
message: item.lastError || 'WhatsApp Webhook 投递进入死信',
|
|
})),
|
|
]
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
accounts: {
|
|
total: accounts.size,
|
|
connected: [...accounts.values()].filter((item) => item.status === 'connected').length,
|
|
disconnected: [...accounts.values()].filter((item) => item.status !== 'connected').length,
|
|
},
|
|
outbound: {
|
|
queued: jobs.filter((item) => ['queued', 'retrying', 'sending'].includes(item.status)).length,
|
|
dead: jobs.filter((item) => item.status === 'dead').length,
|
|
sent: jobs.filter((item) => item.status === 'sent').length,
|
|
riskBackoff: jobs.filter((item) => item.riskBackoff).length,
|
|
averageLatencyMs: sentLatencies.length
|
|
? Math.round(sentLatencies.reduce((sum, value) => sum + value, 0) / sentLatencies.length)
|
|
: 0,
|
|
},
|
|
webhooks: {
|
|
queued: deliveries.filter((item) => ['pending', 'retrying', 'delivering'].includes(item.status)).length,
|
|
dead: deliveries.filter((item) => item.status === 'dead').length,
|
|
delivered: deliveries.filter((item) => item.status === 'delivered').length,
|
|
},
|
|
cache: { chats, messages: cachedMessages },
|
|
alerts,
|
|
generatedAt: new Date().toISOString(),
|
|
},
|
|
})
|
|
})
|
|
app.get('/accounts', (_req, res) => res.json({ code: 0, data: [...accounts.values()].map(accountPublic) }))
|
|
app.post('/accounts', async (req, res) => {
|
|
const phoneNumber = String(req.body.phoneNumber || '').replace(/\D/g, '')
|
|
if (!/^\d{7,15}$/.test(phoneNumber)) return res.status(400).json({ code: 400, message: '请输入包含国家代码的 WhatsApp 手机号' })
|
|
const existing = [...accounts.values()].find((item) => item.phoneNumber === phoneNumber)
|
|
if (existing) return res.status(409).json({ code: 409, message: '该 WhatsApp 手机号已存在', data: accountPublic(existing) })
|
|
const account = {
|
|
id: crypto.randomUUID(),
|
|
name: String(req.body.name || `WhatsApp ${phoneNumber.slice(-4)}`).slice(0, 80),
|
|
phoneNumber,
|
|
ownerRef: String(req.body.ownerRef || '').slice(0, 120),
|
|
status: 'created',
|
|
connectionState: 'idle',
|
|
createdAt: new Date().toISOString(),
|
|
updatedAt: new Date().toISOString(),
|
|
}
|
|
accounts.set(account.id, account)
|
|
await saveAccounts()
|
|
await loadStore(account.id)
|
|
res.status(201).json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.get('/accounts/:id', (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.patch('/accounts/:id/profile', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const sock = accountSocket(account.id)
|
|
const name = req.body.name === undefined ? undefined : String(req.body.name || '').trim().slice(0, 80)
|
|
const status = req.body.status === undefined ? undefined : String(req.body.status || '').trim().slice(0, 139)
|
|
if (name !== undefined) {
|
|
if (!name) return res.status(400).json({ code: 400, message: '账号名称不能为空' })
|
|
await sock.updateProfileName(name)
|
|
account.name = name
|
|
if (account.user) account.user.name = name
|
|
}
|
|
if (status !== undefined) await sock.updateProfileStatus(status)
|
|
account.profileStatus = status === undefined ? account.profileStatus : status
|
|
account.profileUpdatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
const profileUpload = multer({ storage: multer.memoryStorage(), limits: { fileSize: 8 * 1024 * 1024 } })
|
|
app.post('/accounts/:id/profile/picture', profileUpload.single('file'), async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
if (!req.file?.mimetype?.startsWith('image/')) {
|
|
return res.status(400).json({ code: 400, message: '请选择有效图片' })
|
|
}
|
|
const sock = accountSocket(account.id)
|
|
if (!sock.user?.id) return res.status(409).json({ code: 409, message: '账号尚未连接' })
|
|
await sock.updateProfilePicture(jidNormalizedUser(sock.user.id), req.file.buffer)
|
|
account.profilePictureUpdatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.post('/accounts/:id/start', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
await connectAccount(account.id)
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.post('/accounts/:id/pairing-code', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const phoneNumber = String(req.body.phoneNumber || account.phoneNumber || '').replace(/\D/g, '')
|
|
if (!/^\d{7,15}$/.test(phoneNumber)) return res.status(400).json({ code: 400, message: '请输入包含国家代码的 WhatsApp 手机号' })
|
|
|
|
if (account.status === 'logged_out' || (account.lastError && !sessions.has(account.id))) {
|
|
await resetPairingSession(account.id)
|
|
}
|
|
|
|
let lastError
|
|
for (let attempt = 0; attempt < 2; attempt++) {
|
|
if (attempt > 0) await resetPairingSession(account.id)
|
|
await connectAccount(account.id)
|
|
const sock = accountSocket(account.id)
|
|
if (sock.authState.creds.registered) return res.status(409).json({ code: 409, message: '账号已经完成配对' })
|
|
try {
|
|
await waitForPairingReady(account.id, sock)
|
|
const code = await sock.requestPairingCode(phoneNumber)
|
|
account.pairingCode = code
|
|
account.status = 'waiting_pairing'
|
|
account.connectionState = 'pairing'
|
|
delete account.lastError
|
|
account.updatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
log.info({ accountId: account.id, attempt: attempt + 1 }, 'WhatsApp pairing code generated')
|
|
return res.json({ code: 0, data: { pairingCode: code, account: accountPublic(account) } })
|
|
} catch (error) {
|
|
lastError = error
|
|
log.warn({ accountId: account.id, attempt: attempt + 1, error }, 'WhatsApp pairing attempt failed')
|
|
if (!isTransientPairingError(error) || attempt === 1) break
|
|
}
|
|
}
|
|
throw Object.assign(new Error('暂时无法连接 WhatsApp，请稍后重试'), {
|
|
status: 503,
|
|
cause: lastError,
|
|
})
|
|
})
|
|
app.get('/accounts/:id/auth/state', (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.get('/accounts/:id/devices', (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
res.json({
|
|
code: 0,
|
|
data: [{
|
|
id: 'gateway-current',
|
|
name: 'Lingxi Crossborder',
|
|
platform: 'WhatsApp linked device',
|
|
current: true,
|
|
status: account.status,
|
|
connectedAt: account.lastConnectedAt,
|
|
user: account.user,
|
|
}],
|
|
})
|
|
})
|
|
app.delete('/accounts/:id/devices/:deviceId', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
if (req.params.deviceId !== 'gateway-current') {
|
|
return res.status(404).json({ code: 404, message: '设备不存在' })
|
|
}
|
|
if (String(req.body.confirm || '') !== account.id) {
|
|
return res.status(400).json({ code: 400, message: '终止设备需要提交账号 ID 进行确认' })
|
|
}
|
|
const sock = sessions.get(account.id)?.sock
|
|
if (sock) await sock.logout()
|
|
account.status = 'logged_out'
|
|
account.connectionState = 'logged_out'
|
|
account.deviceTerminatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
res.json({ code: 0, data: { success: true, deviceId: req.params.deviceId } })
|
|
})
|
|
app.put('/accounts/:id/webhook', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const url = String(req.body.url || '').trim()
|
|
if (url && !/^https?:\/\//i.test(url)) return res.status(400).json({ code: 400, message: 'Webhook URL 无效' })
|
|
account.webhookUrl = url
|
|
const secret = String(req.body.secret || accountWebhookSecret(account))
|
|
account.webhookSecretEncrypted = encryptCredential(secret)
|
|
delete account.webhookSecret
|
|
account.webhookActive = req.body.active !== false
|
|
account.updatedAt = new Date().toISOString()
|
|
await saveAccounts()
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.put('/accounts/:id/proxy', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const proxyUrl = String(req.body.url || '').trim()
|
|
if (proxyUrl && !/^https?:\/\//i.test(proxyUrl)) {
|
|
return res.status(400).json({ code: 400, message: '代理地址必须使用 http:// 或 https://' })
|
|
}
|
|
if (proxyUrl && req.body.test !== false) {
|
|
const agent = new HttpsProxyAgent(proxyUrl)
|
|
await new Promise((resolve, reject) => {
|
|
const request = https.get('https://web.whatsapp.com/', {
|
|
agent,
|
|
timeout: 12_000,
|
|
headers: { 'user-agent': 'Lingxi-WhatsApp-Gateway/1.0' },
|
|
}, (response) => {
|
|
response.resume()
|
|
if (Number(response.statusCode || 500) >= 500) {
|
|
reject(Object.assign(new Error(`代理测试 HTTP ${response.statusCode}`), { status: 502 }))
|
|
return
|
|
}
|
|
resolve()
|
|
})
|
|
request.on('timeout', () => request.destroy(new Error('代理连接超时')))
|
|
request.on('error', (error) => reject(Object.assign(new Error(`代理不可用：${error.message}`), { status: 502 })))
|
|
})
|
|
}
|
|
account.proxyUrlEncrypted = proxyUrl ? encryptCredential(proxyUrl) : ''
|
|
account.proxyUpdatedAt = new Date().toISOString()
|
|
account.updatedAt = account.proxyUpdatedAt
|
|
await saveAccounts()
|
|
|
|
const session = sessions.get(account.id)
|
|
if (session?.sock) {
|
|
sessions.delete(account.id)
|
|
session.sock.end(new Error('Proxy configuration changed'))
|
|
account.status = 'disconnected'
|
|
account.connectionState = 'reconnecting'
|
|
await connectAccount(account.id)
|
|
}
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.post('/accounts/:id/reauthorize', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
if (String(req.body.confirm || '') !== account.id) {
|
|
return res.status(400).json({ code: 400, message: '重新授权需要提交账号 ID 进行确认' })
|
|
}
|
|
await resetPairingSession(account.id)
|
|
await connectAccount(account.id)
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.get('/accounts/:id/backup', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const store = await loadStore(account.id)
|
|
const accountJobs = [...outboundJobs.values()].filter((item) => item.accountId === account.id)
|
|
const outboundMedia = {}
|
|
for (const job of accountJobs) {
|
|
if (!job.filePath) continue
|
|
const bytes = await fs.readFile(job.filePath).catch(() => null)
|
|
if (bytes) outboundMedia[job.id] = bytes.toString('base64')
|
|
}
|
|
const snapshot = {
|
|
version: 1,
|
|
accountId: account.id,
|
|
createdAt: new Date().toISOString(),
|
|
account: {
|
|
name: account.name,
|
|
phoneNumber: account.phoneNumber,
|
|
ownerRef: account.ownerRef,
|
|
webhookUrl: account.webhookUrl,
|
|
webhookSecretEncrypted: account.webhookSecretEncrypted,
|
|
webhookActive: account.webhookActive,
|
|
proxyUrlEncrypted: account.proxyUrlEncrypted,
|
|
},
|
|
authFiles: await readAuthFiles(account.id),
|
|
store: {
|
|
chats: [...store.chats.values()],
|
|
contacts: [...store.contacts.values()],
|
|
messages: [...store.messages.entries()],
|
|
groups: [...store.groups.values()],
|
|
},
|
|
outboundJobs: accountJobs,
|
|
outboundMedia,
|
|
webhookDeliveries: [...webhookDeliveries.values()].filter((item) => item.accountId === account.id),
|
|
}
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
version: snapshot.version,
|
|
accountId: account.id,
|
|
createdAt: snapshot.createdAt,
|
|
blob: encryptCredential(JSON.stringify(snapshot, BufferJSON.replacer)),
|
|
},
|
|
})
|
|
})
|
|
app.post('/accounts/:id/backup/restore', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
if (String(req.body.confirm || '') !== account.id) {
|
|
return res.status(400).json({ code: 400, message: '恢复备份需要提交账号 ID 进行确认' })
|
|
}
|
|
let snapshot
|
|
try {
|
|
snapshot = JSON.parse(decryptCredential(String(req.body.blob || '')), BufferJSON.reviver)
|
|
} catch {
|
|
return res.status(400).json({ code: 400, message: '备份文件无效或加密密钥不匹配' })
|
|
}
|
|
if (snapshot?.version !== 1 || snapshot?.accountId !== account.id || !snapshot?.authFiles || !snapshot?.store) {
|
|
return res.status(400).json({ code: 400, message: '备份内容与当前账号不匹配' })
|
|
}
|
|
|
|
const session = sessions.get(account.id)
|
|
sessions.delete(account.id)
|
|
if (session?.sock) session.sock.end(new Error('Backup restore'))
|
|
const directory = path.join(authDir, account.id)
|
|
await fs.rm(directory, { recursive: true, force: true })
|
|
await fs.mkdir(directory, { recursive: true, mode: 0o700 })
|
|
for (const [name, content] of Object.entries(snapshot.authFiles)) {
|
|
if (!/^[a-zA-Z0-9_.-]+\.json$/.test(name)) continue
|
|
await fs.writeFile(path.join(directory, name), String(content), { mode: 0o600 })
|
|
}
|
|
await atomicJson(path.join(storeDir, `${account.id}.json`), snapshot.store)
|
|
stores.delete(account.id)
|
|
await loadStore(account.id)
|
|
for (const [key, value] of Object.entries(snapshot.account || {})) {
|
|
if (value !== undefined) account[key] = value
|
|
}
|
|
account.status = 'disconnected'
|
|
account.connectionState = 'reconnecting'
|
|
account.restoredAt = new Date().toISOString()
|
|
for (const [id, job] of outboundJobs.entries()) {
|
|
if (job.accountId === account.id) outboundJobs.delete(id)
|
|
}
|
|
for (const job of snapshot.outboundJobs || []) outboundJobs.set(job.id, job)
|
|
for (const job of snapshot.outboundJobs || []) {
|
|
const base64 = snapshot.outboundMedia?.[job.id]
|
|
if (!base64) continue
|
|
job.filePath = path.join(outboundMediaDir, `${job.id}.bin`)
|
|
await fs.writeFile(job.filePath, Buffer.from(base64, 'base64'), { mode: 0o600 })
|
|
}
|
|
for (const [id, delivery] of webhookDeliveries.entries()) {
|
|
if (delivery.accountId === account.id) webhookDeliveries.delete(id)
|
|
}
|
|
for (const delivery of snapshot.webhookDeliveries || []) webhookDeliveries.set(delivery.id, delivery)
|
|
await Promise.all([saveAccounts(), queueOutboundJobsSave(), queueWebhookDeliveriesSave()])
|
|
await connectAccount(account.id)
|
|
res.json({ code: 0, data: { restored: true, account: accountPublic(account) } })
|
|
})
|
|
app.get('/accounts/:id/webhook-deliveries', async (req, res) => {
|
|
if (!accounts.has(req.params.id)) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const status = String(req.query.status || '')
|
|
const limit = Math.max(1, Math.min(200, Number(req.query.limit || 50)))
|
|
const list = [...webhookDeliveries.values()]
|
|
.filter((item) => item.accountId === req.params.id && (!status || item.status === status))
|
|
.sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || ''))
|
|
.slice(0, limit)
|
|
.map(({ payload: _payload, ...item }) => item)
|
|
res.json({ code: 0, data: list })
|
|
})
|
|
app.post('/accounts/:id/webhook-deliveries/:deliveryId/replay', async (req, res) => {
|
|
if (!accounts.has(req.params.id)) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const delivery = webhookDeliveries.get(req.params.deliveryId)
|
|
if (!delivery || delivery.accountId !== req.params.id) {
|
|
return res.status(404).json({ code: 404, message: 'Webhook 投递记录不存在' })
|
|
}
|
|
delivery.status = 'pending'
|
|
delivery.attempts = 0
|
|
delivery.replayedAt = new Date().toISOString()
|
|
delivery.updatedAt = delivery.replayedAt
|
|
delete delivery.deadAt
|
|
delete delivery.deliveredAt
|
|
delete delivery.lastError
|
|
delete delivery.nextAttemptAt
|
|
await saveWebhookDeliveries()
|
|
scheduleWebhookDelivery(delivery.id)
|
|
const { payload: _payload, ...safe } = delivery
|
|
res.json({ code: 0, data: safe })
|
|
})
|
|
app.post('/accounts/:id/logout', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const sock = sessions.get(account.id)?.sock
|
|
if (sock) await sock.logout()
|
|
account.status = 'logged_out'
|
|
account.connectionState = 'logged_out'
|
|
await saveAccounts()
|
|
res.json({ code: 0, data: accountPublic(account) })
|
|
})
|
|
app.delete('/accounts/:id', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const sock = sessions.get(account.id)?.sock
|
|
if (sock) await sock.end(new Error('account removed'))
|
|
sessions.delete(account.id)
|
|
accounts.delete(account.id)
|
|
stores.delete(account.id)
|
|
await Promise.all([
|
|
fs.rm(path.join(authDir, account.id), { recursive: true, force: true }),
|
|
fs.rm(path.join(storeDir, `${account.id}.json`), { force: true }),
|
|
saveAccounts(),
|
|
])
|
|
res.json({ code: 0, data: { success: true } })
|
|
})
|
|
|
|
app.get('/accounts/:id/chats', async (req, res) => {
|
|
const store = await loadStore(req.params.id)
|
|
const list = [...store.chats.values()]
|
|
.filter((item) => item.id && !String(item.id).endsWith('@broadcast'))
|
|
.sort((a, b) => Number(b.timestamp || 0) - Number(a.timestamp || 0))
|
|
res.json({ code: 0, data: list })
|
|
})
|
|
app.get('/accounts/:id/chats/:jid/history', async (req, res) => {
|
|
const store = await loadStore(req.params.id)
|
|
const limit = Math.max(1, Math.min(2000, Number(req.query.limit || 100)))
|
|
const list = store.messages.get(req.params.jid) || []
|
|
const before = String(req.query.before || req.query.cursor || '')
|
|
const after = String(req.query.after || '')
|
|
let start = 0
|
|
let end = list.length
|
|
if (before) {
|
|
const index = list.findIndex((item) => item.id === before)
|
|
if (index >= 0) end = index
|
|
}
|
|
if (after) {
|
|
const index = list.findIndex((item) => item.id === after)
|
|
if (index >= 0) start = index + 1
|
|
}
|
|
const page = after
|
|
? list.slice(start, start + limit)
|
|
: list.slice(Math.max(start, end - limit), end)
|
|
const firstIndex = page.length ? list.findIndex((item) => item.id === page[0].id) : -1
|
|
const lastIndex = page.length ? list.findIndex((item) => item.id === page.at(-1).id) : -1
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
messages: page,
|
|
beforeCursor: firstIndex > 0 ? page[0].id : null,
|
|
afterCursor: lastIndex >= 0 && lastIndex < list.length - 1 ? page.at(-1).id : null,
|
|
hasMoreBefore: firstIndex > 0,
|
|
hasMoreAfter: lastIndex >= 0 && lastIndex < list.length - 1,
|
|
totalCached: list.length,
|
|
},
|
|
})
|
|
})
|
|
app.post('/accounts/:id/chats/:jid/history/sync', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const store = await loadStore(req.params.id)
|
|
const jid = req.params.jid
|
|
const pages = Math.max(1, Math.min(50, Number(req.body.pages || 2)))
|
|
const count = Math.max(1, Math.min(50, Number(req.body.count || 50)))
|
|
const initialList = store.messages.get(jid) || []
|
|
const initialCount = initialList.length
|
|
let requestedPages = 0
|
|
|
|
for (let page = 0; page < pages; page++) {
|
|
const before = store.messages.get(jid) || []
|
|
const oldest = before[0]
|
|
if (!oldest?.raw?.key || !oldest.timestamp) break
|
|
const beforeCount = before.length
|
|
const beforeOldestId = oldest.id
|
|
await sock.fetchMessageHistory(count, oldest.raw.key, oldest.timestamp)
|
|
requestedPages++
|
|
|
|
let changed = false
|
|
for (let attempt = 0; attempt < 20; attempt++) {
|
|
await new Promise((resolve) => setTimeout(resolve, 400))
|
|
const current = store.messages.get(jid) || []
|
|
if (current.length > beforeCount || current[0]?.id !== beforeOldestId) {
|
|
changed = true
|
|
break
|
|
}
|
|
}
|
|
if (!changed) break
|
|
}
|
|
|
|
const currentCount = (store.messages.get(jid) || []).length
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
requestedPages,
|
|
messages: currentCount,
|
|
added: Math.max(0, currentCount - initialCount),
|
|
},
|
|
})
|
|
})
|
|
app.get('/accounts/:id/chats/:jid/identity', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const store = await loadStore(req.params.id)
|
|
const jid = req.params.jid
|
|
if (!jid.endsWith('@g.us') && !jid.endsWith('@newsletter')) {
|
|
void sock.presenceSubscribe(jid).catch(() => undefined)
|
|
}
|
|
let phoneJid = jid
|
|
if (jid.endsWith('@lid')) {
|
|
try {
|
|
phoneJid = await sock.signalRepository?.lidMapping?.getPNForLID?.(jid) || jid
|
|
} catch {
|
|
phoneJid = jid
|
|
}
|
|
}
|
|
phoneJid = jidNormalizedUser(phoneJid)
|
|
const contact = store.contacts.get(jid) || store.contacts.get(phoneJid) || {}
|
|
const group = jid.endsWith('@g.us')
|
|
? await sock.groupMetadata(jid).catch(() => store.groups.get(jid))
|
|
: null
|
|
const includeProfile = req.query.includeProfile !== 'false'
|
|
const avatarUrl = includeProfile
|
|
? await sock.profilePictureUrl(jid, 'image')
|
|
.catch(() => phoneJid !== jid ? sock.profilePictureUrl(phoneJid, 'image').catch(() => '') : '')
|
|
: ''
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
id: jid,
|
|
canonicalJid: phoneJid.endsWith('@s.whatsapp.net') ? phoneJid : jid,
|
|
name: group?.subject || contact.name || contact.notify || contact.verifiedName || jid.split('@')[0],
|
|
phone: phoneJid.endsWith('@s.whatsapp.net') ? phoneJid.split('@')[0] : '',
|
|
avatarUrl,
|
|
chatType: group ? 'group' : jid.endsWith('@newsletter') ? 'channel' : 'private',
|
|
participants: group?.participants || [],
|
|
description: group?.desc || '',
|
|
},
|
|
})
|
|
})
|
|
app.post('/accounts/:id/contacts/check', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const numbers = (Array.isArray(req.body.numbers) ? req.body.numbers : [req.body.phoneNumber])
|
|
.map((item) => String(item || '').replace(/\D/g, ''))
|
|
.filter(Boolean)
|
|
const result = await sock.onWhatsApp(...numbers)
|
|
res.json({ code: 0, data: result })
|
|
})
|
|
app.get('/accounts/:id/contacts', async (req, res) => {
|
|
const store = await loadStore(req.params.id)
|
|
const query = String(req.query.query || '').trim().toLowerCase()
|
|
const cursor = Math.max(0, Number(req.query.cursor || 0))
|
|
const limit = Math.max(1, Math.min(200, Number(req.query.limit || 50)))
|
|
const contacts = [...store.contacts.values()]
|
|
.map((contact) => ({
|
|
...jsonSafe(contact),
|
|
id: String(contact.id || ''),
|
|
name: contact.name || contact.notify || contact.verifiedName || '',
|
|
phone: String(contact.id || '').endsWith('@s.whatsapp.net')
|
|
? String(contact.id).split('@')[0]
|
|
: '',
|
|
}))
|
|
.filter((contact) => !query || [
|
|
contact.id,
|
|
contact.name,
|
|
contact.phone,
|
|
contact.verifiedName,
|
|
contact.notify,
|
|
].some((value) => String(value || '').toLowerCase().includes(query)))
|
|
.sort((a, b) => String(a.name || a.id).localeCompare(String(b.name || b.id)))
|
|
const items = contacts.slice(cursor, cursor + limit)
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
items,
|
|
nextCursor: cursor + items.length < contacts.length ? String(cursor + items.length) : null,
|
|
total: contacts.length,
|
|
},
|
|
})
|
|
})
|
|
|
|
app.post('/accounts/:id/messages/send', async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const jid = String(req.body.jid || '')
|
|
const text = String(req.body.text || '')
|
|
if (!jid || !text.trim()) return res.status(400).json({ code: 400, message: 'jid 和消息正文必填' })
|
|
const idempotencyKey = String(
|
|
req.body.idempotencyKey
|
|
|| req.get('idempotency-key')
|
|
|| crypto.randomUUID(),
|
|
).slice(0, 160)
|
|
const existing = [...outboundJobs.values()].find((item) =>
|
|
item.accountId === account.id && item.idempotencyKey === idempotencyKey,
|
|
)
|
|
if (existing) {
|
|
res.json({ code: 0, data: existing.result || queuedOutboundMessage(existing) })
|
|
return
|
|
}
|
|
const job = {
|
|
id: crypto.randomUUID(),
|
|
idempotencyKey,
|
|
clientMessageId: String(req.body.clientMessageId || crypto.randomBytes(10).toString('hex').toUpperCase()),
|
|
accountId: account.id,
|
|
jid,
|
|
text,
|
|
quotedMessageId: req.body.quotedMessageId ? String(req.body.quotedMessageId) : undefined,
|
|
status: 'queued',
|
|
attempts: 0,
|
|
createdAt: new Date().toISOString(),
|
|
updatedAt: new Date().toISOString(),
|
|
}
|
|
outboundJobs.set(job.id, job)
|
|
await queueOutboundJobsSave()
|
|
const result = await processOutboundJob(job.id)
|
|
res.json({ code: 0, data: result || queuedOutboundMessage(job) })
|
|
})
|
|
app.post('/accounts/:id/messages/action', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const jid = String(req.body.jid || '')
|
|
const action = String(req.body.action || '')
|
|
if (!jid || !action) return res.status(400).json({ code: 400, message: 'jid 和 action 必填' })
|
|
|
|
const targetId = String(req.body.messageId || '')
|
|
const target = targetId ? storedMessage(req.params.id, jid, targetId) : null
|
|
const targetKey = target?.raw?.key
|
|
let result
|
|
if (action === 'react') {
|
|
if (!targetKey) return res.status(404).json({ code: 404, message: '目标消息不存在' })
|
|
result = await sock.sendMessage(jid, {
|
|
react: { text: String(req.body.emoji || ''), key: targetKey },
|
|
})
|
|
} else if (action === 'edit') {
|
|
if (!targetKey) return res.status(404).json({ code: 404, message: '目标消息不存在' })
|
|
const text = String(req.body.text || '').trim()
|
|
if (!text) return res.status(400).json({ code: 400, message: '编辑后的消息不能为空' })
|
|
result = await sock.sendMessage(jid, { text, edit: targetKey })
|
|
} else if (action === 'delete') {
|
|
if (!targetKey) return res.status(404).json({ code: 404, message: '目标消息不存在' })
|
|
result = await sock.sendMessage(jid, { delete: targetKey })
|
|
} else if (action === 'forward') {
|
|
if (!target?.raw) return res.status(404).json({ code: 404, message: '目标消息不存在' })
|
|
const destinationJid = String(req.body.destinationJid || jid)
|
|
result = await sock.sendMessage(destinationJid, { forward: target.raw, force: true })
|
|
} else if (action === 'poll') {
|
|
const name = String(req.body.name || '').trim()
|
|
const values = (Array.isArray(req.body.values) ? req.body.values : [])
|
|
.map((item) => String(item || '').trim())
|
|
.filter(Boolean)
|
|
if (!name || values.length < 2) {
|
|
return res.status(400).json({ code: 400, message: '投票标题和至少两个选项必填' })
|
|
}
|
|
result = await sock.sendMessage(jid, {
|
|
poll: {
|
|
name,
|
|
values,
|
|
selectableCount: Math.max(1, Math.min(values.length, Number(req.body.selectableCount || 1))),
|
|
},
|
|
})
|
|
} else if (action === 'location') {
|
|
const latitude = Number(req.body.latitude)
|
|
const longitude = Number(req.body.longitude)
|
|
if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) {
|
|
return res.status(400).json({ code: 400, message: '经纬度格式不正确' })
|
|
}
|
|
result = await sock.sendMessage(jid, {
|
|
location: {
|
|
degreesLatitude: latitude,
|
|
degreesLongitude: longitude,
|
|
name: String(req.body.name || ''),
|
|
address: String(req.body.address || ''),
|
|
},
|
|
})
|
|
} else if (action === 'contact') {
|
|
const displayName = String(req.body.displayName || '').trim()
|
|
const phone = String(req.body.phone || '').replace(/[^\d+]/g, '')
|
|
if (!displayName || !phone) {
|
|
return res.status(400).json({ code: 400, message: '联系人姓名和手机号必填' })
|
|
}
|
|
const digits = phone.replace(/\D/g, '')
|
|
const vcard = [
|
|
'BEGIN:VCARD',
|
|
'VERSION:3.0',
|
|
`FN:${displayName.replace(/[\r\n:;]/g, ' ')}`,
|
|
`TEL;type=CELL;type=VOICE;waid=${digits}:${phone.startsWith('+') ? phone : `+${phone}`}`,
|
|
'END:VCARD',
|
|
].join('\n')
|
|
result = await sock.sendMessage(jid, {
|
|
contacts: { displayName, contacts: [{ displayName, vcard }] },
|
|
})
|
|
} else if (action === 'disappearing') {
|
|
const seconds = req.body.seconds === false ? false : Number(req.body.seconds || 0)
|
|
const allowed = [false, 0, 86400, 604800, 7776000]
|
|
if (!allowed.includes(seconds)) {
|
|
return res.status(400).json({ code: 400, message: '临时消息仅支持关闭、24 小时、7 天或 90 天' })
|
|
}
|
|
result = await sock.sendMessage(jid, { disappearingMessagesInChat: seconds || false })
|
|
} else {
|
|
return res.status(400).json({ code: 400, message: '不支持的消息操作' })
|
|
}
|
|
|
|
const message = result?.message ? upsertMessage(req.params.id, result) : null
|
|
res.json({ code: 0, data: message || jsonSafe(result || { success: true }) })
|
|
})
|
|
app.get('/accounts/:id/outbound-jobs', async (req, res) => {
|
|
if (!accounts.has(req.params.id)) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const status = String(req.query.status || '')
|
|
const limit = Math.max(1, Math.min(200, Number(req.query.limit || 50)))
|
|
const list = [...outboundJobs.values()]
|
|
.filter((item) => item.accountId === req.params.id && (!status || item.status === status))
|
|
.sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || ''))
|
|
.slice(0, limit)
|
|
.map(({ text: _text, caption: _caption, result: _result, filePath: _filePath, ...item }) => item)
|
|
res.json({ code: 0, data: list })
|
|
})
|
|
app.post('/accounts/:id/outbound-jobs/:jobId/retry', async (req, res) => {
|
|
const job = outboundJobs.get(req.params.jobId)
|
|
if (!job || job.accountId !== req.params.id) {
|
|
return res.status(404).json({ code: 404, message: '发送任务不存在' })
|
|
}
|
|
job.status = 'queued'
|
|
job.attempts = 0
|
|
job.updatedAt = new Date().toISOString()
|
|
delete job.deadAt
|
|
delete job.lastError
|
|
delete job.nextAttemptAt
|
|
await queueOutboundJobsSave()
|
|
const result = await processOutboundJob(job.id)
|
|
res.json({ code: 0, data: result || queuedOutboundMessage(job) })
|
|
})
|
|
|
|
const upload = multer({ storage: multer.memoryStorage(), limits: { fileSize: 50 * 1024 * 1024, files: 10 } })
|
|
app.post('/accounts/:id/messages/media', upload.single('file'), async (req, res) => {
|
|
const account = accounts.get(req.params.id)
|
|
if (!account) return res.status(404).json({ code: 404, message: '账号不存在' })
|
|
const jid = String(req.body.jid || '')
|
|
if (!jid || !req.file) return res.status(400).json({ code: 400, message: 'jid 和文件必填' })
|
|
const mime = req.file.mimetype || 'application/octet-stream'
|
|
const caption = String(req.body.caption || '')
|
|
const kind = String(req.body.kind || '')
|
|
if (kind === 'sticker') {
|
|
if (mime !== 'image/webp') {
|
|
return res.status(415).json({ code: 415, message: 'WhatsApp 贴纸必须为 WebP 格式' })
|
|
}
|
|
}
|
|
const idempotencyKey = String(
|
|
req.body.idempotencyKey
|
|
|| req.get('idempotency-key')
|
|
|| crypto.randomUUID(),
|
|
).slice(0, 160)
|
|
const existing = [...outboundJobs.values()].find((item) =>
|
|
item.accountId === account.id && item.idempotencyKey === idempotencyKey,
|
|
)
|
|
if (existing) {
|
|
res.json({ code: 0, data: existing.result || queuedOutboundMessage(existing) })
|
|
return
|
|
}
|
|
const type = kind === 'sticker'
|
|
? 'sticker'
|
|
: mime.startsWith('image/')
|
|
? 'image'
|
|
: mime.startsWith('video/')
|
|
? 'video'
|
|
: mime.startsWith('audio/')
|
|
? 'audio'
|
|
: 'file'
|
|
const job = {
|
|
id: crypto.randomUUID(),
|
|
idempotencyKey,
|
|
clientMessageId: String(req.body.clientMessageId || crypto.randomBytes(10).toString('hex').toUpperCase()),
|
|
accountId: account.id,
|
|
jid,
|
|
type,
|
|
mimetype: mime,
|
|
fileName: req.file.originalname,
|
|
caption,
|
|
ptt: req.body.ptt === 'true',
|
|
status: 'queued',
|
|
attempts: 0,
|
|
createdAt: new Date().toISOString(),
|
|
updatedAt: new Date().toISOString(),
|
|
}
|
|
job.filePath = path.join(outboundMediaDir, `${job.id}.bin`)
|
|
await fs.writeFile(job.filePath, req.file.buffer, { mode: 0o600 })
|
|
outboundJobs.set(job.id, job)
|
|
await queueOutboundJobsSave()
|
|
const result = await processOutboundJob(job.id)
|
|
res.json({ code: 0, data: result || queuedOutboundMessage(job) })
|
|
})
|
|
app.post('/accounts/:id/messages/album', upload.array('files', 10), async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const jid = String(req.body.jid || '')
|
|
const files = Array.isArray(req.files) ? req.files : []
|
|
if (!jid || files.length < 2) {
|
|
return res.status(400).json({ code: 400, message: '相册至少需要两张图片或视频' })
|
|
}
|
|
const unsupported = files.find((file) =>
|
|
!file.mimetype.startsWith('image/') && !file.mimetype.startsWith('video/'),
|
|
)
|
|
if (unsupported) {
|
|
return res.status(415).json({ code: 415, message: 'WhatsApp 相册只支持图片和视频' })
|
|
}
|
|
const imageCount = files.filter((file) => file.mimetype.startsWith('image/')).length
|
|
const videoCount = files.length - imageCount
|
|
const parent = await sock.sendMessage(jid, {
|
|
album: {
|
|
expectedImageCount: imageCount || undefined,
|
|
expectedVideoCount: videoCount || undefined,
|
|
},
|
|
})
|
|
const sent = []
|
|
for (const [index, file] of files.entries()) {
|
|
const caption = index === 0 ? String(req.body.caption || '') : ''
|
|
const content = file.mimetype.startsWith('image/')
|
|
? { image: file.buffer, caption, mimetype: file.mimetype, albumParentKey: parent.key }
|
|
: { video: file.buffer, caption, mimetype: file.mimetype, albumParentKey: parent.key }
|
|
const result = await sock.sendMessage(jid, content)
|
|
const normalized = upsertMessage(req.params.id, result)
|
|
if (normalized) sent.push(normalized)
|
|
}
|
|
res.json({ code: 0, data: { parent: jsonSafe(parent.key), messages: sent } })
|
|
})
|
|
app.get('/accounts/:id/messages/:messageId/media', async (req, res) => {
|
|
const store = await loadStore(req.params.id)
|
|
const message = [...store.messages.values()].flat().find((item) => item.id === req.params.messageId)
|
|
if (!message?.raw) return res.status(404).json({ code: 404, message: '媒体消息不存在' })
|
|
if (!['image', 'audio', 'video', 'file', 'sticker'].includes(message.type)) {
|
|
return res.status(404).json({ code: 404, message: '该消息不是媒体消息' })
|
|
}
|
|
const cacheKey = crypto.createHash('sha256')
|
|
.update(`${req.params.id}:${req.params.messageId}`)
|
|
.digest('hex')
|
|
const cacheFile = path.join(mediaCacheDir, `${cacheKey}.bin`)
|
|
const cacheMetaFile = path.join(mediaCacheDir, `${cacheKey}.json`)
|
|
const cacheMeta = await readJson(cacheMetaFile, null)
|
|
let buffer
|
|
if (cacheMeta?.expiresAt && Date.parse(cacheMeta.expiresAt) > Date.now()) {
|
|
buffer = await fs.readFile(cacheFile).catch(() => null)
|
|
}
|
|
if (!buffer) {
|
|
buffer = await downloadMediaMessage(message.raw, 'buffer', {}, {
|
|
logger: log,
|
|
reuploadRequest: accountSocket(req.params.id).updateMediaMessage,
|
|
})
|
|
await fs.writeFile(cacheFile, buffer, { mode: 0o600 })
|
|
await atomicJson(cacheMetaFile, {
|
|
accountId: req.params.id,
|
|
messageId: req.params.messageId,
|
|
size: buffer.length,
|
|
cachedAt: new Date().toISOString(),
|
|
expiresAt: new Date(Date.now() + 7 * 86400_000).toISOString(),
|
|
})
|
|
}
|
|
res.json({
|
|
code: 0,
|
|
data: {
|
|
base64: buffer.toString('base64'),
|
|
mimetype: message.mimetype || 'application/octet-stream',
|
|
fileName: message.fileName || `${message.id}.bin`,
|
|
},
|
|
})
|
|
})
|
|
app.post('/accounts/:id/chats/:jid/read', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const store = await loadStore(req.params.id)
|
|
const list = store.messages.get(req.params.jid) || []
|
|
const keys = list.filter((item) => !item.fromMe).slice(-50).map((item) => item.raw.key)
|
|
if (keys.length) await sock.readMessages(keys)
|
|
const chat = store.chats.get(req.params.jid)
|
|
if (chat) chat.unreadCount = 0
|
|
scheduleStoreSave(req.params.id)
|
|
res.json({ code: 0, data: { success: true, count: keys.length } })
|
|
})
|
|
app.post('/accounts/:id/chats/:jid/presence', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const presence = ['composing', 'recording', 'paused', 'available', 'unavailable'].includes(req.body.presence)
|
|
? req.body.presence
|
|
: 'composing'
|
|
await sock.sendPresenceUpdate(presence, req.params.jid)
|
|
res.json({ code: 0, data: { success: true } })
|
|
})
|
|
app.post('/accounts/:id/chats/:jid/action', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const store = await loadStore(req.params.id)
|
|
const jid = req.params.jid
|
|
const action = String(req.body.action || '')
|
|
if (action === 'archive' || action === 'unarchive') {
|
|
await sock.chatModify({ archive: action === 'archive', lastMessages: [] }, jid)
|
|
} else if (action === 'pin' || action === 'unpin') {
|
|
await sock.chatModify({ pin: action === 'pin' }, jid)
|
|
} else if (action === 'mute' || action === 'unmute') {
|
|
await sock.chatModify({ mute: action === 'mute' ? Number(req.body.until || Date.now() + 8 * 3600_000) : null }, jid)
|
|
} else if (action === 'mark-unread') {
|
|
await sock.chatModify({ markRead: false, lastMessages: [] }, jid)
|
|
} else if (action === 'clear') {
|
|
const list = store.messages.get(jid) || []
|
|
const lastMessages = list.slice(-1).map((item) => ({
|
|
key: item.raw?.key,
|
|
messageTimestamp: item.timestamp,
|
|
})).filter((item) => item.key)
|
|
await sock.chatModify({ clear: true, lastMessages }, jid)
|
|
store.messages.set(jid, [])
|
|
} else if (action === 'delete') {
|
|
await sock.chatModify({ delete: true, lastMessages: [] }, jid)
|
|
store.chats.delete(jid)
|
|
store.messages.delete(jid)
|
|
} else if (action === 'block' || action === 'unblock') {
|
|
await sock.updateBlockStatus(jid, action === 'block' ? 'block' : 'unblock')
|
|
} else if (action === 'leave' && jid.endsWith('@g.us')) {
|
|
await sock.groupLeave(jid)
|
|
} else {
|
|
return res.status(400).json({ code: 400, message: '不支持的会话操作' })
|
|
}
|
|
scheduleStoreSave(req.params.id)
|
|
res.json({ code: 0, data: { success: true, action } })
|
|
})
|
|
app.post('/accounts/:id/groups', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const subject = String(req.body.subject || '').trim()
|
|
const participants = (req.body.participants || []).map((item) => {
|
|
const value = String(item || '').replace(/\D/g, '')
|
|
return `${value}@s.whatsapp.net`
|
|
})
|
|
if (!subject || !participants.length) return res.status(400).json({ code: 400, message: '群名称和成员必填' })
|
|
const group = await sock.groupCreate(subject, participants)
|
|
res.json({ code: 0, data: jsonSafe(group) })
|
|
})
|
|
app.post('/accounts/:id/groups/:jid/action', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const jid = req.params.jid
|
|
const action = String(req.body.action || '')
|
|
let result
|
|
if (['add', 'remove', 'promote', 'demote'].includes(action)) {
|
|
const participants = (req.body.participants || []).map((item) => String(item).includes('@')
|
|
? String(item)
|
|
: `${String(item).replace(/\D/g, '')}@s.whatsapp.net`)
|
|
result = await sock.groupParticipantsUpdate(jid, participants, action)
|
|
} else if (action === 'subject') result = await sock.groupUpdateSubject(jid, String(req.body.subject || ''))
|
|
else if (action === 'description') result = await sock.groupUpdateDescription(jid, String(req.body.description || ''))
|
|
else if (action === 'invite-code') result = await sock.groupInviteCode(jid)
|
|
else if (action === 'revoke-invite') result = await sock.groupRevokeInvite(jid)
|
|
else return res.status(400).json({ code: 400, message: '不支持的群组操作' })
|
|
res.json({ code: 0, data: jsonSafe(result) })
|
|
})
|
|
app.post('/accounts/:id/groups/join', async (req, res) => {
|
|
const sock = accountSocket(req.params.id)
|
|
const code = String(req.body.code || req.body.inviteLink || '').split('/').pop()
|
|
if (!code) return res.status(400).json({ code: 400, message: '邀请链接或代码必填' })
|
|
const jid = await sock.groupAcceptInvite(code)
|
|
res.json({ code: 0, data: { jid } })
|
|
})
|
|
|
|
app.use((error, req, res, _next) => {
|
|
const status = Number(error.status || error.output?.statusCode || 500)
|
|
log.error({ error, requestId: res.getHeader('x-request-id'), path: req.path }, 'request failed')
|
|
res.status(status).json({ code: status, message: error.message || 'WhatsApp 网关内部错误' })
|
|
})
|
|
|
|
for (const account of await readJson(accountsFile, [])) {
|
|
accounts.set(account.id, account)
|
|
await loadStore(account.id)
|
|
if (account.status !== 'logged_out') {
|
|
void connectAccount(account.id).catch((error) => log.error({ error, accountId: account.id }, 'startup connect failed'))
|
|
}
|
|
}
|
|
|
|
for (const delivery of webhookDeliveries.values()) {
|
|
if (['pending', 'retrying', 'delivering'].includes(delivery.status)) {
|
|
if (delivery.status === 'delivering') delivery.status = 'retrying'
|
|
const delay = Math.max(0, Date.parse(delivery.nextAttemptAt || '') - Date.now())
|
|
scheduleWebhookDelivery(delivery.id, delay)
|
|
}
|
|
}
|
|
for (const job of outboundJobs.values()) {
|
|
if (['queued', 'retrying', 'sending'].includes(job.status)) {
|
|
if (job.status === 'sending') job.status = 'retrying'
|
|
const delay = Math.max(0, Date.parse(job.nextAttemptAt || '') - Date.now())
|
|
scheduleOutboundJob(job.id, delay)
|
|
}
|
|
}
|
|
|
|
app.listen(port, '0.0.0.0', () => log.info({ port }, '灵犀 WhatsApp Gateway listening'))
|