diff --git a/.gitignore b/.gitignore index e0aa16bc..ac94d47f 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,5 @@ data/stores/* data/sessions/* .env .yarn/* -.yarnrc.yml \ No newline at end of file +.yarnrc.yml +/.idea/ diff --git a/public/index.html b/public/index.html index 67a15855..f396cb18 100644 --- a/public/index.html +++ b/public/index.html @@ -244,6 +244,16 @@
+
+
+ + +
+
@@ -510,6 +520,8 @@ let currentPhoneNumber = '' let currentQrTimeOut = '' let lastTimeout = undefined + let qrPollIntervalId = undefined + let statusPollIntervalId = undefined const onBroadcast = function(data) { if (data.phone == currentPhoneNumber) { @@ -712,24 +724,35 @@ table.clear(); sessions.forEach(session => { + const statusLower = (session.status || '').toLowerCase() + const disconnectBtn = ['online', 'connecting'].includes(statusLower) + ? `` + : ''; + const connectBtn = !['online', 'connecting'].includes(statusLower) + ? `` + : ''; + const buttons = ` + + ${connectBtn} + ${disconnectBtn} + + `; table.row.add([ session.label, session.display_phone_number, session.status, session.server, - ` - - - - ` + buttons ]).draw(false); }); } @@ -760,6 +783,27 @@ }); + //Disconnect button + $('#sessionsTable tbody').on('click', '.disconnect-btn', function() { + const number = $(this).data('number'); + const token = localStorage.getItem(tokenKey); + if (!token) { alert('Token não encontrado.'); return } + fetch(`${apiUrl}/sessions/${number}/disconnect`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${token}` + } + }) + .then(response => { + if (!response.ok) throw new Error('Erro ao desconectar sessão.'); + alert('Sessão desconectada.'); + // opcional: recarrega dados + fetchSessions(token, false); + }) + .catch(error => alert(error.message)); + }); + //Edit button $('#sessionsTable tbody').on('click', '.edit-btn', function () { const number = $(this).data('number'); @@ -793,6 +837,7 @@ //Edit form for session number function populateEditForm(session) { + $('#provider').val(session.provider || 'baileys'); $('#label').val(session.label); $('#authToken').val(session.authToken); $('#wavoipToken').val(session.wavoipToken); @@ -886,6 +931,7 @@ if (token && number) { const sessionData = { + provider: $('#provider').val(), label: $('#label').val(), authToken: $('#authToken').val(), wavoipToken: $('#wavoipToken').val(), @@ -969,15 +1015,20 @@ return response.json(); }) .then(data => { - if (data.status === 'offline' || data.status === 'disconnected') { - attemptRegister(number, token); - } else if (data.status === 'connecting') { - connectToWebSocket(number, data.qrTimeoutMs); - } else if (data.status === 'online') { - document.getElementById('qrCodeContainer').innerHTML = ''; - document.getElementById('qrCodeMessage').textContent = 'Sessão já está Conectada'; + const provider = data.provider || 'baileys'; + if (provider === 'whatsmeow') { + startWhatsmeowConnectFlow(number, token); } else { - console.log("Status não está offline nem connecting."); + if (data.status === 'offline' || data.status === 'disconnected') { + attemptRegister(number, token); + } else if (data.status === 'connecting') { + connectToWebSocket(number, data.qrTimeoutMs); + } else if (data.status === 'online') { + document.getElementById('qrCodeContainer').innerHTML = ''; + document.getElementById('qrCodeMessage').textContent = 'Sessão já está Conectada'; + } else { + console.log("Status não está offline nem connecting."); + } } }) .catch(error => { @@ -1062,10 +1113,67 @@ } } + // Whatsmeow connect flow + function startWhatsmeowConnectFlow(number, token) { + // clear old timers + if (qrPollIntervalId) { clearInterval(qrPollIntervalId); qrPollIntervalId = undefined } + if (statusPollIntervalId) { clearInterval(statusPollIntervalId); statusPollIntervalId = undefined } + + currentPhoneNumber = number + document.getElementById('qrCodeMessage').textContent = 'Preparando conexão (whatsmeow)...'; + document.getElementById('qrCodeContainer').innerHTML = ''; + + // trigger adapter connect via UNO + fetch(`${apiUrl}/sessions/${number}/connect`, { + method: 'POST', + headers: { Authorization: `Bearer ${token}` } + }).catch(_ => {}) + + // poll QR image + qrPollIntervalId = setInterval(() => { + fetch(`${apiUrl}/sessions/${number}/qr`, { headers: { Authorization: `Bearer ${token}` } }) + .then(resp => { + if (resp.ok) { + return resp.text() + } + throw new Error('no qr') + }) + .then(base64 => { + const img = `data:image/png;base64,${base64}` + const qrcodeDiv = document.getElementById('qrCodeContainer'); + qrcodeDiv.innerHTML = `QR Code`; + document.getElementById('qrCodeMessage').textContent = 'Leia o QR Code Abaixo'; + }) + .catch(_ => {}) + }, 1500) + + // poll status + statusPollIntervalId = setInterval(() => { + fetch(`${apiUrl}/v15.0/${number}`, { headers: { Authorization: `Bearer ${token}` } }) + .then(r => r.json()) + .then(data => { + const statusDiv = document.getElementById('qrCodeMessage'); + if (data.status === 'online') { + statusDiv.textContent = 'Conectado!'; + if (qrPollIntervalId) { clearInterval(qrPollIntervalId); qrPollIntervalId = undefined } + if (statusPollIntervalId) { clearInterval(statusPollIntervalId); statusPollIntervalId = undefined } + document.getElementById('qrCodeContainer').innerHTML = ''; + } else if (data.status === 'connecting') { + statusDiv.textContent = 'Aguardando leitura do QR Code...'; + } else if (data.status === 'disconnected' || data.status === 'offline') { + statusDiv.textContent = `Status: ${data.status}`; + } + }) + .catch(_ => {}) + }, 2000) + } + // disconnect websocket on modal close $('#connectModal').on('hidden.bs.modal', function () { currentPhoneNumber = '' currentQrTimeOut = '' + if (qrPollIntervalId) { clearInterval(qrPollIntervalId); qrPollIntervalId = undefined } + if (statusPollIntervalId) { clearInterval(statusPollIntervalId); statusPollIntervalId = undefined } document.getElementById('qrCodeContainer').innerHTML = ''; document.getElementById('qrCodeMessage').textContent = 'Aguarde o QrCode/PairingCode'; }); @@ -1119,5 +1227,5 @@ } } - - + + diff --git a/src/amqp.ts b/src/amqp.ts index 39c5d8e3..fe8cbd73 100644 --- a/src/amqp.ts +++ b/src/amqp.ts @@ -116,6 +116,7 @@ export const amqpGetChannel = async () => { logger.info('Creating channel...') await amqpConnect() amqpChannel = await amqpChannelModel?.createChannel() + amqpChannel?.setMaxListeners(0) logger.info('Created channel!') } return amqpChannel @@ -340,6 +341,11 @@ export const amqpConsume = async ( const bindingKey = await bindQueue(channel, exchange, queue, routingKey) const bindingKeyDelayed = await bindQueue(channel, exchange, queue, routingKey, true) + // For bridge (direct) exchange, also bind a plain routing key without queue prefix + // so external adapters can publish using only the phone number as routing key. + if (exchange === UNOAPI_EXCHANGE_BRIDGE_NAME && routingKey) { + await channel?.bindQueue(queue, exchange, routingKey) + } channel?.on('close', () => { channel.unbindQueue(queue, exchange, bindingKey) diff --git a/src/broker.ts b/src/broker.ts index 0367dd92..fcff7b1c 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -17,7 +17,7 @@ import { UNOAPI_QUEUE_TRANSCRIBER, } from './defaults' -import { amqpConsume } from './amqp' +import { amqpConsume, amqpGetChannel, extractRoutingKeyFromBindingKey } from './amqp' import { startRedis } from './services/redis' import { OutgoingCloudApi } from './services/outgoing_cloud_api' import { getConfigRedis } from './services/config_redis' @@ -37,6 +37,11 @@ import { addToBlacklist } from './jobs/add_to_blacklist' import { TimerJob } from './jobs/timer' import { TranscriberJob } from './jobs/transcriber' import { OutgoingAmqp } from './services/outgoing_amqp' +import { IncomingBaileys } from './services/incoming_baileys' +import { getClientBaileys } from './services/client_baileys' +import { onNewLoginGenerateToken } from './services/on_new_login_generate_token' +import { ListenerAmqp } from './services/listener_amqp' +import { IncomingJob } from './jobs/incoming' const incomingAmqp: Incoming = new IncomingAmqp(getConfigRedis) const outgoingCloudApi: Outgoing = new OutgoingCloudApi(getConfigRedis, isInBlacklistInRedis, addToBlacklistRedis) @@ -137,13 +142,35 @@ const startBroker = async () => { } logger.info('Starting blacklist add consumer %s', UNOAPI_SERVER_NAME) - await amqpConsume( - UNOAPI_EXCHANGE_BROKER_NAME, - UNOAPI_QUEUE_BLACKLIST_ADD, - '*', - addToBlacklist, - { notifyFailedMessages, prefetch, type: 'topic' } - ) + await amqpConsume(UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_BLACKLIST_ADD, '*', addToBlacklist, { notifyFailedMessages, prefetch, type: 'topic' }) + + // Consume provider-specific outgoing messages for Baileys sessions + const channel = await amqpGetChannel() + await channel?.assertExchange('unoapi.outgoing', 'topic', { durable: true }) + await channel?.assertQueue('outgoing.baileys', { durable: true }) + await channel?.bindQueue('outgoing.baileys', 'unoapi.outgoing', 'provider.baileys.*') + // Ensure Whatsmeow queues exist too (created but not consumed here) + await channel?.assertQueue('outgoing.baileys.dlq', { durable: true }) + await channel?.assertQueue('outgoing.whatsmeow', { durable: true }) + await channel?.bindQueue('outgoing.whatsmeow', 'unoapi.outgoing', 'provider.whatsmeow.*') + await channel?.assertQueue('outgoing.whatsmeow.dlq', { durable: true }) + const listenerAmqpWorker = new ListenerAmqp() + const onNewLogin = onNewLoginGenerateToken(outgoingCloudApi) + const incomingBaileysWorker = new IncomingBaileys(listenerAmqpWorker, getConfigRedis, getClientBaileys, onNewLogin) + const providerJob = new IncomingJob(incomingBaileysWorker, outgoingAmqp, getConfigRedis) + channel?.consume('outgoing.baileys', async (payload) => { + if (!payload) { + return + } + const phone = extractRoutingKeyFromBindingKey(payload.fields.routingKey) + const data = JSON.parse(payload.content.toString()) + try { + await providerJob.consume(phone, data) + } catch (error) { + logger.error(error, 'Error consuming provider.baileys message') + } + channel.ack(payload) + }) logger.info('Unoapi Cloud version %s started broker!', version) } diff --git a/src/controllers/phone_number_controller.ts b/src/controllers/phone_number_controller.ts index b5c71eed..472e4799 100644 --- a/src/controllers/phone_number_controller.ts +++ b/src/controllers/phone_number_controller.ts @@ -23,6 +23,7 @@ export class PhoneNumberController { try { const { phone } = req.params const config = await this.getConfig(phone) + config.provider = config.provider || 'baileys' const store = await config.getStore(phone, config) logger.debug('Session store retrieved!') const { sessionStore } = store @@ -53,6 +54,7 @@ export class PhoneNumberController { for (let i = 0, j = phones.length; i < j; i++) { const phone = phones[i] const config = await this.getConfig(phone) + config.provider = config.provider || 'baileys' const store = await config.getStore(phone, config) const { sessionStore } = store const status = config.provider == 'forwarder' ? 'forwarder' : await sessionStore.getStatus(phone) diff --git a/src/controllers/session_command_controller.ts b/src/controllers/session_command_controller.ts new file mode 100644 index 00000000..d7a3df0f --- /dev/null +++ b/src/controllers/session_command_controller.ts @@ -0,0 +1,89 @@ +import { Request, Response } from 'express' +import axios from 'axios' +import { getConfig } from '../services/config' +import { Reload } from '../services/reload' +import { Logout } from '../services/logout' +import { WHATSMEOW_ADAPTER_BASEURL } from '../defaults' +import logger from '../services/logger' + +export class SessionCommandController { + private getConfig: getConfig + private reloader: Reload + private logout: Logout + + constructor(getConfig: getConfig, reload: Reload, logout: Logout) { + this.getConfig = getConfig + this.reloader = reload + this.logout = logout + } + + public async connect(req: Request, res: Response) { + const { phone } = req.params + try { + const config = await this.getConfig(phone) + const provider = config.provider || 'baileys' + if (provider === 'whatsmeow') { + await axios.post(`${WHATSMEOW_ADAPTER_BASEURL}/sessions/${phone}/connect`) + return res.status(204).send() + } + await this.reloader.run(phone) + return res.status(204).send() + } catch (e) { + logger.error(e, 'Error on connect session %s', phone) + return res.status(500).json({ status: 'error', message: e.message }) + } + } + + public async disconnect(req: Request, res: Response) { + const { phone } = req.params + try { + const config = await this.getConfig(phone) + const provider = config.provider || 'baileys' + if (provider === 'whatsmeow') { + await axios.post(`${WHATSMEOW_ADAPTER_BASEURL}/sessions/${phone}/disconnect`) + return res.status(204).send() + } + await this.logout.run(phone) + return res.status(204).send() + } catch (e) { + logger.error(e, 'Error on disconnect session %s', phone) + return res.status(500).json({ status: 'error', message: e.message }) + } + } + + public async reload(req: Request, res: Response) { + const { phone } = req.params + try { + const config = await this.getConfig(phone) + const provider = config.provider || 'baileys' + if (provider === 'whatsmeow') { + await axios.post(`${WHATSMEOW_ADAPTER_BASEURL}/sessions/${phone}/reload`) + return res.status(204).send() + } + await this.reloader.run(phone) + return res.status(204).send() + } catch (e) { + logger.error(e, 'Error on reload session %s', phone) + return res.status(500).json({ status: 'error', message: e.message }) + } + } + + public async qr(req: Request, res: Response) { + const { phone } = req.params + try { + const config = await this.getConfig(phone) + const provider = config.provider || 'baileys' + if (provider === 'whatsmeow') { + const response = await axios.get(`${WHATSMEOW_ADAPTER_BASEURL}/sessions/${phone}/qr`, { + responseType: 'arraybuffer', + }) + const base64 = Buffer.from(response.data, 'binary').toString('base64') + return res.status(200).send(base64) + } + return res.status(404).json({ status: 'error', message: 'QR not supported for provider' }) + } catch (e) { + logger.error(e, 'Error on get qr for session %s', phone) + return res.status(500).json({ status: 'error', message: e.message }) + } + } +} diff --git a/src/defaults.ts b/src/defaults.ts index 2c239fcc..0f8e2e9c 100644 --- a/src/defaults.ts +++ b/src/defaults.ts @@ -34,6 +34,7 @@ export const WEBHOOK_FORWARD_TIMEOUT_MS = parseInt(process.env.WEBHOOK_TIMEOUT_M // comunication export const UNOAPI_URL = process.env.UNOAPI_URL || 'http://localhost:9876' +export const WHATSMEOW_ADAPTER_BASEURL = process.env.WHATSMEOW_ADAPTER_BASEURL || 'http://provider-whatsmeow:8080' export const WEBHOOK_URL_ABSOLUTE = process.env.WEBHOOK_URL_ABSOLUTE || '' export const WEBHOOK_URL = process.env.WEBHOOK_URL || 'http://localhost:9876/webhooks/fake' export const WEBHOOK_HEADER = process.env.WEBHOOK_HEADER || 'Authorization' diff --git a/src/jobs/bind_bridge.ts b/src/jobs/bind_bridge.ts index ab590f0b..547c5b52 100644 --- a/src/jobs/bind_bridge.ts +++ b/src/jobs/bind_bridge.ts @@ -1,4 +1,5 @@ import { IncomingJob } from './incoming' +import { IncomingWhatsmeow } from './incoming_whatsmeow' import { ListenerJob } from './listener' import { Broadcast } from '../services/broadcast' import { @@ -40,7 +41,7 @@ const processeds = new Map() export class BindBridgeJob { async consume(server: string, { routingKey }: { routingKey: string }) { const config = await getConfigLocal(routingKey) - if (config.provider && !['forwarder', 'baileys'].includes(config.provider!)) { + if (config.provider && !['forwarder', 'baileys', 'whatsmeow'].includes(config.provider!)) { logger.info(`Ignore connecting routingKey ${routingKey} provider ${config.provider}...`) return } @@ -58,32 +59,51 @@ export class BindBridgeJob { const notifyFailedMessages = config.notifyFailedMessages - logger.info('Starting listener baileys consumer %s', routingKey) - await amqpConsume( - UNOAPI_EXCHANGE_BRIDGE_NAME, - `${UNOAPI_QUEUE_LISTENER}.${UNOAPI_SERVER_NAME}`, - routingKey, - listenerJob.consume.bind(listenerJob), - { - notifyFailedMessages, - priority: 5, - prefetch: 1, - type: 'direct' - } - ) + if (config.provider === 'baileys' || !config.provider || config.provider === 'forwarder') { + logger.info('Starting listener baileys consumer %s', routingKey) + await amqpConsume( + UNOAPI_EXCHANGE_BRIDGE_NAME, + `${UNOAPI_QUEUE_LISTENER}.${UNOAPI_SERVER_NAME}`, + routingKey, + listenerJob.consume.bind(listenerJob), + { + notifyFailedMessages, + priority: 5, + prefetch: 1, + type: 'direct', + }, + ) - logger.info('Starting incoming consumer %s', routingKey) - await amqpConsume( - UNOAPI_EXCHANGE_BRIDGE_NAME, - `${UNOAPI_QUEUE_INCOMING}.${UNOAPI_SERVER_NAME}`, - routingKey, - incomingJob.consume.bind(incomingJob), - { - notifyFailedMessages, - priority: 5, - prefetch: 1 /* allways 1 */, - type: 'direct' - } - ) + logger.info('Starting incoming consumer %s', routingKey) + await amqpConsume( + UNOAPI_EXCHANGE_BRIDGE_NAME, + `${UNOAPI_QUEUE_INCOMING}.${UNOAPI_SERVER_NAME}`, + routingKey, + incomingJob.consume.bind(incomingJob), + { + notifyFailedMessages, + priority: 5, + prefetch: 1 /* allways 1 */, + type: 'direct', + }, + ) + } else if (config.provider === 'whatsmeow') { + // For Whatsmeow, consume bridge payloads and forward them into + // the standard outgoing flow so webhooks are delivered. + const incomingWhatsmeow = new IncomingWhatsmeow() + logger.info('Starting incoming whatsmeow consumer %s', routingKey) + await amqpConsume( + UNOAPI_EXCHANGE_BRIDGE_NAME, + `${UNOAPI_QUEUE_INCOMING}.${UNOAPI_SERVER_NAME}`, + routingKey, + incomingWhatsmeow.consume.bind(incomingWhatsmeow), + { + notifyFailedMessages, + priority: 5, + prefetch: 1, + type: 'direct', + }, + ) + } } } diff --git a/src/jobs/incoming_whatsmeow.ts b/src/jobs/incoming_whatsmeow.ts new file mode 100644 index 00000000..ecb4bea9 --- /dev/null +++ b/src/jobs/incoming_whatsmeow.ts @@ -0,0 +1,17 @@ +import { UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_OUTGOING } from '../defaults' +import { amqpPublish } from '../amqp' +import logger from '../services/logger' + +export class IncomingWhatsmeow { + async consume(phone: string, data: object) { + // The adapter publishes Cloud API webhook payloads as { payload } + // We forward them into the standard outgoing flow, where + // OutgoingJob will fan-out per webhook and apply provider-specific handling. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const a = { ...(data as any) } + const payload = a.payload || data + logger.debug('IncomingWhatsmeow phone %s payload enqueued to outgoing', phone) + await amqpPublish(UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_OUTGOING, phone, { payload }, { type: 'topic' }) + } +} + diff --git a/src/jobs/outgoing.ts b/src/jobs/outgoing.ts index fc849cb2..3c4eac9c 100644 --- a/src/jobs/outgoing.ts +++ b/src/jobs/outgoing.ts @@ -10,6 +10,8 @@ import { UNOAPI_QUEUE_WEBHOOK_STATUS_FAILED } from '../defaults' import { extractDestinyPhone, isAudioMessage, isIncomingMessage, jidToPhoneNumber, TYPE_MESSAGES_MEDIA } from '../services/transformer' +import { downloadContentFromMessage } from 'baileys' +import mime from 'mime-types' import logger from '../services/logger' import { getConfig } from '../services/config' import { isUpdateMessage, isFailedStatus } from '../services/transformer' @@ -61,6 +63,19 @@ export class OutgoingJob { // eslint-disable-next-line @typescript-eslint/no-explicit-any const a = { ...data as any } const payload: any = a.payload + // Provider adapters (e.g., Whatsmeow) can publish only { payload } + // to let UnoAPI fan-out to configured webhooks. In this case, + // re-enqueue one message per webhook to reuse the single-webhook path + // below (which applies provider-specific transforms when needed). + if (!a.webhooks && !a.webhook && payload) { + const config = await this.getConfig(phone) + await Promise.all( + config.webhooks.map(async (webhook) => + amqpPublish(UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_OUTGOING, phone, { payload, webhook }, { type: 'topic' }), + ), + ) + return + } if (a.webhooks) { const webhooks: Webhook[] = a.webhooks if (isFailedStatus(payload) && STATUS_FAILED_WEBHOOK_URL) { @@ -137,7 +152,87 @@ export class OutgoingJob { }) ) } + } else if (config.provider == 'whatsmeow') { + // Handle media using Baileys download/decrypt when payload comes with + // directPath/mediaKey (camelCase or snake_case), regardless of direction. + const store = await config.getStore(phone, config) + const { dataStore, mediaStore } = store + if (!isUpdateMessage(payload)) { + payload.entry[0].changes[0].value.messages = await Promise.all( + payload.entry[0].changes[0].value.messages.map(async message => { + if (TYPE_MESSAGES_MEDIA.includes(message.type)) { + const media = message[message.type] || {} + // Try both camelCase and snake_case keys + let mediaKey = media.mediaKey || media.media_key + const directPath = media.directPath || media.direct_path + const url = media.url || (directPath ? `https://mmg.whatsapp.net${directPath}` : undefined) + if (!mediaKey || (!directPath && !url)) { + return message + } + if (typeof mediaKey === 'string') { + // assume base64 + try { + mediaKey = Buffer.from(mediaKey, 'base64') + } catch (_) { /* ignore conversion errors */ } + } else if (typeof mediaKey === 'object' && mediaKey !== null) { + try { + mediaKey = Uint8Array.from(Object.values(mediaKey)) + } catch (_) { /* ignore conversion errors */ } + } + const mapMediaType = { + image: 'image', + video: 'video', + document: 'document', + sticker: 'sticker', + audio: 'audio', + } as const + try { + const stream = await downloadContentFromMessage( + { + mediaKey, + directPath, + url, + } as any, + mapMediaType[message.type], + {}, + ) + const chunks: Buffer[] = [] + for await (const chunk of stream) { + chunks.push(chunk as Buffer) + } + const buffer = Buffer.concat(chunks) + const mimetype: string = media.mime_type || media.mimetype || (media.filename ? (mime.lookup(media.filename) as string) : 'application/octet-stream') + const filePath = mediaStore.getFilePath(phone, message.id, mimetype) + await mediaStore.saveMediaBuffer(filePath, buffer) + const mediaId = `${phone}/${message.id}` + const payloadMedia = { + messaging_product: 'whatsapp', + mime_type: mimetype, + sha256: media.fileSha256 || media.sha256, + file_size: media.fileLength || media.file_size, + id: mediaId, + filename: media.filename || filePath, + } + await dataStore.setMediaPayload(message.id, payloadMedia) + message[message.type].id = mediaId + } catch (err) { + // If download/decrypt fails, leave message as-is + } + } + // Preserve digits-only Cloud payloads from adapters (e.g., whatsmeow) + // to avoid BR-specific normalization that inserts a '9' and + // breaks alignment with contacts[0].wa_id used by Chatwoot. + if (!/^\d+$/.test(message.from)) { + message.from = jidToPhoneNumber(message.from, '') + } + return message + }) + ) + } } + // If this is a status/receipt payload, optionally delay to ensure + // related media messages are delivered first to the webhook consumer. + await sleep(1000) await this.service.sendHttp(phone, a.webhook, payload, {}) } } diff --git a/src/router.ts b/src/router.ts index b1662aa4..db0fef76 100644 --- a/src/router.ts +++ b/src/router.ts @@ -20,6 +20,7 @@ import { SessionController } from './controllers/session_controller' import { BlacklistController } from './controllers/blacklist_controller' import { PairingCodeController } from './controllers/pairing_code_controller' import { ConnectController } from './controllers/connect_controller' +import { SessionCommandController } from './controllers/session_command_controller' import { Server } from 'socket.io' import { OnNewLogin } from './services/socket' import { addToBlacklist } from './services/blacklist' @@ -61,7 +62,7 @@ export const router = ( const pairingCodeController = new PairingCodeController(getConfig, incoming) const connectController = new ConnectController(reload) const timerController = new TimerController() - + const sessionCommandController = new SessionCommandController(getConfig, reload, logout) // Webhook for forward connection router.post('/webhooks/whatsapp/:phone', webhookController.whatsapp.bind(webhookController)) @@ -80,6 +81,10 @@ export const router = ( router.get('/:version/debug_token', indexController.debugToken) router.get('/sessions', middleware, phoneNumberController.list.bind(phoneNumberController)) router.get('/sessions/:phone', sessionController.index.bind(sessionController)) + router.post('/sessions/:phone/connect', middleware, sessionCommandController.connect.bind(sessionCommandController)) + router.post('/sessions/:phone/disconnect', middleware, sessionCommandController.disconnect.bind(sessionCommandController)) + router.post('/sessions/:phone/reload', middleware, sessionCommandController.reload.bind(sessionCommandController)) + router.get('/sessions/:phone/qr', sessionCommandController.qr.bind(sessionCommandController)) router.post('/:phone/contacts', middleware, contactsController.post.bind(contactsController)) router.post('/:version/:phone/register', middleware, registrationController.register.bind(registrationController)) router.post('/:version/:phone/deregister', middleware, registrationController.deregister.bind(registrationController)) diff --git a/src/services/auto_connect.ts b/src/services/auto_connect.ts index b4604b39..7717bf5d 100644 --- a/src/services/auto_connect.ts +++ b/src/services/auto_connect.ts @@ -4,7 +4,8 @@ import { SessionStore } from './session_store' import { Listener } from './listener' import { OnNewLogin } from './socket' import logger from './logger' -import { UNOAPI_SERVER_NAME } from '../defaults' +import { UNOAPI_EXCHANGE_BRIDGE_NAME, UNOAPI_QUEUE_BIND, UNOAPI_SERVER_NAME } from '../defaults' +import { amqpPublish } from '../amqp' export const autoConnect = async ( sessionStore: SessionStore, @@ -20,9 +21,23 @@ export const autoConnect = async ( const phone = phones[i] try { const config = await getConfig(phone) - if (config.provider && !['forwarder', 'baileys'].includes(config.provider)) { + if (config.provider && !['forwarder', 'baileys', 'whatsmeow'].includes(config.provider)) { logger.info(`Ignore connecting phone ${phone} provider ${config.provider}...`) - continue; + continue + } + // Do not attempt local connection for external providers + if (config.provider === 'whatsmeow' || config.provider === 'forwarder') { + // For external providers, proactively request BindBridge to attach + // per-phone consumers on incoming bridge queues. + await amqpPublish( + UNOAPI_EXCHANGE_BRIDGE_NAME, + `${UNOAPI_QUEUE_BIND}.${UNOAPI_SERVER_NAME}`, + '', + { routingKey: phone }, + { type: 'direct' }, + ) + logger.info(`Provider ${config.provider} uses external adapter. Skipping auto connect for ${phone}...`) + continue } if (config.server && config.server !== UNOAPI_SERVER_NAME) { logger.info(`Ignore connecting phone ${phone} server ${config.server} is not server current server ${UNOAPI_SERVER_NAME}...`) diff --git a/src/services/client_baileys.ts b/src/services/client_baileys.ts index 6d7ea07c..96506b99 100644 --- a/src/services/client_baileys.ts +++ b/src/services/client_baileys.ts @@ -367,7 +367,18 @@ export class ClientBaileys implements Client { this.calls.set(from, true) if (this.config.rejectCalls && this.rejectCall) { await this.rejectCall(id, from) - await this.sendMessage(from, { text: this.config.rejectCalls }, {}); + const response = await this.sendMessage(from, { text: this.config.rejectCalls }, {}) + const message = { + key: { + fromMe: true, + remoteJid: from, + id: response.key.id, + }, + message: { + conversation: this.config.rejectCalls, + }, + } + await this.listener.process(this.phone, [message], 'append') logger.info('Rejecting calls %s %s', this.phone, this.config.rejectCalls) } const messageCallsWebhook = this.config.rejectCallsWebhook || this.config.messageCallsWebhook diff --git a/src/services/config.ts b/src/services/config.ts index 2f043ea1..3dad974c 100644 --- a/src/services/config.ts +++ b/src/services/config.ts @@ -73,7 +73,7 @@ export type Config = { sendProfilePicture: boolean authToken: string | undefined authHeader: string | undefined - provider: 'baileys' | 'forwarder' | undefined + provider: 'baileys' | 'forwarder' | 'whatsmeow' | undefined server: string | undefined connectionType: connectionType wavoipToken: string | undefined diff --git a/src/services/incoming_amqp.ts b/src/services/incoming_amqp.ts index e6411681..37391093 100644 --- a/src/services/incoming_amqp.ts +++ b/src/services/incoming_amqp.ts @@ -1,10 +1,25 @@ import { Incoming } from './incoming' -import { amqpPublish } from '../amqp' -import { UNOAPI_EXCHANGE_BRIDGE_NAME, UNOAPI_QUEUE_INCOMING } from '../defaults' -import { jidToPhoneNumber } from './transformer' +import { amqpGetChannel } from '../amqp' +import { jidToPhoneNumber, getGroupId } from './transformer' import { getConfig } from './config' import { generateUnoId } from '../utils/id' +const EXCHANGE = 'unoapi.outgoing' +let initialized = false + +const initExchange = async () => { + if (initialized) { + return + } + const channel = await amqpGetChannel() + await channel?.assertExchange(EXCHANGE, 'topic', { durable: true }) + await channel?.assertQueue('outgoing.baileys', { durable: true }) + await channel?.bindQueue('outgoing.baileys', EXCHANGE, 'provider.baileys.*') + await channel?.assertQueue('outgoing.baileys.dlq', { durable: true }) + // Do not declare or bind whatsmeow queues here; adapter owns its queue/bindings + initialized = true +} + export class IncomingAmqp implements Incoming { private getConfig: getConfig @@ -14,37 +29,55 @@ export class IncomingAmqp implements Incoming { public async send(phone: string, payload: object, options: object = {}) { // eslint-disable-next-line @typescript-eslint/no-explicit-any - const { status, type, to } = payload as any - const config = await this.getConfig(phone); + const pl: any = { ...payload } + // Normalize Graph API reply context for providers expecting either + // context.id or context.message_id. Ensure both are present when one is. + if (pl && typeof pl === 'object' && pl.context && typeof pl.context === 'object') { + if (pl.context.message_id && !pl.context.id) { + pl.context.id = pl.context.message_id + } else if (pl.context.id && !pl.context.message_id) { + pl.context.message_id = pl.context.id + } + } + // Fallback: if "to" is blank, try to extract group id from Chatwoot payload + if (!pl?.to || `${pl.to}`.trim() === '') { + const gid = getGroupId(pl) + if (gid && typeof gid === 'string') { + pl.to = gid + } + } + const { status, type, to } = pl as any + const config = await this.getConfig(phone) + const provider = config.provider || 'baileys' + await initExchange() + const channel = await amqpGetChannel() + const routingKey = `provider.${provider}.${phone}` if (status) { options['type'] = 'direct' options['priority'] = 3 // update status is always middle important - await amqpPublish( - UNOAPI_EXCHANGE_BRIDGE_NAME, - `${UNOAPI_QUEUE_INCOMING}.${config.server!}`, - phone, - { payload, options }, - options - ) + const data = { payload: pl, options } + channel?.publish(EXCHANGE, routingKey, Buffer.from(JSON.stringify(data)), { + contentType: 'application/json', + messageId: (payload as any).message_id, + persistent: true, + }) return { ok: { success: true } } } else if (type) { const id = generateUnoId('INC') if (!options['priority']) { options['priority'] = 5 // send message without bulk is very important } - options['type'] = 'direct' - await amqpPublish( - UNOAPI_EXCHANGE_BRIDGE_NAME, - `${UNOAPI_QUEUE_INCOMING}.${config.server!}`, - phone, - { payload, id, options }, - options - ) + const data = { payload: pl, id, options } + channel?.publish(EXCHANGE, routingKey, Buffer.from(JSON.stringify(data)), { + contentType: 'application/json', + messageId: id, + persistent: true, + }) const ok = { messaging_product: 'whatsapp', contacts: [ { - wa_id: jidToPhoneNumber(to, ''), + wa_id: jidToPhoneNumber(pl.to, ''), }, ], messages: [ @@ -55,7 +88,7 @@ export class IncomingAmqp implements Incoming { } return { ok } } else { - throw `Unknown incoming message ${JSON.stringify(payload)}` + throw `Unknown incoming message ${JSON.stringify(pl)}` } } } diff --git a/src/services/media_store_file.ts b/src/services/media_store_file.ts index bb722e56..421e7858 100644 --- a/src/services/media_store_file.ts +++ b/src/services/media_store_file.ts @@ -2,6 +2,7 @@ import { proto, WAMessage, downloadMediaMessage, downloadContentFromMessage, Con import { getBinMessage, jidToPhoneNumberIfUser, toBuffer } from './transformer' import { writeFile } from 'fs/promises' import { existsSync, mkdirSync, rmSync, createReadStream } from 'fs' +import path from 'path' import { MediaStore, getMediaStore, mediaStores } from './media_store' import mime from 'mime-types' import { Response } from 'express' @@ -187,13 +188,31 @@ export const mediaStoreFile = (phone: string, config: Config, getDataStore: getD const dataStore = await getDataStore(phone, config) const mediaPayload = await dataStore.loadMediaPayload(mediaId!) if (!mediaPayload) { - logger.warn('Media payload nor found %s', mediaId) - return; + logger.warn('Media payload not found %s', mediaId) + return undefined } - const filePath = mediaStore.getFilePath(phone, mediaId!, mediaPayload?.mime_type!) + + let filePath: string | undefined + if (mediaPayload.mime_type) { + filePath = mediaStore.getFilePath(phone, mediaId!, mediaPayload.mime_type) + } else if (mediaPayload.filename) { + logger.debug('Missing mime type for media %s, deriving from filename', mediaId) + const ext = path.extname(mediaPayload.filename) + if (ext) { + filePath = `${phone}/${mediaId}${ext}` + } + } + + if (!filePath) { + logger.warn('Media payload missing mime type and filename %s', mediaId) + return undefined + } + const url = await mediaStore.getDownloadUrl(baseUrl, filePath) + const mimeType = mediaPayload.mime_type || mime.lookup(filePath) const payload = { ...mediaPayload, + ...(mimeType ? { mime_type: mimeType } : {}), url, } return payload diff --git a/src/standalone.ts b/src/standalone.ts index a3ee6bc1..8e6406cc 100644 --- a/src/standalone.ts +++ b/src/standalone.ts @@ -61,7 +61,7 @@ import { Reload } from './services/reload' import { Logout } from './services/logout' import { LogoutAmqp } from './services/logout_amqp' import { ReloadJob } from './jobs/reload' -import { amqpConnect, amqpConsume } from './amqp' +import { amqpConnect, amqpConsume, amqpGetChannel, extractRoutingKeyFromBindingKey } from './amqp' import { startRedis } from './services/redis' import ContactBaileys from './services/contact_baileys' import injectRouteDummy from './services/inject_route_dummy' @@ -72,6 +72,7 @@ import atbl from './jobs/add_to_blacklist' import { MediaJob } from './jobs/media' import { OutgoingJob } from './jobs/outgoing' import { NotificationJob } from './jobs/notification' +import { IncomingJob } from './jobs/incoming' import * as Sentry from '@sentry/node' if (process.env.SENTRY_DSN) { @@ -177,13 +178,35 @@ if (process.env.AMQP_URL) { } logger.info('Starting blacklist add consumer %s', UNOAPI_SERVER_NAME) - amqpConsume( - UNOAPI_EXCHANGE_BROKER_NAME, - UNOAPI_QUEUE_BLACKLIST_ADD, - '*', - atbl, - { notifyFailedMessages, prefetch, type: 'topic' } - ) + amqpConsume(UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_BLACKLIST_ADD, '*', atbl, { notifyFailedMessages, prefetch, type: 'topic' }) + + // Consume provider-specific outgoing messages for Baileys sessions + ;(async () => { + const channel = await amqpGetChannel() + await channel?.assertExchange('unoapi.outgoing', 'topic', { durable: true }) + await channel?.assertQueue('outgoing.baileys', { durable: true }) + await channel?.bindQueue('outgoing.baileys', 'unoapi.outgoing', 'provider.baileys.*') + // Ensure Whatsmeow queues exist too (created but not consumed here) + await channel?.assertQueue('outgoing.baileys.dlq', { durable: true }) + await channel?.assertQueue('outgoing.whatsmeow', { durable: true }) + await channel?.bindQueue('outgoing.whatsmeow', 'unoapi.outgoing', 'provider.whatsmeow.*') + await channel?.assertQueue('outgoing.whatsmeow.dlq', { durable: true }) + const incomingBaileysWorker = new IncomingBaileys(listener, getConfigVar, getClientBaileys, onNewLoginn) + const providerJob = new IncomingJob(incomingBaileysWorker, outgoing, getConfigVar) + channel?.consume('outgoing.baileys', async (payload) => { + if (!payload) { + return + } + const phone = extractRoutingKeyFromBindingKey(payload.fields.routingKey) + const data = JSON.parse(payload.content.toString()) + try { + await providerJob.consume(phone, data) + } catch (error) { + logger.error(error, 'Error consuming provider.baileys message') + } + channel.ack(payload) + }) + })() } else { logger.info('Starting standard mode') }