Skip to content
Open
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
4 changes: 4 additions & 0 deletions src/api/controllers/label.controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,4 +12,8 @@ export class LabelController {
public async handleLabel({ instanceName }: InstanceDto, data: HandleLabelDto) {
return await this.waMonitor.waInstances[instanceName].handleLabel(data);
}

public async syncLabels({ instanceName }: InstanceDto) {
return await this.waMonitor.waInstances[instanceName].syncLabels();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -889,6 +889,9 @@ export class EvolutionStartupService extends ChannelStartupService {
public async fetchLabels() {
throw new BadRequestException('Method not available on Evolution Channel');
}
public async syncLabels() {
throw new BadRequestException('Method not available on Evolution Channel');
}
public async handleLabel() {
throw new BadRequestException('Method not available on Evolution Channel');
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1868,6 +1868,9 @@ export class BusinessStartupService extends ChannelStartupService {
public async fetchLabels() {
throw new BadRequestException('Method not available on WhatsApp Business API');
}
public async syncLabels() {
throw new BadRequestException('Method not available on WhatsApp Business API');
}
public async handleLabel() {
throw new BadRequestException('Method not available on WhatsApp Business API');
}
Expand Down
106 changes: 103 additions & 3 deletions src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { getCollectionsDto } from '@api/dto/business.dto';
ο»Ώimport { getCollectionsDto } from '@api/dto/business.dto';
import { OfferCallDto } from '@api/dto/call.dto';
import {
ArchiveChatDto,
Expand Down Expand Up @@ -263,6 +263,10 @@ export class BaileysStartupService extends ChannelStartupService {
private isDeleting = false; // Flag to prevent reconnection during deletion
private logBaileys = this.configService.get<Log>('LOG').BAILEYS;
private eventProcessingQueue: Promise<void> = Promise.resolve();
private labelAssociationSnapshotCollector?: {
observedLabelState: boolean;
labelsByChatId: Map<string, Set<string>>;
};
private _lastStream515At = 0;

// Cumulative history sync counters (reset on new sync or completion)
Expand Down Expand Up @@ -2046,6 +2050,9 @@ export class BaileysStartupService extends ChannelStartupService {

private readonly labelHandle = {
[Events.LABELS_EDIT]: async (label: Label) => {
if (this.labelAssociationSnapshotCollector) {
this.labelAssociationSnapshotCollector.observedLabelState = true;
}
this.sendDataWebhook(Events.LABELS_EDIT, { ...label, instance: this.instance.name });

const labelsRepository = await this.prismaRepository.label.findMany({ where: { instanceId: this.instanceId } });
Expand Down Expand Up @@ -2090,6 +2097,8 @@ export class BaileysStartupService extends ChannelStartupService {
const chatId = data.association.chatId;
const labelId = data.association.labelId;

this.collectLabelAssociationSnapshot(data);

if (data.type === 'add') {
await this.addLabel(labelId, instanceId, chatId);
} else if (data.type === 'remove') {
Expand Down Expand Up @@ -2229,13 +2238,13 @@ export class BaileysStartupService extends ChannelStartupService {

if (events[Events.LABELS_ASSOCIATION]) {
const payload = events[Events.LABELS_ASSOCIATION];
this.labelHandle[Events.LABELS_ASSOCIATION](payload, database);
await this.labelHandle[Events.LABELS_ASSOCIATION](payload, database);
return;
}

if (events[Events.LABELS_EDIT]) {
const payload = events[Events.LABELS_EDIT];
this.labelHandle[Events.LABELS_EDIT](payload);
await this.labelHandle[Events.LABELS_EDIT](payload);
return;
}
}
Expand Down Expand Up @@ -4750,6 +4759,97 @@ export class BaileysStartupService extends ChannelStartupService {
}));
}

public async syncLabels(): Promise<LabelDto[]> {
const collector = {
observedLabelState: false,
labelsByChatId: new Map<string, Set<string>>(),
};

this.labelAssociationSnapshotCollector = collector;
try {
await this.instance.authState.state.keys.set({ 'app-state-sync-version': { regular: null } });
await this.client.resyncAppState(['regular'], true);

await this.eventProcessingQueue.catch(() => undefined);

if (collector.observedLabelState) {
await this.replaceChatLabelsFromSnapshot(this.instanceId, collector.labelsByChatId);
} else {
this.logger.warn(
'labels sync snapshot finished without label state mutations; keeping existing chat label projection',
);
}
} finally {
this.labelAssociationSnapshotCollector = undefined;
}

// Wait for LABELS_EDIT and LABELS_ASSOCIATION events to be processed
await new Promise((resolve) => setTimeout(resolve, 3000));

// Return the now-updated labels from database
return this.fetchLabels();
}

private collectLabelAssociationSnapshot(data: { association: LabelAssociation; type: 'remove' | 'add' }) {
const collector = this.labelAssociationSnapshotCollector;
if (!collector) {
return;
}

collector.observedLabelState = true;
if (data.association.type !== 'label_jid') {
return;
}

const chatId = data.association.chatId;
const labelId = data.association.labelId;
const labels = collector.labelsByChatId.get(chatId) ?? new Set<string>();

if (data.type === 'add') {
labels.add(labelId);
} else {
labels.delete(labelId);
}

if (labels.size) {
collector.labelsByChatId.set(chatId, labels);
} else {
collector.labelsByChatId.delete(chatId);
}
}

private async replaceChatLabelsFromSnapshot(instanceId: string, labelsByChatId: Map<string, Set<string>>) {
await this.prismaRepository.$transaction(async (transaction) => {
await transaction.chat.updateMany({
where: { instanceId },
data: { labels: [] },
});

for (const [chatId, labelSet] of labelsByChatId) {
const labels = [...labelSet].sort();
if (!labels.length) {
continue;
}

await transaction.chat.upsert({
where: {
instanceId_remoteJid: {
instanceId,
remoteJid: chatId,
},
},
update: { labels },
create: {
id: cuid(),
instanceId,
remoteJid: chatId,
labels,
},
});
}
});
}

public async handleLabel(data: HandleLabelDto) {
const whatsappContact = await this.whatsappNumber({ numbers: [data.number] });
if (whatsappContact.length === 0) {
Expand Down
10 changes: 10 additions & 0 deletions src/api/routes/label.router.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@ export class LabelRouter extends RouterBroker {

return res.status(HttpStatus.OK).json(response);
})
.get(this.routerPath('syncLabels'), ...guards, async (req, res) => {
const response = await this.dataValidate<LabelDto>({
request: req,
schema: null,
ClassRef: LabelDto,
execute: (instance) => labelController.syncLabels(instance),
});

return res.status(HttpStatus.OK).json(response);
})
.post(this.routerPath('handleLabel'), ...guards, async (req, res) => {
const response = await this.dataValidate<HandleLabelDto>({
request: req,
Expand Down
148 changes: 148 additions & 0 deletions tests/label-sync-snapshot.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
import assert from 'node:assert/strict';
import test from 'node:test';

import { BaileysStartupService } from '../src/api/integrations/channel/whatsapp/whatsapp.baileys.service';

type SnapshotCollector = {
observedLabelState: boolean;
labelsByChatId: Map<string, Set<string>>;
};

type TestableBaileysStartupService = {
labelAssociationSnapshotCollector?: SnapshotCollector;
eventProcessingQueue: Promise<void>;
syncLabels(): Promise<unknown>;
};

function immediateTimeouts() {
const original = globalThis.setTimeout;
globalThis.setTimeout = ((callback: (...args: unknown[]) => void, _delay?: number, ...args: unknown[]) => {
callback(...args);
return 0 as unknown as NodeJS.Timeout;
}) as typeof setTimeout;
return () => {
globalThis.setTimeout = original;
};
}

test('syncLabels replaces stale chat labels with the full app-state snapshot', async () => {
const updateManyCalls: unknown[] = [];
const upsertCalls: unknown[] = [];
const keyWrites: unknown[] = [];
const transaction = {
chat: {
updateMany: async (input: unknown) => updateManyCalls.push(input),
upsert: async (input: unknown) => upsertCalls.push(input),
},
};
const service = Object.create(BaileysStartupService.prototype) as TestableBaileysStartupService;
Object.assign(service, {
instance: {
id: 'instance-1',
name: 'test-instance',
authState: {
state: {
keys: {
set: async (input: unknown) => keyWrites.push(input),
},
},
},
},
eventProcessingQueue: Promise.resolve(),
logger: { warn: () => undefined },
prismaRepository: {
label: { findMany: async () => [] },
$transaction: async (callback: (client: typeof transaction) => Promise<void>) => callback(transaction),
},
client: {
resyncAppState: async () => {
const collector = service.labelAssociationSnapshotCollector;
assert.ok(collector, 'syncLabels must collect the forced app-state snapshot');
collector.observedLabelState = true;
collector.labelsByChatId.set('chat-1@s.whatsapp.net', new Set(['label-2', 'label-1']));
},
},
});
const restoreTimeouts = immediateTimeouts();

try {
await service.syncLabels();
} finally {
restoreTimeouts();
}

assert.deepEqual(keyWrites, [{ 'app-state-sync-version': { regular: null } }]);
assert.deepEqual(updateManyCalls, [
{
where: { instanceId: 'instance-1' },
data: { labels: [] },
},
]);
assert.equal(upsertCalls.length, 1);
const [upsert] = upsertCalls as Array<{
where: unknown;
update: unknown;
create: { id: string; instanceId: string; remoteJid: string; labels: string[] };
}>;
assert.match(upsert.create.id, /^[a-z0-9]+$/);
assert.deepEqual(
{
...upsert,
create: { ...upsert.create, id: '<generated>' },
},
{
where: {
instanceId_remoteJid: {
instanceId: 'instance-1',
remoteJid: 'chat-1@s.whatsapp.net',
},
},
update: { labels: ['label-1', 'label-2'] },
create: {
id: '<generated>',
instanceId: 'instance-1',
remoteJid: 'chat-1@s.whatsapp.net',
labels: ['label-1', 'label-2'],
},
},
);
assert.equal(service.labelAssociationSnapshotCollector, undefined);
});

test('syncLabels keeps existing chat labels when the forced sync returns no label state', async () => {
let transactionCalls = 0;
let warnings = 0;
const service = Object.create(BaileysStartupService.prototype) as TestableBaileysStartupService;
Object.assign(service, {
instance: {
id: 'instance-1',
name: 'test-instance',
authState: {
state: {
keys: {
set: async () => undefined,
},
},
},
},
eventProcessingQueue: Promise.resolve(),
logger: { warn: () => warnings++ },
prismaRepository: {
label: { findMany: async () => [] },
$transaction: async () => transactionCalls++,
},
client: {
resyncAppState: async () => undefined,
},
});
const restoreTimeouts = immediateTimeouts();

try {
await service.syncLabels();
} finally {
restoreTimeouts();
}

assert.equal(transactionCalls, 0);
assert.equal(warnings, 1);
});
5 changes: 3 additions & 2 deletions tsconfig.json
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
"exclude": ["node_modules", "./test", "./dist", "./prisma"],
"include": [
"src/**/*",
"src/**/*.json"
"src/**/*.json",
"tests/**/*.ts"
]
}
}