Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
171 changes: 167 additions & 4 deletions src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,24 @@ export class BaileysStartupService extends ChannelStartupService {
private eventProcessingQueue: Promise<void> = Promise.resolve();
private presenceInterval: NodeJS.Timeout | null = null;

// Uma instância só pode ter uma socket viva. Duas sockets com a mesma sessão se derrubam pelo WhatsApp
// (stream:error conflict/replaced, 440) e, como o 'close' de cada uma reconectava, o loop só parava no restart.
private unsubscribeClientEvents: (() => void) | null = null;
private connectInFlight: Promise<WASocket> | null = null;

// Reconexão com espera: um timer por instância, e a espera cresce a cada queda seguida do mesmo tipo.
// As sequências zeram quando a conexão fica aberta por STABLE_CONNECTION_MS.
private reconnectTimer: NodeJS.Timeout | null = null;
private stableConnectionTimer: NodeJS.Timeout | null = null;
private replacedStreak = 0;
private reconnectFailureStreak = 0;
// logout/exclusão: nenhuma reconexão agendada pode reabrir a instância depois disso.
private disposed = false;
private static readonly REPLACED_BACKOFF_BASE_MS = 10_000;
private static readonly RECONNECT_FAILURE_BACKOFF_BASE_MS = 5_000;
private static readonly RECONNECT_BACKOFF_MAX_MS = 5 * 60_000;
private static readonly STABLE_CONNECTION_MS = 2 * 60_000;

// Erro 463 (LID addressing): cache negativo de PNs sem mapeamento LID, para não
// repetir USync a cada envio — getLIDForPN consulta a rede quando não há mapping local.
private readonly lidUnmappedPns = new Map<string, number>();
Expand All @@ -276,6 +294,9 @@ export class BaileysStartupService extends ChannelStartupService {
}

public async logoutInstance() {
this.disposed = true;
this.clearReconnectTimer();
this.clearStableConnectionTimer();
this.stopPresenceInterval();
this.messageProcessor.onDestroy();
await this.client?.logout('Log out instance: ' + this.instanceName);
Expand Down Expand Up @@ -441,7 +462,23 @@ export class BaileysStartupService extends ChannelStartupService {
const codesToNotReconnect = [DisconnectReason.loggedOut, DisconnectReason.forbidden, 402, 406];
const shouldReconnect = !codesToNotReconnect.includes(statusCode);
if (shouldReconnect) {
await this.connectToWhatsapp(this.phoneNumber);
this.clearStableConnectionTimer();
// O banco passa a refletir a queda antes de reconectar. Sem isso, uma reconexão que falhava ou demorava
// deixava a instância 'close' em memória e 'open' no banco: o painel mostrava conectado e os envios falhavam.
await this.persistConnectionDrop(statusCode, lastDisconnect);

if (statusCode === DisconnectReason.connectionReplaced) {
// 440: outra socket abriu com a mesma sessão. Reconectar na hora derrubaria a outra, que reconectaria
// e derrubaria esta de volta. A espera dobra a cada 440 seguido (10s, 20s, 40s... até 5min), então uma
// sessão aberta em outro servidor deixa de brigar a cada segundo e o log mostra qual instância é.
const delayMs = this.nextBackoffMs('replaced');
this.logger.error(
`[${this.instance.name}] Conexão substituída por outra sessão (440 conflict/replaced, ${this.replacedStreak}ª seguida); reconectando em ${delayMs / 1000}s`,
);
this.scheduleReconnect(delayMs);
} else {
await this.reconnectNow();
}
} else {
this.sendDataWebhook(Events.STATUS_INSTANCE, {
instance: this.instance.name,
Expand Down Expand Up @@ -478,6 +515,7 @@ export class BaileysStartupService extends ChannelStartupService {
}

if (connection === 'open') {
this.markConnectionOpened();
this.instance.wuid = this.client.user.id.replace(/:\d+/, '');
try {
const profilePic = await this.profilePicture(this.instance.wuid);
Expand Down Expand Up @@ -713,6 +751,7 @@ export class BaileysStartupService extends ChannelStartupService {

this.endSession = false;

this.endPreviousClient();
this.client = makeWASocket(socketConfig);

if (this.localSettings.wavoipToken && this.localSettings.wavoipToken.length > 0) {
Expand All @@ -738,6 +777,101 @@ export class BaileysStartupService extends ChannelStartupService {
return this.client;
}

private async persistConnectionDrop(statusCode: number, lastDisconnect: any) {
try {
await this.prismaRepository.instance.update({
where: { id: this.instanceId },
data: {
connectionStatus: 'close',
disconnectionAt: new Date(),
disconnectionReasonCode: statusCode,
disconnectionObject: JSON.stringify(lastDisconnect),
},
});
} catch (error) {
this.logger.warn(`[${this.instance.name}] Não foi possível gravar a queda da conexão: ${error?.message}`);
}
this.sendDataWebhook(Events.CONNECTION_UPDATE, { instance: this.instance.name, ...this.stateConnection });
}

private nextBackoffMs(kind: 'replaced' | 'failure') {
const streak = kind === 'replaced' ? ++this.replacedStreak : ++this.reconnectFailureStreak;
const base =
kind === 'replaced'
? BaileysStartupService.REPLACED_BACKOFF_BASE_MS
: BaileysStartupService.RECONNECT_FAILURE_BACKOFF_BASE_MS;
return Math.min(base * 2 ** (streak - 1), BaileysStartupService.RECONNECT_BACKOFF_MAX_MS);
}

// Reconecta já; se a tentativa falhar, agenda outra com espera em vez de largar a instância fechada para sempre.
private async reconnectNow() {
try {
await this.connectToWhatsapp(this.phoneNumber);
} catch (error) {
const delayMs = this.nextBackoffMs('failure');
this.logger.error(
`[${this.instance.name}] Falha ao reconectar (${this.reconnectFailureStreak}ª seguida): ${error?.message}; nova tentativa em ${delayMs / 1000}s`,
);
this.scheduleReconnect(delayMs);
}
}

private scheduleReconnect(delayMs: number) {
this.clearReconnectTimer();
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
// Alguém já reabriu (API, outra reconexão) ou a instância foi encerrada: não abre outra socket.
if (this.disposed || this.endSession || this.stateConnection.state !== 'close') {
return;
}
this.reconnectNow();
}, delayMs);
}

private clearReconnectTimer() {
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
}

private clearStableConnectionTimer() {
if (this.stableConnectionTimer) {
clearTimeout(this.stableConnectionTimer);
this.stableConnectionTimer = null;
}
}

// Conexão aberta: cancela reconexão pendente e, se ela ficar de pé por um tempo, zera as sequências de espera.
private markConnectionOpened() {
this.clearReconnectTimer();
this.reconnectFailureStreak = 0;
this.clearStableConnectionTimer();
this.stableConnectionTimer = setTimeout(() => {
this.stableConnectionTimer = null;
this.replacedStreak = 0;
}, BaileysStartupService.STABLE_CONNECTION_MS);
}

// Desliga os eventos e encerra a socket anterior antes de criar outra. O unsubscribe vem antes do end()
// de propósito: o 'close' que o end() emite não pode disparar mais uma reconexão.
private endPreviousClient() {
const previous = this.client;

this.unsubscribeClientEvents?.();
this.unsubscribeClientEvents = null;

if (!previous) {
return;
}

try {
previous.end(undefined);
} catch (error) {
this.logger.warn(`[${this.instance.name}] Erro ao encerrar a socket anterior: ${error?.message}`);
}
}

private startPresenceInterval() {
// Limpa interval anterior se existir
this.stopPresenceInterval();
Expand Down Expand Up @@ -775,6 +909,20 @@ export class BaileysStartupService extends ChannelStartupService {
}

public async connectToWhatsapp(number?: string): Promise<WASocket> {
// A reconexão automática do 'close' e uma chamada da API (/instance/restart, /instance/connect) podem chegar
// juntas; a segunda reaproveita a conexão em andamento em vez de abrir outra socket.
if (this.connectInFlight) {
return this.connectInFlight;
}

this.connectInFlight = this.openConnection(number).finally(() => {
this.connectInFlight = null;
});

return this.connectInFlight;
}

private async openConnection(number?: string): Promise<WASocket> {
try {
this.loadChatwoot();
this.loadSettings();
Expand All @@ -795,7 +943,15 @@ export class BaileysStartupService extends ChannelStartupService {

public async reloadConnection(): Promise<WASocket> {
try {
return await this.createClient(this.phoneNumber);
// Mesma trava do connectToWhatsapp: privacidade/foto de perfil chamando junto de uma reconexão não abre
// uma segunda socket.
if (this.connectInFlight) {
return await this.connectInFlight;
}
this.connectInFlight = this.createClient(this.phoneNumber).finally(() => {
this.connectInFlight = null;
});
return await this.connectInFlight;
} catch (error) {
this.logger.error(error);
throw new InternalServerErrorException(error?.toString());
Expand Down Expand Up @@ -1927,10 +2083,17 @@ export class BaileysStartupService extends ChannelStartupService {
};

private eventHandler() {
this.client.ev.process(async (events) => {
const socket = this.client;

this.unsubscribeClientEvents = socket.ev.process(async (events) => {
// Eventos de uma socket já substituída são descartados: um 'close' atrasado dela reabriria a conexão.
if (socket !== this.client) {
return;
}

this.eventProcessingQueue = this.eventProcessingQueue.then(async () => {
try {
if (!this.endSession) {
if (!this.endSession && socket === this.client) {
const database = this.configService.get<Database>('DATABASE');
const settings = await this.findSettings();

Expand Down
10 changes: 10 additions & 0 deletions src/api/services/monitor.service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -483,6 +483,16 @@ export class WAMonitoringService {
});
} finally {
this.logger.warn(`Instance "${instanceName}" - NOT CONNECTION`);
// Limite de QR atingido: a instância fica fechada em memória, e o banco precisa dizer o mesmo. Antes ele
// ficava com o último status gravado ('connecting'/'open'), e quem lê o banco via a instância como viva.
await this.prismaRepository.instance
.updateMany({
where: { name: instanceName },
data: { connectionStatus: 'close', disconnectionAt: new Date() },
})
.catch((error) =>
this.logger.error(`Instance "${instanceName}" - falha ao gravar close no banco: ${error?.message}`),
);
}
});
}
Expand Down