feat: implement WAMonitoringService for instance lifecycle management and add WhatsApp Baileys integration support - #2671
Open
codedalex wants to merge 1 commit into
Open
Conversation
… and add WhatsApp Baileys integration support
Contributor
Reviewer's GuideImplements additional robustness and deduplication in the WhatsApp Baileys integration, improves instance lifecycle handling in WAMonitoringService to prevent zombie instances, and updates docker-compose networking/port configuration. Sequence diagram for WhatsApp Baileys message handling with pushName fallback and deduplicationsequenceDiagram
participant BaileysStartupService
participant PrismaRepository
participant MessageTable as Message
participant ContactTable as Contact
BaileysStartupService->>BaileysStartupService: receive message
alt missing pushName and not fromMe
BaileysStartupService->>PrismaRepository: contact.findFirst(instanceId, participantJid)
PrismaRepository-->>BaileysStartupService: contact.pushName
BaileysStartupService->>BaileysStartupService: set received.pushName
end
BaileysStartupService->>BaileysStartupService: prepareMessage(received)
BaileysStartupService->>PrismaRepository: $queryRaw SELECT FROM Message WHERE instanceId, key.id
PrismaRepository-->>BaileysStartupService: existingMessage[]
alt existingMessage found
BaileysStartupService->>MessageTable: reuse existingMessage[0]
else no existingMessage
BaileysStartupService->>PrismaRepository: message.create(messageData)
PrismaRepository-->>BaileysStartupService: msg
end
BaileysStartupService->>BaileysStartupService: use msg.messageTimestamp
Sequence diagram for WAMonitoringService setInstance zombie instance cleanupsequenceDiagram
participant WAMonitoringService
participant waInstances
participant Client
participant WebSocket as client.ws
participant ChannelController as channelController
WAMonitoringService->>waInstances: lookup instanceData.instanceName
alt existing instance found
WAMonitoringService->>Client: existing.client
opt client.ws exists
WAMonitoringService->>WebSocket: close()
end
opt client.end exists
WAMonitoringService->>Client: end(undefined)
end
WAMonitoringService->>waInstances: delete existing instance
end
WAMonitoringService->>ChannelController: init(instanceData, deps)
ChannelController-->>WAMonitoringService: new instance
WAMonitoringService->>waInstances: store new instance
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In the duplicate-message handling block,
existingMessageis fetched via$queryRawwith onlyid,status, andmessageTimestamp, but the result is then used asmsgjust like the Prisma-created entity; if downstream code relies on otherMessagefields or expects a Prisma type, consider normalizing this (e.g., usingfindFirstwith the same selection or at least clearly typing the result). - The
contactRaw.pushNamefield has been made optional and is now only set when!received.key.fromMe && received.pushName; if the underlying schema or any consumer code assumes a non-nullpushName, consider ensuring a consistent fallback (e.g., empty string) to avoid unexpectedundefinedvalues. - In
WAMonitoringService.setInstance, the forceful shutdown of existing instances (ws.close()andclient.end()) before re-init may introduce subtle race conditions ifsetInstanceis invoked concurrently; it could be safer to guard this path (e.g., with a per-instance lock or a state flag) to prevent overlapping close/re-init cycles.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In the duplicate-message handling block, `existingMessage` is fetched via `$queryRaw` with only `id`, `status`, and `messageTimestamp`, but the result is then used as `msg` just like the Prisma-created entity; if downstream code relies on other `Message` fields or expects a Prisma type, consider normalizing this (e.g., using `findFirst` with the same selection or at least clearly typing the result).
- The `contactRaw.pushName` field has been made optional and is now only set when `!received.key.fromMe && received.pushName`; if the underlying schema or any consumer code assumes a non-null `pushName`, consider ensuring a consistent fallback (e.g., empty string) to avoid unexpected `undefined` values.
- In `WAMonitoringService.setInstance`, the forceful shutdown of existing instances (`ws.close()` and `client.end()`) before re-init may introduce subtle race conditions if `setInstance` is invoked concurrently; it could be safer to guard this path (e.g., with a per-instance lock or a state flag) to prevent overlapping close/re-init cycles.
## Individual Comments
### Comment 1
<location path="src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts" line_range="1205-1212" />
<code_context>
}
+ // FIX: Fallback pushName if not present in received payload
+ if (!received.pushName && !received.key.fromMe) {
+ const participantJid = received.participant || received.key.participant || received.key.remoteJid;
+ if (participantJid) {
+ const contact = await this.prismaRepository.contact.findFirst({
+ where: { instanceId: this.instanceId, remoteJid: participantJid },
+ select: { pushName: true }
+ });
+ if (contact && contact.pushName) {
+ received.pushName = contact.pushName;
+ }
</code_context>
<issue_to_address>
**suggestion (performance):** Avoid per-message lookup for pushName where possible to reduce database load.
This path triggers a `contact.findFirst` for every message without `pushName`, which can become an N+1 query pattern under high throughput. To mitigate this, you could restrict the fallback (e.g. only for group messages where `participant` is set) and/or add a short-lived in-memory cache keyed by `participantJid` and `instanceId` so repeated messages from the same JID don’t repeatedly hit the DB. Alternatively, consider batching contact lookups earlier in the pipeline if you can determine the relevant JIDs in advance.
Suggested implementation:
```typescript
// FIX: Fallback pushName if not present in received payload.
// Restrict to group messages (participant present) and use a small in-memory cache
// to avoid per-message DB lookups for the same JID.
if (
!received.pushName &&
!received.key.fromMe &&
(received.participant || received.key?.participant)
) {
const participantJid =
received.participant || received.key.participant;
if (participantJid) {
const cacheKey = `${this.instanceId}:${participantJid}`;
// In-memory cache for pushName lookups, keyed by instance + JID
let cachedPushName =
this.pushNameCache && this.pushNameCache.get(cacheKey);
if (!cachedPushName) {
const contact = await this.prismaRepository.contact.findFirst({
where: {
instanceId: this.instanceId,
remoteJid: participantJid,
},
select: { pushName: true },
});
if (contact?.pushName) {
cachedPushName = contact.pushName;
if (this.pushNameCache) {
this.pushNameCache.set(cacheKey, cachedPushName);
}
}
}
if (cachedPushName) {
received.pushName = cachedPushName;
}
}
```
1. Add an in-memory cache field to the WhatsApp Baileys service class, for example:
`private readonly pushNameCache = new Map<string, string>();`
Place this alongside other private fields on the service.
2. If you want cache eviction, you can later wrap this Map with a TTL mechanism or use an LRU cache implementation already present in your codebase (if any).
</issue_to_address>
### Comment 2
<location path="src/api/integrations/channel/whatsapp/whatsapp.baileys.service.ts" line_range="1373-1380" />
<code_context>
const { pollUpdates, ...messageData } = messageRaw;
- const msg = await this.prismaRepository.message.create({ data: messageData });
+
+ const existingMessage = await this.prismaRepository.$queryRaw`
+ SELECT id, status, "messageTimestamp" FROM "Message"
+ WHERE "instanceId" = ${this.instanceId}
+ AND "key"->>'id' = ${received.key.id}
+ ` as any[];
+
+ let msg;
+ if (existingMessage && existingMessage.length > 0) {
+ msg = existingMessage[0];
+ this.logger.info(`Message already exists, ignoring create: ${received.key.id}`);
</code_context>
<issue_to_address>
**issue (bug_risk):** Use a uniqueness constraint or upsert instead of a raw pre-check query to prevent duplicates.
Because the `SELECT` + conditional `create` flow is not atomic, concurrent deliveries of the same message ID can still result in duplicate rows. It would be safer to enforce a unique index on `(instanceId, key->>'id')` and rely on the DB to handle conflicts (e.g., via an upsert or `create` with `ON CONFLICT DO NOTHING` using `$executeRaw`).
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+1205
to
+1212
| if (!received.pushName && !received.key.fromMe) { | ||
| const participantJid = received.participant || received.key.participant || received.key.remoteJid; | ||
| if (participantJid) { | ||
| const contact = await this.prismaRepository.contact.findFirst({ | ||
| where: { instanceId: this.instanceId, remoteJid: participantJid }, | ||
| select: { pushName: true } | ||
| }); | ||
| if (contact && contact.pushName) { |
Contributor
There was a problem hiding this comment.
suggestion (performance): Avoid per-message lookup for pushName where possible to reduce database load.
This path triggers a contact.findFirst for every message without pushName, which can become an N+1 query pattern under high throughput. To mitigate this, you could restrict the fallback (e.g. only for group messages where participant is set) and/or add a short-lived in-memory cache keyed by participantJid and instanceId so repeated messages from the same JID don’t repeatedly hit the DB. Alternatively, consider batching contact lookups earlier in the pipeline if you can determine the relevant JIDs in advance.
Suggested implementation:
// FIX: Fallback pushName if not present in received payload.
// Restrict to group messages (participant present) and use a small in-memory cache
// to avoid per-message DB lookups for the same JID.
if (
!received.pushName &&
!received.key.fromMe &&
(received.participant || received.key?.participant)
) {
const participantJid =
received.participant || received.key.participant;
if (participantJid) {
const cacheKey = `${this.instanceId}:${participantJid}`;
// In-memory cache for pushName lookups, keyed by instance + JID
let cachedPushName =
this.pushNameCache && this.pushNameCache.get(cacheKey);
if (!cachedPushName) {
const contact = await this.prismaRepository.contact.findFirst({
where: {
instanceId: this.instanceId,
remoteJid: participantJid,
},
select: { pushName: true },
});
if (contact?.pushName) {
cachedPushName = contact.pushName;
if (this.pushNameCache) {
this.pushNameCache.set(cacheKey, cachedPushName);
}
}
}
if (cachedPushName) {
received.pushName = cachedPushName;
}
}- Add an in-memory cache field to the WhatsApp Baileys service class, for example:
private readonly pushNameCache = new Map<string, string>();
Place this alongside other private fields on the service. - If you want cache eviction, you can later wrap this Map with a TTL mechanism or use an LRU cache implementation already present in your codebase (if any).
Comment on lines
+1373
to
+1380
| const existingMessage = await this.prismaRepository.$queryRaw` | ||
| SELECT id, status, "messageTimestamp" FROM "Message" | ||
| WHERE "instanceId" = ${this.instanceId} | ||
| AND "key"->>'id' = ${received.key.id} | ||
| ` as any[]; | ||
|
|
||
| let msg; | ||
| if (existingMessage && existingMessage.length > 0) { |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Use a uniqueness constraint or upsert instead of a raw pre-check query to prevent duplicates.
Because the SELECT + conditional create flow is not atomic, concurrent deliveries of the same message ID can still result in duplicate rows. It would be safer to enforce a unique index on (instanceId, key->>'id') and rely on the DB to handle conflicts (e.g., via an upsert or create with ON CONFLICT DO NOTHING using $executeRaw).
|
@codedalex send to develop, so we can review that PR (we dont accept anything to main) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Improve WhatsApp Baileys instance lifecycle handling and message/contact persistence, and adjust local docker networking defaults.
Bug Fixes:
Enhancements:
Deployment: