feat(worker): seal WhatsApp adapter — normalize inside session, reactions handled internally

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/worker/src/whatsapp/normalizer.test.ts

@@ -1,5 +1,5 @@
 import { proto } from '@whiskeysockets/baileys';
-import { normalizeMessage } from './normalizer';
+import { normalizeMessage, normalizeReaction } from './normalizer';
 
 function makeMsg(overrides: Partial<proto.IWebMessageInfo> = {}): proto.IWebMessageInfo {
   return {
@@ -17,61 +17,127 @@ function makeMsg(overrides: Partial<proto.IWebMessageInfo> = {}): proto.IWebMess
 
 describe('normalizeMessage', () => {
   it('normalizes a plain text conversation message', () => {
-    const result = normalizeMessage(makeMsg());
+    const result = normalizeMessage(makeMsg(), 'acc_1');
     expect(result).toMatchObject({
       platformMsgId: 'ABCDEF123456',
       sourceGroupJid: '120363043312345678@g.us',
       senderJid: '919876543210@s.whatsapp.net',
       senderName: 'Alice',
       content: 'Hello world',
+      accountId: 'acc_1',
     });
   });
 
   it('normalizes an extendedTextMessage', () => {
     const result = normalizeMessage(makeMsg({
       message: { extendedTextMessage: { text: 'Extended text here' } },
-    }));
+    }), 'acc_1');
     expect(result?.content).toBe('Extended text here');
   });
 
   it('normalizes an imageMessage caption', () => {
     const result = normalizeMessage(makeMsg({
       message: { imageMessage: { caption: 'Photo caption' } },
-    }));
+    }), 'acc_1');
     expect(result?.content).toBe('Photo caption');
   });
 
   it('normalizes a videoMessage caption', () => {
     const result = normalizeMessage(makeMsg({
       message: { videoMessage: { caption: 'Video caption' } },
-    }));
+    }), 'acc_1');
     expect(result?.content).toBe('Video caption');
   });
 
   it('returns null for own messages (fromMe = true)', () => {
-    const result = normalizeMessage(makeMsg({ key: { remoteJid: '120363043312345678@g.us', fromMe: true, id: 'XYZ', participant: '919876543210@s.whatsapp.net' } }));
+    const result = normalizeMessage(makeMsg({ key: { remoteJid: '120363043312345678@g.us', fromMe: true, id: 'XYZ', participant: '919876543210@s.whatsapp.net' } }), 'acc_1');
     expect(result).toBeNull();
   });
 
   it('returns null for non-group messages (DMs)', () => {
     const result = normalizeMessage(makeMsg({
       key: { remoteJid: '919876543210@s.whatsapp.net', fromMe: false, id: 'XYZ' },
-    }));
+    }), 'acc_1');
     expect(result).toBeNull();
   });
 
   it('returns null when message payload is missing', () => {
-    const result = normalizeMessage(makeMsg({ message: null }));
+    const result = normalizeMessage(makeMsg({ message: null }), 'acc_1');
     expect(result).toBeNull();
   });
 
   it('returns null when key is missing', () => {
-    const result = normalizeMessage({ message: { conversation: 'hi' } } as proto.IWebMessageInfo);
+    const result = normalizeMessage({ message: { conversation: 'hi' } } as proto.IWebMessageInfo, 'acc_1');
     expect(result).toBeNull();
   });
 
   it('returns null when message id is missing (prevents empty-key upsert collisions)', () => {
-    const result = normalizeMessage(makeMsg({ key: { remoteJid: '120363043312345678@g.us', fromMe: false, id: undefined, participant: '919876543210@s.whatsapp.net' } }));
+    const result = normalizeMessage(makeMsg({ key: { remoteJid: '120363043312345678@g.us', fromMe: false, id: undefined, participant: '919876543210@s.whatsapp.net' } }), 'acc_1');
     expect(result).toBeNull();
   });
 });
+
+describe('normalizeReaction', () => {
+  it('normalizes a star reaction from a group participant', () => {
+    const msg = {
+      key: {
+        remoteJid: '120363043312345678@g.us',
+        fromMe: false,
+        id: 'REACTION_ID',
+        participant: '919876543210@s.whatsapp.net',
+      },
+      message: {
+        reactionMessage: {
+          key: { remoteJid: '120363043312345678@g.us', id: 'TARGET_MSG_ID' },
+          text: '⭐',
+        },
+      },
+    } as proto.IWebMessageInfo;
+
+    const result = normalizeReaction(msg, 'acc_1');
+    expect(result).toMatchObject({
+      reactorJid: '919876543210@s.whatsapp.net',
+      targetMsgId: 'TARGET_MSG_ID',
+      sourceGroupJid: '120363043312345678@g.us',
+      emoji: '⭐',
+      accountId: 'acc_1',
+    });
+  });
+
+  it('returns null when reactionMessage is missing (regular message)', () => {
+    const msg = {
+      key: { remoteJid: '120363043312345678@g.us', fromMe: false, id: 'ID' },
+      message: { conversation: 'hello' },
+    } as proto.IWebMessageInfo;
+    expect(normalizeReaction(msg, 'acc_1')).toBeNull();
+  });
+
+  it('returns null for own reactions (fromMe=true)', () => {
+    const msg = {
+      key: { remoteJid: '120363043312345678@g.us', fromMe: true, id: 'ID' },
+      message: { reactionMessage: { key: { id: 'TARGET' }, text: '⭐' } },
+    } as proto.IWebMessageInfo;
+    expect(normalizeReaction(msg, 'acc_1')).toBeNull();
+  });
+
+  it('returns null for DM reactions (non-group jid)', () => {
+    const msg = {
+      key: { remoteJid: '919876543210@s.whatsapp.net', fromMe: false, id: 'ID' },
+      message: { reactionMessage: { key: { id: 'TARGET' }, text: '⭐' } },
+    } as proto.IWebMessageInfo;
+    expect(normalizeReaction(msg, 'acc_1')).toBeNull();
+  });
+
+  it('returns null when targetMsgId is missing', () => {
+    const msg = {
+      key: {
+        remoteJid: '120363043312345678@g.us',
+        fromMe: false,
+        id: 'ID',
+        participant: '91@s.whatsapp.net',
+      },
+      message: { reactionMessage: { key: {}, text: '⭐' } },
+    } as proto.IWebMessageInfo;
+    expect(normalizeReaction(msg, 'acc_1')).toBeNull();
+  });
+});

apps/worker/src/whatsapp/normalizer.ts

@@ -1,12 +1,5 @@
 import { proto } from '@whiskeysockets/baileys';
-
+import { NormalizedMessage, NormalizedReaction } from '@tower/types';
-export interface NormalizedMessage {
-  platformMsgId: string;
-  sourceGroupJid: string;
-  senderJid: string;
-  senderName?: string;
-  content: string;
-}
 
 function extractText(msg: proto.IWebMessageInfo): string {
   const m = msg.message;
@@ -21,18 +14,16 @@ function extractText(msg: proto.IWebMessageInfo): string {
   );
 }
 
-export function normalizeMessage(msg: proto.IWebMessageInfo): NormalizedMessage | null {
+export function normalizeMessage(
+  msg: proto.IWebMessageInfo,
+  accountId: string,
+): NormalizedMessage | null {
   const key = msg.key;
   if (!key) return null;
 
   const remoteJid = key.remoteJid ?? '';
-
-  // Only process group messages (group JIDs end with @g.us)
   if (!remoteJid.endsWith('@g.us')) return null;
-
-  // Skip our own outgoing messages
   if (key.fromMe) return null;
-
   if (!msg.message) return null;
 
   const platformMsgId = key.id;
@@ -46,5 +37,32 @@ export function normalizeMessage(msg: proto.IWebMessageInfo): NormalizedMessage
     senderJid: key.participant ?? '',
     senderName: msg.pushName ?? undefined,
     content,
+    accountId,
+  };
+}
+
+export function normalizeReaction(
+  msg: proto.IWebMessageInfo,
+  accountId: string,
+): NormalizedReaction | null {
+  const key = msg.key;
+  if (!key) return null;
+
+  const remoteJid = key.remoteJid ?? '';
+  if (!remoteJid.endsWith('@g.us')) return null;
+  if (key.fromMe) return null;
+
+  const reaction = msg.message?.reactionMessage;
+  if (!reaction) return null;
+
+  const targetMsgId = reaction.key?.id;
+  if (!targetMsgId) return null;
+
+  return {
+    reactorJid: key.participant ?? '',
+    targetMsgId,
+    sourceGroupJid: remoteJid,
+    emoji: reaction.text ?? '',
+    accountId,
   };
 }

apps/worker/src/whatsapp/session.ts

@@ -3,20 +3,25 @@ import makeWASocket, {
   fetchLatestBaileysVersion,
   DisconnectReason,
   WASocket,
-  proto,
   GroupMetadata,
 } from '@whiskeysockets/baileys';
 import { Boom } from '@hapi/boom';
+import qrcode from 'qrcode-terminal';
+import { NormalizedMessage, NormalizedReaction } from '@tower/types';
+import { normalizeMessage, normalizeReaction } from './normalizer';
 import { createLogger } from '@tower/logger';
 
 const logger = createLogger('whatsapp-session');
 
-export type OnMessageCallback = (msg: proto.IWebMessageInfo) => void;
+export type OnMessageCallback = (msg: NormalizedMessage) => Promise<void> | void;
-export type OnGroupsCallback = (groups: Record<string, GroupMetadata>) => void;
+export type OnReactionCallback = (reaction: NormalizedReaction) => Promise<void> | void;
+export type OnGroupsCallback = (groups: Record<string, GroupMetadata>) => Promise<void> | void;
 
 export async function createWhatsAppSession(
+  accountId: string,
   sessionPath: string,
   onMessage: OnMessageCallback,
+  onReaction: OnReactionCallback,
   onGroups: OnGroupsCallback,
 ): Promise<WASocket> {
   const { state, saveCreds } = await useMultiFileAuthState(sessionPath);
@@ -25,24 +30,30 @@ export async function createWhatsAppSession(
   const sock = makeWASocket({
     version,
     auth: state,
-    printQRInTerminal: true,
+    printQRInTerminal: false,
     logger: logger as any,
   });
 
   sock.ev.on('creds.update', saveCreds);
 
-  sock.ev.on('connection.update', async ({ connection, lastDisconnect }) => {
+  sock.ev.on('connection.update', async ({ connection, qr, lastDisconnect }) => {
+    if (qr) {
+      qrcode.generate(qr, { small: true });
+    }
     if (connection === 'close') {
       const reason = (lastDisconnect?.error as Boom)?.output?.statusCode;
       const shouldReconnect = reason !== DisconnectReason.loggedOut;
       logger.info({ reason, shouldReconnect }, 'Connection closed');
       if (shouldReconnect) {
         logger.info('Reconnecting in 5s...');
-        setTimeout(() => createWhatsAppSession(sessionPath, onMessage, onGroups), 5000);
+        setTimeout(
+          () => createWhatsAppSession(accountId, sessionPath, onMessage, onReaction, onGroups),
+          5000,
+        );
       }
     } else if (connection === 'open') {
       try {
-        logger.info('WhatsApp connected');
+        logger.info({ accountId }, 'WhatsApp connected');
         const groups = await sock.groupFetchAllParticipating();
         await Promise.resolve(onGroups(groups)).catch((err) =>
           logger.error({ err }, 'Group sync error'),
@@ -56,7 +67,18 @@ export async function createWhatsAppSession(
   sock.ev.on('messages.upsert', ({ messages, type }) => {
     if (type !== 'notify') return;
     for (const msg of messages) {
-      void Promise.resolve(onMessage(msg)).catch((err) =>
+      if (msg.message?.reactionMessage) {
+        const reaction = normalizeReaction(msg, accountId);
+        if (reaction) {
+          void Promise.resolve(onReaction(reaction)).catch((err) =>
+            logger.error({ err }, 'Error processing reaction'),
+          );
+        }
+        continue;
+      }
+      const normalized = normalizeMessage(msg, accountId);
+      if (!normalized) continue;
+      void Promise.resolve(onMessage(normalized)).catch((err) =>
         logger.error({ err }, 'Error processing message'),
       );
     }