diff --git a/Example/example.ts b/Example/example.ts index 1e99f09..afba810 100644 --- a/Example/example.ts +++ b/Example/example.ts @@ -98,21 +98,10 @@ const startSock = async() => { console.log('recv call event', events.call) } - // chat history received - if(events['chats.set']) { - const { chats, isLatest } = events['chats.set'] - console.log(`recv ${chats.length} chats (is latest: ${isLatest})`) - } - - // message history received - if(events['messages.set']) { - const { messages, isLatest } = events['messages.set'] - console.log(`recv ${messages.length} messages (is latest: ${isLatest})`) - } - - if(events['contacts.set']) { - const { contacts, isLatest } = events['contacts.set'] - console.log(`recv ${contacts.length} contacts (is latest: ${isLatest})`) + // history received + if(events['messaging-history.set']) { + const { chats, contacts, messages, isLatest } = events['messaging-history.set'] + console.log(`recv ${chats.length} chats, ${contacts.length} contacts, ${messages.length} msgs (is latest: ${isLatest})`) } // received a new message diff --git a/README.md b/README.md index 17f2865..49939e2 100644 --- a/README.md +++ b/README.md @@ -91,10 +91,40 @@ You can configure the connection by passing a `SocketConfig` object. The entire `SocketConfig` structure is mentioned here with default values: ``` ts type SocketConfig = { + /** the WS url to connect to WA */ + waWebSocketUrl: string | URL + /** Fails the connection if the socket times out in this interval */ + connectTimeoutMs: number + /** Default timeout for queries, undefined for no timeout */ + defaultQueryTimeoutMs: number | undefined + /** ping-pong interval for WS connection */ + keepAliveIntervalMs: number + /** proxy agent */ + agent?: Agent + /** pino logger */ + logger: Logger + /** version to connect with */ + version: WAVersion + /** override browser config */ + browser: WABrowserDescription + /** agent used for fetch requests -- uploading/downloading media */ + fetchAgent?: Agent + /** should the QR be printed in the terminal */ + printQRInTerminal: boolean + /** should events be emitted for actions done by this socket connection */ + emitOwnEvents: boolean + /** provide a cache to store media, so does not have to be re-uploaded */ + mediaCache?: NodeCache + /** custom upload hosts to upload media to */ + customUploadHosts: MediaConnInfo['hosts'] + /** time to wait between sending new retry requests */ + retryRequestDelayMs: number + /** time to wait for the generation of the next QR in ms */ + qrTimeout?: number; /** provide an auth state object to maintain the auth state */ auth: AuthenticationState - /** By default true, should history messages be downloaded and processed */ - downloadHistory: boolean + /** manage history processing with this control; by default will sync up everything */ + shouldSyncHistoryMessage: (msg: proto.Message.IHistorySyncNotification) => boolean /** transaction capability options for SignalKeyStore */ transactionOpts: TransactionCapabilityOptions /** provide a cache to store a user's device list */ @@ -107,13 +137,18 @@ type SocketConfig = { msgRetryCounterMap?: MessageRetryMap /** width for link preview images */ linkPreviewImageThumbnailWidth: number + /** Should Baileys ask the phone for full history, will be received async */ + syncFullHistory: boolean + /** Should baileys fire init queries automatically, default true */ + fireInitQueries: boolean /** * generate a high quality link preview, * entails uploading the jpegThumbnail to WA * */ generateHighQualityLinkPreview: boolean - /** Should Baileys ask the phone for full history, will be received async */ - syncFullHistory: boolean + + /** options for axios */ + options: AxiosRequestConfig /** * fetch a message from your store * implement this so that messages failed to send (solves the "this message can take a while" issue) can be retried @@ -184,21 +219,22 @@ type ConnectionState = { Baileys uses the EventEmitter syntax for events. They're all nicely typed up, so you shouldn't have any issues with an Intellisense editor like VS Code. -The events are typed up in a type map, as mentioned here: +The events are typed as mentioned here: ``` ts -export type BaileysEventMap = { +export type BaileysEventMap = { /** connection state has been updated -- WS closed, opened, connecting etc. */ 'connection.update': Partial /** credentials updated -- some metadata, keys or something */ - 'creds.update': Partial - /** set chats (history sync), chats are reverse chronologically sorted */ - 'chats.set': { chats: Chat[], isLatest: boolean } - /** set messages (history sync), messages are reverse chronologically sorted */ - 'messages.set': { messages: WAMessage[], isLatest: boolean } - /** set contacts (history sync) */ - 'contacts.set': { contacts: Contact[], isLatest: boolean } + 'creds.update': Partial + /** history sync, everything is reverse chronologically sorted */ + 'messaging-history.set': { + chats: Chat[] + contacts: Contact[] + messages: WAMessage[] + isLatest: boolean + } /** upsert chats */ 'chats.upsert': Chat[] /** update the given chats */ diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index d316715..b5581f2 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -1,3 +1,4 @@ +import { proto } from '../../WAProto' import type { MediaType, SocketConfig } from '../Types' import { Browsers } from '../Utils' import logger from '../Utils/logger' @@ -26,6 +27,13 @@ export const WA_CERT_DETAILS = { SERIAL: 0, } +export const PROCESSABLE_HISTORY_TYPES = [ + proto.Message.HistorySyncNotification.HistorySyncType.INITIAL_BOOTSTRAP, + proto.Message.HistorySyncNotification.HistorySyncType.PUSH_NAME, + proto.Message.HistorySyncNotification.HistorySyncType.RECENT, + proto.Message.HistorySyncNotification.HistorySyncType.FULL +] + export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { version: version as any, browser: Browsers.baileys('Chrome'), @@ -40,9 +48,9 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { retryRequestDelayMs: 250, fireInitQueries: true, auth: undefined as any, - downloadHistory: true, markOnlineOnConnect: true, syncFullHistory: false, + shouldSyncHistoryMessage: () => true, linkPreviewImageThumbnailWidth: 192, transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 3000 }, generateHighQualityLinkPreview: false, diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index 4acc1e8..4a53f8f 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -1,7 +1,8 @@ import { Boom } from '@hapi/boom' import { proto } from '../../WAProto' -import { ALL_WA_PATCH_NAMES, ChatModification, ChatMutation, InitialReceivedChatsState, LTHashState, MessageUpsertType, PresenceData, SocketConfig, WABusinessHoursConfig, WABusinessProfile, WAMediaUpload, WAMessage, WAPatchCreate, WAPatchName, WAPresence } from '../Types' -import { chatModificationToAppPatch, debouncedTimeout, decodePatches, decodeSyncdSnapshot, encodeSyncdPatch, extractSyncdPatches, generateProfilePicture, isHistoryMsg, newLTHashState, processSyncAction } from '../Utils' +import { PROCESSABLE_HISTORY_TYPES } from '../Defaults' +import { ALL_WA_PATCH_NAMES, ChatModification, ChatMutation, LTHashState, MessageUpsertType, PresenceData, SocketConfig, WABusinessHoursConfig, WABusinessProfile, WAMediaUpload, WAMessage, WAPatchCreate, WAPatchName, WAPresence } from '../Types' +import { chatModificationToAppPatch, decodePatches, decodeSyncdSnapshot, encodeSyncdPatch, extractSyncdPatches, generateProfilePicture, getHistoryMsg, newLTHashState, processSyncAction } from '../Utils' import { makeMutex } from '../Utils/make-mutex' import processMessage from '../Utils/process-message' import { BinaryNode, getBinaryNodeChild, getBinaryNodeChildren, jidNormalizedUser, reduceBinaryNodeToDictionary, S_WHATSAPP_NET } from '../WABinary' @@ -9,10 +10,8 @@ import { makeSocket } from './socket' const MAX_SYNC_ATTEMPTS = 5 -const APP_STATE_SYNC_TIMEOUT_MS = 10_000 - export const makeChatsSocket = (config: SocketConfig) => { - const { logger, markOnlineOnConnect, downloadHistory, fireInitQueries } = config + const { logger, markOnlineOnConnect, shouldSyncHistoryMessage, fireInitQueries } = config const sock = makeSocket(config) const { ev, @@ -26,40 +25,8 @@ export const makeChatsSocket = (config: SocketConfig) => { } = sock let privacySettings: { [_: string]: string } | undefined - /** this mutex ensures that the notifications (receipts, messages etc.) are processed in order */ const processingMutex = makeMutex() - /** cache to ensure new history sync events do not have duplicate items */ - const historyCache = new Set() - let recvChats: InitialReceivedChatsState = { } - - const appStateSyncTimeout = debouncedTimeout( - APP_STATE_SYNC_TIMEOUT_MS, - async() => { - if(!authState.creds.myAppStateKeyId) { - logger.warn('myAppStateKeyId not synced, bad link') - await logout('Incomplete app state key sync') - return - } - - if(ws.readyState === ws.OPEN) { - - logger.info( - { recvChats: Object.keys(recvChats).length }, - 'doing initial app state sync' - ) - await resyncMainAppState(recvChats) - - const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 - ev.emit('creds.update', { accountSyncCounter }) - } else { - logger.warn('connection closed before app state sync') - } - - historyCache.clear() - recvChats = { } - } - ) /** helper function to fetch the given app state sync key */ const getAppStateSyncKey = async(keyId: string) => { @@ -310,22 +277,22 @@ export const makeChatsSocket = (config: SocketConfig) => { }) } - const newAppStateChunkHandler = (recvChats: InitialReceivedChatsState | undefined) => { + const newAppStateChunkHandler = (isInitialSync: boolean) => { return { onMutation(mutation: ChatMutation) { processSyncAction( mutation, ev, authState.creds.me!, - recvChats ? { recvChats, accountSettings: authState.creds.accountSettings } : undefined, + isInitialSync ? { accountSettings: authState.creds.accountSettings } : undefined, logger ) } } } - const resyncAppState = ev.createBufferedFunction(async(collections: readonly WAPatchName[], recvChats: InitialReceivedChatsState | undefined) => { - const { onMutation } = newAppStateChunkHandler(recvChats) + const resyncAppState = ev.createBufferedFunction(async(collections: readonly WAPatchName[], isInitialSync: boolean) => { + const { onMutation } = newAppStateChunkHandler(isInitialSync) // we use this to determine which events to fire // otherwise when we resync from scratch -- all notifications will fire const initialVersionMap: { [T in WAPatchName]?: number } = { } @@ -543,19 +510,6 @@ export const makeChatsSocket = (config: SocketConfig) => { } } - const resyncMainAppState = async(ctx?: InitialReceivedChatsState) => { - logger.debug('resyncing main app state') - - await ( - processingMutex.mutex( - () => resyncAppState(ALL_WA_PATCH_NAMES, ctx) - ) - .catch(err => ( - onUnexpectedError(err, 'main app sync') - )) - ) - } - const appPatch = async(patchCreate: WAPatchCreate) => { const name = patchCreate.type const myAppStateKeyId = authState.creds.myAppStateKeyId @@ -572,7 +526,7 @@ export const makeChatsSocket = (config: SocketConfig) => { async() => { logger.debug({ patch: patchCreate }, 'applying app patch') - await resyncAppState([name], undefined) + await resyncAppState([name], false) const { [name]: currentSyncVersion } = await authState.keys.get('app-state-sync-version', [name]) initial = currentSyncVersion || newLTHashState() @@ -625,7 +579,7 @@ export const makeChatsSocket = (config: SocketConfig) => { ) if(config.emitOwnEvents) { - const { onMutation } = newAppStateChunkHandler(undefined) + const { onMutation } = newAppStateChunkHandler(false) await decodePatches( name, [{ ...encodeResult!.patch, version: { version: encodeResult!.state.version }, }], @@ -726,33 +680,49 @@ export const makeChatsSocket = (config: SocketConfig) => { } // update our pushname too - if(msg.key.fromMe && authState.creds.me?.name !== msg.pushName) { + if(msg.key.fromMe && msg.pushName && authState.creds.me?.name !== msg.pushName) { ev.emit('creds.update', { me: { ...authState.creds.me!, name: msg.pushName! } }) } } - // process message and emit events - await processMessage( - msg, - { - downloadHistory, - ev, - historyCache, - recvChats, - creds: authState.creds, - keyStore: authState.keys, - logger, - options: config.options, + const historyMsg = getHistoryMsg(msg.message!) + const shouldProcessHistoryMsg = historyMsg + ? ( + shouldSyncHistoryMessage(historyMsg) + && PROCESSABLE_HISTORY_TYPES.includes(historyMsg.syncType!) + ) + : false + // we should have app state keys before we process any history + if(shouldProcessHistoryMsg) { + if(!authState.creds.myAppStateKeyId) { + logger.warn('myAppStateKeyId not synced, bad link') + await logout('Incomplete app state key sync') + return } - ) - - const isAnyHistoryMsg = isHistoryMsg(msg.message!) - if(isAnyHistoryMsg) { - // we only want to sync app state once we've all the history - // restart the app state sync timeout - logger.debug('restarting app sync timeout') - appStateSyncTimeout.start() } + + await Promise.all([ + (async() => { + if(shouldProcessHistoryMsg && !authState.creds.accountSyncCounter) { + logger.info('doing initial app state sync') + await resyncAppState(ALL_WA_PATCH_NAMES, true) + + const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 + ev.emit('creds.update', { accountSyncCounter }) + } + })(), + processMessage( + msg, + { + shouldProcessHistoryMsg, + ev, + creds: authState.creds, + keyStore: authState.keys, + logger, + options: config.options, + } + ) + ]) }) ws.on('CB:presence', handlePresenceUpdate) @@ -814,7 +784,6 @@ export const makeChatsSocket = (config: SocketConfig) => { updateBlockStatus, getBusinessProfile, resyncAppState, - chatModify, - resyncMainAppState, + chatModify } } diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index c6403af..af82533 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -2,7 +2,7 @@ import { proto } from '../../WAProto' import { KEY_BUNDLE_TYPE, MIN_PREKEY_COUNT } from '../Defaults' import { MessageReceiptType, MessageRelayOptions, MessageUserReceipt, SocketConfig, WACallEvent, WAMessageKey, WAMessageStubType, WAPatchName } from '../Types' -import { decodeMediaRetryNode, decodeMessageStanza, delay, encodeBigEndian, encodeSignedDeviceIdentity, getCallStatusFromNode, getNextPreKeys, getStatusFromReceiptType, isHistoryMsg, unixTimestampSeconds, xmppPreKey, xmppSignedPreKey } from '../Utils' +import { decodeMediaRetryNode, decodeMessageStanza, delay, encodeBigEndian, encodeSignedDeviceIdentity, getCallStatusFromNode, getHistoryMsg, getNextPreKeys, getStatusFromReceiptType, unixTimestampSeconds, xmppPreKey, xmppSignedPreKey } from '../Utils' import { makeMutex } from '../Utils/make-mutex' import { cleanMessage } from '../Utils/process-message' import { areJidsSameUser, BinaryNode, getAllBinaryNodeChildren, getBinaryNodeChild, getBinaryNodeChildren, isJidGroup, isJidUser, jidDecode, jidNormalizedUser, S_WHATSAPP_NET } from '../WABinary' @@ -286,7 +286,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { const update = getBinaryNodeChild(node, 'collection') if(update) { const name = update.attrs.name as WAPatchName - await resyncAppState([name], undefined) + await resyncAppState([name], false) } break @@ -529,7 +529,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { // send ack for history message - const isAnyHistoryMsg = isHistoryMsg(msg.message!) + const isAnyHistoryMsg = getHistoryMsg(msg.message!) if(isAnyHistoryMsg) { const jid = jidNormalizedUser(msg.key.remoteJid!) await sendReceipt(jid, undefined, [msg.key.id!], 'hist_sync') @@ -618,6 +618,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { logger.info(`handled ${offlineNotifs} offline messages/notifications`) await ev.flush() + ev.emit('connection.update', { receivedPendingNotifications: true }) }) diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 47ca73f..e53f97d 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -535,7 +535,7 @@ export const makeSocket = ({ const name = update.me?.name // if name has just been received if(creds.me?.name !== name) { - logger.info({ name }, 'updated pushName') + logger.debug({ name }, 'updated pushName') sendNode({ tag: 'presence', attrs: { name: name! } diff --git a/src/Store/make-in-memory-store.ts b/src/Store/make-in-memory-store.ts index aab2785..f8d7a6b 100644 --- a/src/Store/make-in-memory-store.ts +++ b/src/Store/make-in-memory-store.ts @@ -70,28 +70,30 @@ export default ( ev.on('connection.update', update => { Object.assign(state, update) }) - ev.on('chats.set', ({ chats: newChats, isLatest }) => { + + ev.on('messaging-history.set', ({ + chats: newChats, + contacts: newContacts, + messages: newMessages, + isLatest + }) => { if(isLatest) { chats.clear() + + for(const id in messages) { + delete messages[id] + } } const chatsAdded = chats.insertIfAbsent(...newChats).length logger.debug({ chatsAdded }, 'synced chats') - }) - ev.on('contacts.set', ({ contacts: newContacts }) => { + const oldContacts = contactsUpsert(newContacts) for(const jid of oldContacts) { delete contacts[jid] } logger.debug({ deletedContacts: oldContacts.size, newContacts }, 'synced contacts') - }) - ev.on('messages.set', ({ messages: newMessages, isLatest }) => { - if(isLatest) { - for(const id in messages) { - delete messages[id] - } - } for(const msg of newMessages) { const jid = msg.key.remoteJid! @@ -101,6 +103,7 @@ export default ( logger.debug({ messages: newMessages.length }, 'synced messages') }) + ev.on('contacts.update', updates => { for(const update of updates) { if(contacts[update.id!]) { diff --git a/src/Tests/test.event-buffer.ts b/src/Tests/test.event-buffer.ts index 5e088cd..71c98f7 100644 --- a/src/Tests/test.event-buffer.ts +++ b/src/Tests/test.event-buffer.ts @@ -8,7 +8,9 @@ describe('Event Buffer Tests', () => { let ev: ReturnType beforeEach(() => { - ev = makeEventBuffer(logger) + const _logger = logger.child({ }) + _logger.level = 'trace' + ev = makeEventBuffer(_logger) }) it('should buffer a chat upsert & update event', async() => { @@ -71,6 +73,123 @@ describe('Event Buffer Tests', () => { expect(chatsDeleted).toHaveLength(1) }) + it('should release a conditional update at the right time', async() => { + const chatId = randomJid() + const chatId2 = randomJid() + const chatsUpserted: Chat[] = [] + const chatsSynced: Chat[] = [] + + ev.on('chats.upsert', c => chatsUpserted.push(...c)) + ev.on('messaging-history.set', c => chatsSynced.push(...c.chats)) + ev.on('chats.update', () => fail('not should have emitted')) + + ev.buffer() + ev.emit('chats.update', [{ + id: chatId, + archived: true, + conditional(buff) { + if(buff.chatUpserts[chatId]) { + return true + } + } + }]) + ev.emit('chats.update', [{ + id: chatId2, + archived: true, + conditional(buff) { + if(buff.historySets.chats[chatId2]) { + return true + } + } + }]) + + await ev.flush() + + ev.buffer() + ev.emit('chats.upsert', [{ + id: chatId, + conversationTimestamp: 123, + unreadCount: 1, + muteEndTime: 123 + }]) + ev.emit('messaging-history.set', { + chats: [{ + id: chatId2, + conversationTimestamp: 123, + unreadCount: 1, + muteEndTime: 123 + }], + contacts: [], + messages: [], + isLatest: false + }) + await ev.flush() + + expect(chatsUpserted).toHaveLength(1) + expect(chatsUpserted[0].id).toEqual(chatId) + expect(chatsUpserted[0].archived).toEqual(true) + expect(chatsUpserted[0].muteEndTime).toEqual(123) + + expect(chatsSynced).toHaveLength(1) + expect(chatsSynced[0].id).toEqual(chatId2) + expect(chatsSynced[0].archived).toEqual(true) + }) + + it('should discard a conditional update', async() => { + const chatId = randomJid() + const chatsUpserted: Chat[] = [] + + ev.on('chats.upsert', c => chatsUpserted.push(...c)) + ev.on('chats.update', () => fail('not should have emitted')) + + ev.buffer() + ev.emit('chats.update', [{ + id: chatId, + archived: true, + conditional(buff) { + if(buff.chatUpserts[chatId]) { + return false + } + } + }]) + ev.emit('chats.upsert', [{ + id: chatId, + conversationTimestamp: 123, + unreadCount: 1, + muteEndTime: 123 + }]) + + await ev.flush() + + expect(chatsUpserted).toHaveLength(1) + expect(chatsUpserted[0].archived).toBeUndefined() + }) + + it('should overwrite a chats.update event with a history event', async() => { + const chatId = randomJid() + let chatRecv: Chat | undefined + + ev.on('messaging-history.set', ({ chats }) => { + chatRecv = chats[0] + }) + ev.on('chats.update', () => fail('not should have emitted')) + + ev.buffer() + + ev.emit('messaging-history.set', { + chats: [{ id: chatId, conversationTimestamp: 123, unreadCount: 1 }], + messages: [], + contacts: [], + isLatest: true + }) + ev.emit('chats.update', [{ id: chatId, archived: true }]) + + await ev.flush() + + expect(chatRecv).toBeDefined() + expect(chatRecv?.archived).toBeTruthy() + }) + it('should buffer message upsert events', async() => { const messageTimestamp = unixTimestampSeconds() const msg: proto.IWebMessageInfo = { diff --git a/src/Types/Chat.ts b/src/Types/Chat.ts index 6d9eff2..cd29cba 100644 --- a/src/Types/Chat.ts +++ b/src/Types/Chat.ts @@ -1,5 +1,6 @@ import type { proto } from '../../WAProto' import type { AccountSettings } from './Auth' +import type { BufferedEventData } from './Events' import type { MinimalMessage } from './Message' /** set of statuses visible to other people; see updatePresence() in WhatsAppWeb.Send */ @@ -33,7 +34,23 @@ export type WAPatchCreate = { operation: proto.SyncdMutation.SyncdOperation } -export type Chat = proto.IConversation +export type Chat = proto.IConversation & { + /** unix timestamp of when the last message was received in the chat */ + lastMessageRecvTimestamp?: number +} + +export type ChatUpdate = Partial boolean | undefined +}> /** * the last messages in a chat, sorted reverse-chronologically. That is, the latest message should be first in the chat @@ -77,6 +94,5 @@ export type InitialReceivedChatsState = { } export type InitialAppStateSyncOptions = { - recvChats: InitialReceivedChatsState accountSettings: AccountSettings } \ No newline at end of file diff --git a/src/Types/Events.ts b/src/Types/Events.ts index 341a129..4e1866e 100644 --- a/src/Types/Events.ts +++ b/src/Types/Events.ts @@ -2,7 +2,7 @@ import type { Boom } from '@hapi/boom' import { proto } from '../../WAProto' import { AuthenticationCreds } from './Auth' import { WACallEvent } from './Call' -import { Chat, PresenceData } from './Chat' +import { Chat, ChatUpdate, PresenceData } from './Chat' import { Contact } from './Contact' import { GroupMetadata, ParticipantAction } from './GroupMetadata' import { MessageUpsertType, MessageUserReceiptUpdate, WAMessage, WAMessageKey, WAMessageUpdate } from './Message' @@ -13,16 +13,17 @@ export type BaileysEventMap = { 'connection.update': Partial /** credentials updated -- some metadata, keys or something */ 'creds.update': Partial - /** set chats (history sync), chats are reverse chronologically sorted */ - 'chats.set': { chats: Chat[], isLatest: boolean } - /** set messages (history sync), messages are reverse chronologically sorted */ - 'messages.set': { messages: WAMessage[], isLatest: boolean } - /** set contacts (history sync) */ - 'contacts.set': { contacts: Contact[], isLatest: boolean } + /** set chats (history sync), everything is reverse chronologically sorted */ + 'messaging-history.set': { + chats: Chat[] + contacts: Contact[] + messages: WAMessage[] + isLatest: boolean + } /** upsert chats */ 'chats.upsert': Chat[] /** update the given chats */ - 'chats.update': Partial[] + 'chats.update': ChatUpdate[] /** delete chats with given ID */ 'chats.delete': string[] /** presence of contact in a chat updated */ @@ -56,8 +57,15 @@ export type BaileysEventMap = { } export type BufferedEventData = { + historySets: { + chats: { [jid: string]: Chat } + contacts: { [jid: string]: Contact } + messages: { [uqId: string]: WAMessage } + empty: boolean + isLatest: boolean + } chatUpserts: { [jid: string]: Chat } - chatUpdates: { [jid: string]: Partial } + chatUpdates: { [jid: string]: ChatUpdate } chatDeletes: Set contactUpserts: { [jid: string]: Contact } contactUpdates: { [jid: string]: Partial } diff --git a/src/Types/Socket.ts b/src/Types/Socket.ts index 3a00333..f693a37 100644 --- a/src/Types/Socket.ts +++ b/src/Types/Socket.ts @@ -46,8 +46,8 @@ export type SocketConfig = { qrTimeout?: number; /** provide an auth state object to maintain the auth state */ auth: AuthenticationState - /** By default true, should history messages be downloaded and processed */ - downloadHistory: boolean + /** manage history processing with this control; by default will sync up everything */ + shouldSyncHistoryMessage: (msg: proto.Message.IHistorySyncNotification) => boolean /** transaction capability options for SignalKeyStore */ transactionOpts: TransactionCapabilityOptions /** provide a cache to store a user's device list */ diff --git a/src/Utils/chat-utils.ts b/src/Utils/chat-utils.ts index 56b6379..b3e8d5c 100644 --- a/src/Utils/chat-utils.ts +++ b/src/Utils/chat-utils.ts @@ -2,7 +2,7 @@ import { Boom } from '@hapi/boom' import { AxiosRequestConfig } from 'axios' import type { Logger } from 'pino' import { proto } from '../../WAProto' -import { BaileysEventEmitter, ChatModification, ChatMutation, Contact, InitialAppStateSyncOptions, LastMessageList, LTHashState, WAPatchCreate, WAPatchName } from '../Types' +import { BaileysEventEmitter, Chat, ChatModification, ChatMutation, ChatUpdate, Contact, InitialAppStateSyncOptions, LastMessageList, LTHashState, WAPatchCreate, WAPatchName } from '../Types' import { BinaryNode, getBinaryNodeChild, getBinaryNodeChildren, isJidGroup, jidNormalizedUser } from '../WABinary' import { aesDecrypt, aesEncrypt, hkdf, hmacSign } from './crypto' import { toNumber } from './generics' @@ -618,7 +618,6 @@ export const processSyncAction = ( logger?: Logger, ) => { const isInitialSync = !!initialSyncOpts - const recvChats = initialSyncOpts?.recvChats const accountSettings = initialSyncOpts?.accountSettings const { @@ -631,13 +630,14 @@ export const processSyncAction = ( [ { id, - muteEndTime: action.muteAction?.muted ? - toNumber(action.muteAction!.muteEndTimestamp!) : - null + muteEndTime: action.muteAction?.muted + ? toNumber(action.muteAction!.muteEndTimestamp!) + : null, + conditional: getChatUpdateConditional(id, undefined) } ] ) - } else if(action?.archiveChatAction) { + } else if(action?.archiveChatAction || type === 'archive' || type === 'unarchive') { // okay so we've to do some annoying computation here // when we're initially syncing the app state // there are a few cases we need to handle @@ -648,35 +648,36 @@ export const processSyncAction = ( // we compare the timestamp of latest message from the other person to determine this // 2. if the account unarchiveChats setting is false -- then it doesn't matter, // it'll always take an app state action to mark in unarchived -- which we'll get anyway - const archiveAction = action.archiveChatAction - if( - isValidPatchBasedOnMessageRange(id, archiveAction.messageRange) - || !isInitialSync - || !accountSettings?.unarchiveChats - ) { - // basically we don't need to fire an "archive" update if the chat is being marked unarchvied - // this only applies for the initial sync - if(isInitialSync && !archiveAction.archived) { - ev.emit('chats.update', [{ id, archived: false }]) - } else { - ev.emit('chats.update', [{ id, archived: !!archiveAction?.archived }]) - } - } + const archiveAction = action?.archiveChatAction + const isArchived = archiveAction + ? archiveAction.archived + : type === 'archive' + // // basically we don't need to fire an "archive" update if the chat is being marked unarchvied + // // this only applies for the initial sync + // if(isInitialSync && !isArchived) { + // isArchived = false + // } + + const msgRange = !accountSettings?.unarchiveChats ? undefined : archiveAction?.messageRange + // logger?.debug({ chat: id, syncAction }, 'message range archive') + + ev.emit('chats.update', [{ + id, + archived: isArchived, + conditional: getChatUpdateConditional(id, msgRange) + }]) } else if(action?.markChatAsReadAction) { const markReadAction = action.markChatAsReadAction - if( - isValidPatchBasedOnMessageRange(id, markReadAction.messageRange) - || !isInitialSync - ) { - // basically we don't need to fire an "read" update if the chat is being marked as read - // because the chat is read by default - // this only applies for the initial sync - if(isInitialSync && markReadAction.read) { - ev.emit('chats.update', [{ id, unreadCount: null }]) - } else { - ev.emit('chats.update', [{ id, unreadCount: !!markReadAction?.read ? 0 : -1 }]) - } - } + // basically we don't need to fire an "read" update if the chat is being marked as read + // because the chat is read by default + // this only applies for the initial sync + const isNullUpdate = isInitialSync && markReadAction.read + + ev.emit('chats.update', [{ + id, + unreadCount: isNullUpdate ? null : !!markReadAction?.read ? 0 : -1, + conditional: getChatUpdateConditional(id, markReadAction?.messageRange) + }]) } else if(action?.deleteMessageForMeAction || type === 'deleteMessageForMe') { ev.emit('messages.delete', { keys: [ { @@ -688,11 +689,16 @@ export const processSyncAction = ( } else if(action?.contactAction) { ev.emit('contacts.upsert', [{ id, name: action.contactAction!.fullName! }]) } else if(action?.pushNameSetting) { - if(me?.name !== action?.pushNameSetting) { - ev.emit('creds.update', { me: { ...me, name: action?.pushNameSetting?.name! } }) + const name = action?.pushNameSetting?.name + if(name && me?.name !== name) { + ev.emit('creds.update', { me: { ...me, name } }) } } else if(action?.pinAction) { - ev.emit('chats.update', [{ id, pinned: action.pinAction?.pinned ? toNumber(action.timestamp!) : null }]) + ev.emit('chats.update', [{ + id, + pinned: action.pinAction?.pinned ? toNumber(action.timestamp!) : null, + conditional: getChatUpdateConditional(id, undefined) + }]) } else if(action?.unarchiveChatsSetting) { const unarchiveChats = !!action.unarchiveChatsSetting.unarchiveChats ev.emit('creds.update', { accountSettings: { unarchiveChats } }) @@ -714,23 +720,27 @@ export const processSyncAction = ( } ]) } else if(action?.deleteChatAction || type === 'deleteChat') { - if( - ( - action?.deleteChatAction?.messageRange - && isValidPatchBasedOnMessageRange(id, action?.deleteChatAction?.messageRange) - ) - || !isInitialSync - ) { + if(!isInitialSync) { ev.emit('chats.delete', [id]) } } else { logger?.debug({ syncAction, id }, 'unprocessable update') } - function isValidPatchBasedOnMessageRange(id: string, msgRange: proto.SyncActionValue.ISyncActionMessageRange | null | undefined) { - const chat = recvChats?.[id] + function getChatUpdateConditional(id: string, msgRange: proto.SyncActionValue.ISyncActionMessageRange | null | undefined): ChatUpdate['conditional'] { + return isInitialSync + ? (data) => { + const chat = data.historySets.chats[id] || data.chatUpserts[id] + if(chat) { + return msgRange ? isValidPatchBasedOnMessageRange(chat, msgRange) : true + } + } + : undefined + } + + function isValidPatchBasedOnMessageRange(chat: Chat, msgRange: proto.SyncActionValue.ISyncActionMessageRange | null | undefined) { const lastMsgTimestamp = msgRange?.lastMessageTimestamp || msgRange?.lastSystemMessageTimestamp || 0 - const chatLastMsgTimestamp = chat?.lastMsgRecvTimestamp || 0 + const chatLastMsgTimestamp = chat?.lastMessageRecvTimestamp || 0 return lastMsgTimestamp >= chatLastMsgTimestamp } } \ No newline at end of file diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index 9ed314b..9888f64 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -1,11 +1,13 @@ import EventEmitter from 'events' import { Logger } from 'pino' import { proto } from '../../WAProto' -import { BaileysEvent, BaileysEventEmitter, BaileysEventMap, BufferedEventData, Chat, Contact, WAMessage, WAMessageStatus } from '../Types' +import { BaileysEvent, BaileysEventEmitter, BaileysEventMap, BufferedEventData, Chat, ChatUpdate, Contact, WAMessage, WAMessageStatus } from '../Types' +import { trimUndefineds } from './generics' import { updateMessageWithReaction, updateMessageWithReceipt } from './messages' import { isRealMessage, shouldIncrementChatUnread } from './process-message' const BUFFERABLE_EVENT = [ + 'messaging-history.set', 'chats.upsert', 'chats.update', 'chats.delete', @@ -55,10 +57,10 @@ type BaileysBufferableEventEmitter = BaileysEventEmitter & { */ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter => { const ev = new EventEmitter() + const historyCache = new Set() let data = makeBufferData() let isBuffering = false - let preBufferTask: Promise = Promise.resolve() // take the generic event and fire it as a baileys event @@ -88,14 +90,29 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter = isBuffering = false + const newData = makeBufferData() + const chatUpdates = Object.values(data.chatUpdates) + // gather the remaining conditional events so we re-queue them + let conditionalChatUpdatesLeft = 0 + for(const update of chatUpdates) { + if(update.conditional) { + conditionalChatUpdatesLeft += 1 + newData.chatUpdates[update.id!] = update + delete data.chatUpdates[update.id!] + } + } + const consolidatedData = consolidateEvents(data) if(Object.keys(consolidatedData).length) { ev.emit('event', consolidatedData) } - data = makeBufferData() + data = newData - logger.trace('released buffered events') + logger.trace( + { conditionalChatUpdatesLeft }, + 'released buffered events' + ) } return { @@ -111,7 +128,7 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter = }, emit(event: BaileysEvent, evData: BaileysEventMap[T]) { if(isBuffering && BUFFERABLE_EVENT_SET.has(event)) { - append(data, event as any, evData, logger) + append(data, historyCache, event as any, evData, logger) return true } @@ -145,6 +162,13 @@ export const makeEventBuffer = (logger: Logger): BaileysBufferableEventEmitter = const makeBufferData = (): BufferedEventData => { return { + historySets: { + chats: { }, + messages: { }, + contacts: { }, + isLatest: false, + empty: true + }, chatUpserts: { }, chatUpdates: { }, chatDeletes: new Set(), @@ -161,41 +185,100 @@ const makeBufferData = (): BufferedEventData => { function append( data: BufferedEventData, + historyCache: Set, event: E, eventData: any, logger: Logger ) { switch (event) { + case 'messaging-history.set': + for(const chat of eventData.chats as Chat[]) { + const existingChat = data.historySets.chats[chat.id] + if(existingChat) { + existingChat.endOfHistoryTransferType = chat.endOfHistoryTransferType + } + + if(!existingChat && !historyCache.has(chat.id)) { + data.historySets.chats[chat.id] = chat + historyCache.add(chat.id) + + absorbingChatUpdate(chat) + } + } + + for(const contact of eventData.contacts as Contact[]) { + const existingContact = data.historySets.contacts[contact.id] + if(existingContact) { + Object.assign(existingContact, trimUndefineds(contact)) + } else { + const historyContactId = `c:${contact.id}` + const hasAnyName = contact.notify || contact.name || contact.verifiedName + if(!historyCache.has(historyContactId) || hasAnyName) { + data.historySets.contacts[contact.id] = contact + historyCache.add(historyContactId) + } + } + } + + for(const message of eventData.messages as WAMessage[]) { + const key = stringifyMessageKey(message.key) + const existingMsg = data.historySets.messages[key] + if(!existingMsg && !historyCache.has(key)) { + data.historySets.messages[key] = message + historyCache.add(key) + } + } + + data.historySets.empty = false + data.historySets.isLatest = eventData.isLatest || data.historySets.isLatest + + break case 'chats.upsert': for(const chat of eventData as Chat[]) { - let upsert = data.chatUpserts[chat.id] || { } as Chat - upsert = concatChats(upsert, chat) - if(data.chatUpdates[chat.id]) { - logger.debug({ chatId: chat.id }, 'absorbed chat update in chat upsert') - upsert = concatChats(data.chatUpdates[chat.id] as Chat, upsert) - delete data.chatUpdates[chat.id] + let upsert = data.chatUpserts[chat.id] + if(!upsert) { + upsert = data.historySets[chat.id] + if(upsert) { + logger.debug({ chatId: chat.id }, 'absorbed chat upsert in chat set') + } } + if(upsert) { + upsert = concatChats(upsert, chat) + } else { + upsert = chat + data.chatUpserts[chat.id] = upsert + } + + absorbingChatUpdate(upsert) + if(data.chatDeletes.has(chat.id)) { data.chatDeletes.delete(chat.id) } - - data.chatUpserts[chat.id] = upsert } break case 'chats.update': - for(const update of eventData as Partial[]) { + for(const update of eventData as ChatUpdate[]) { const chatId = update.id! - // if there is an existing upsert, merge the update into it - const upsert = data.chatUpserts[chatId] - if(upsert) { - concatChats(upsert, update) - } else { - // merge the update into the existing update - const chatUpdate = data.chatUpdates[chatId] || { } - data.chatUpdates[chatId] = concatChats(chatUpdate, update) + const conditionMatches = update.conditional ? update.conditional(data) : true + if(conditionMatches) { + delete update.conditional + + // if there is an existing upsert, merge the update into it + const upsert = data.historySets.chats[chatId] || data.chatUpserts[chatId] + if(upsert) { + concatChats(upsert, update) + } else { + // merge the update into the existing update + const chatUpdate = data.chatUpdates[chatId] || { } + data.chatUpdates[chatId] = concatChats(chatUpdate, update) + } + } else if(conditionMatches === undefined) { + // condition yet to be fulfilled + data.chatUpdates[chatId] = update } + // otherwise -- condition not met, update is invalid // if the chat has been updated // ignore any existing chat delete @@ -207,7 +290,10 @@ function append( break case 'chats.delete': for(const chatId of eventData as string[]) { - data.chatDeletes.add(chatId) + if(!data.chatDeletes.has(chatId)) { + data.chatDeletes.add(chatId) + } + // remove any prior updates & upserts if(data.chatUpdates[chatId]) { delete data.chatUpdates[chatId] @@ -215,20 +301,36 @@ function append( if(data.chatUpserts[chatId]) { delete data.chatUpserts[chatId] + + } + + if(data.historySets.chats[chatId]) { + delete data.historySets.chats[chatId] } } break case 'contacts.upsert': for(const contact of eventData as Contact[]) { - let upsert = data.contactUpserts[contact.id] || { } as Contact - upsert = Object.assign(upsert, contact) - if(data.contactUpdates[contact.id]) { - upsert = Object.assign(data.contactUpdates[contact.id], upsert) - delete data.contactUpdates[contact.id] + let upsert = data.contactUpserts[contact.id] + if(!upsert) { + upsert = data.historySets.contacts[contact.id] + if(upsert) { + logger.debug({ contactId: contact.id }, 'absorbed contact upsert in contact set') + } } - data.contactUpserts[contact.id] = upsert + if(upsert) { + upsert = Object.assign(upsert, trimUndefineds(contact)) + } else { + upsert = contact + data.contactUpserts[contact.id] = upsert + } + + if(data.contactUpdates[contact.id]) { + upsert = Object.assign(data.contactUpdates[contact.id], trimUndefineds(contact)) + delete data.contactUpdates[contact.id] + } } break @@ -237,7 +339,7 @@ function append( for(const update of contactUpdates) { const id = update.id! // merge into prior upsert - const upsert = data.contactUpserts[update.id!] + const upsert = data.historySets.contacts[id] || data.contactUpserts[id] if(upsert) { Object.assign(upsert, update) } else { @@ -252,9 +354,16 @@ function append( const { messages, type } = eventData as BaileysEventMap['messages.upsert'] for(const message of messages) { const key = stringifyMessageKey(message.key) - const existing = data.messageUpserts[key] + let existing = data.messageUpserts[key]?.message + if(!existing) { + existing = data.historySets.messages[key] + if(existing) { + logger.debug({ messageId: key }, 'absorbed message upsert in message set') + } + } + if(existing) { - message.messageTimestamp = existing.message.messageTimestamp + message.messageTimestamp = existing.messageTimestamp } if(data.messageUpdates[key]) { @@ -263,11 +372,15 @@ function append( delete data.messageUpdates[key] } - data.messageUpserts[key] = { - message, - type: type === 'notify' || existing?.type === 'notify' - ? 'notify' - : type + if(data.historySets.messages[key]) { + data.historySets.messages[key] = message + } else { + data.messageUpserts[key] = { + message, + type: type === 'notify' || data.messageUpserts[key]?.type === 'notify' + ? 'notify' + : type + } } } @@ -276,14 +389,14 @@ function append( const msgUpdates = eventData as BaileysEventMap['messages.update'] for(const { key, update } of msgUpdates) { const keyStr = stringifyMessageKey(key) - const existing = data.messageUpserts[keyStr] + const existing = data.historySets.messages[keyStr] || data.messageUpserts[keyStr]?.message if(existing) { - Object.assign(existing.message, update) + Object.assign(existing, update) // if the message was received & read by us // the chat counter must have been incremented // so we need to decrement it if(update.status === WAMessageStatus.READ && !key.fromMe) { - decrementChatReadCounterIfMsgDidUnread(existing.message) + decrementChatReadCounterIfMsgDidUnread(existing) } } else { const msgUpdate = data.messageUpdates[keyStr] || { key, update: { } } @@ -299,7 +412,10 @@ function append( const { keys } = deleteData for(const key of keys) { const keyStr = stringifyMessageKey(key) - data.messageDeletes[keyStr] = key + if(!data.messageDeletes[keyStr]) { + data.messageDeletes[keyStr] = key + + } if(data.messageUpserts[keyStr]) { delete data.messageUpserts[keyStr] @@ -349,7 +465,10 @@ function append( for(const update of groupUpdates) { const id = update.id! const groupUpdate = data.groupUpdates[id] || { } - data.groupUpdates[id] = Object.assign(groupUpdate, update) + if(!data.groupUpdates[id]) { + data.groupUpdates[id] = Object.assign(groupUpdate, update) + + } } break @@ -357,6 +476,23 @@ function append( throw new Error(`"${event}" cannot be buffered`) } + function absorbingChatUpdate(existing: Chat) { + const chatId = existing.id + const update = data.chatUpdates[chatId] + if(update) { + const conditionMatches = update.conditional ? update.conditional(data) : true + if(conditionMatches) { + delete update.conditional + logger.debug({ chatId }, 'absorbed chat update in existing chat') + Object.assign(existing, concatChats(update as Chat, existing)) + delete data.chatUpdates[chatId] + } else if(conditionMatches === false) { + logger.debug({ chatId }, 'chat update condition fail, removing') + delete data.chatUpdates[chatId] + } + } + } + function decrementChatReadCounterIfMsgDidUnread(message: WAMessage) { // decrement chat unread counter // if the message has already been marked read by us @@ -380,6 +516,15 @@ function append( function consolidateEvents(data: BufferedEventData) { const map: BaileysEventData = { } + if(!data.historySets.empty) { + map['messaging-history.set'] = { + chats: Object.values(data.historySets.chats), + messages: Object.values(data.historySets.messages), + contacts: Object.values(data.historySets.contacts), + isLatest: data.historySets.isLatest + } + } + const chatUpsertList = Object.values(data.chatUpserts) if(chatUpsertList.length) { map['chats.upsert'] = chatUpsertList @@ -446,7 +591,7 @@ function consolidateEvents(data: BufferedEventData) { return map } -function concatChats>(a: C, b: C) { +function concatChats>(a: C, b: Partial) { if(b.unreadCount === null) { // neutralize unread counter if(a.unreadCount! < 0) { diff --git a/src/Utils/generics.ts b/src/Utils/generics.ts index bf49d98..ea730b6 100644 --- a/src/Utils/generics.ts +++ b/src/Utils/generics.ts @@ -366,4 +366,14 @@ export const getCodeFromWSError = (error: Error) => { */ export const isWABusinessPlatform = (platform: string) => { return platform === 'smbi' || platform === 'smba' +} + +export function trimUndefineds(obj: any) { + for(const key in obj) { + if(typeof obj[key] === 'undefined') { + delete obj[key] + } + } + + return obj } \ No newline at end of file diff --git a/src/Utils/history.ts b/src/Utils/history.ts index 35aa8e0..9c49291 100644 --- a/src/Utils/history.ts +++ b/src/Utils/history.ts @@ -2,7 +2,7 @@ import { AxiosRequestConfig } from 'axios' import { promisify } from 'util' import { inflate } from 'zlib' import { proto } from '../../WAProto' -import { Chat, Contact, InitialReceivedChatsState } from '../Types' +import { Chat, Contact, WAMessageStubType } from '../Types' import { isJidUser } from '../WABinary' import { toNumber } from './generics' import { normalizeMessageContent } from './messages' @@ -29,11 +29,7 @@ export const downloadHistory = async( return syncData } -export const processHistoryMessage = ( - item: proto.IHistorySync, - historyCache: Set, - recvChats: InitialReceivedChatsState -) => { +export const processHistoryMessage = (item: proto.IHistorySync) => { const messages: proto.IWebMessageInfo[] = [] const contacts: Contact[] = [] const chats: Chat[] = [] @@ -41,91 +37,75 @@ export const processHistoryMessage = ( switch (item.syncType) { case proto.HistorySync.HistorySyncType.INITIAL_BOOTSTRAP: case proto.HistorySync.HistorySyncType.RECENT: - for(const chat of item.conversations!) { - const contactId = `c:${chat.id}` - if(chat.name && !historyCache.has(contactId)) { - contacts.push({ id: chat.id, name: chat.name }) - historyCache.add(contactId) - } + case proto.HistorySync.HistorySyncType.FULL: + for(const chat of item.conversations! as Chat[]) { + contacts.push({ id: chat.id, name: chat.name || undefined }) const msgs = chat.messages || [] delete chat.messages + delete chat.archived + delete chat.muteEndTime + delete chat.pinned for(const item of msgs) { const message = item.message! - const uqId = `${message.key.remoteJid}:${message.key.id}` - if(!historyCache.has(uqId)) { - messages.push(message) + messages.push(message) - let curItem = recvChats[message.key.remoteJid!] - const timestamp = toNumber(message.messageTimestamp) - if(!curItem || timestamp > curItem.lastMsgTimestamp) { - curItem = { lastMsgTimestamp: timestamp } - recvChats[chat.id] = curItem - // keep only the most recent message in the chat array - chat.messages = [{ message }] - } + if(!chat.messages) { + // keep only the most recent message in the chat array + chat.messages = [{ message }] + } - if( - !message.key.fromMe - && (!curItem?.lastMsgRecvTimestamp || timestamp > curItem.lastMsgRecvTimestamp) - ) { - curItem.lastMsgRecvTimestamp = timestamp - } + if(!message.key.fromMe && !chat.lastMessageRecvTimestamp) { + chat.lastMessageRecvTimestamp = toNumber(message.messageTimestamp) + } - historyCache.add(uqId) + if( + !message.key.fromMe + && message.messageStubType === WAMessageStubType.BIZ_PRIVACY_MODE_TO_BSP + && message.messageStubParameters?.[0] + ) { + contacts.push({ + id: message.key.participant || message.key.remoteJid!, + verifiedName: message.messageStubParameters?.[0], + }) } } - if(!historyCache.has(chat.id)) { - if(isJidUser(chat.id) && chat.readOnly && chat.archived) { - chat.readOnly = false - } - - chats.push(chat) - historyCache.add(chat.id) + if(isJidUser(chat.id) && chat.readOnly && chat.archived) { + delete chat.readOnly } + + chats.push({ ...chat }) } break case proto.HistorySync.HistorySyncType.PUSH_NAME: for(const c of item.pushnames!) { - const contactId = `c:${c.id}` - if(!historyCache.has(contactId)) { - contacts.push({ notify: c.pushname!, id: c.id! }) - historyCache.add(contactId) - } + contacts.push({ notify: c.pushname!, id: c.id! }) } - break - case proto.HistorySync.HistorySyncType.INITIAL_STATUS_V3: - // TODO break } - const didProcess = !!(chats.length || messages.length || contacts.length) - return { chats, contacts, messages, - didProcess, } } export const downloadAndProcessHistorySyncNotification = async( msg: proto.Message.IHistorySyncNotification, - historyCache: Set, - recvChats: InitialReceivedChatsState, options: AxiosRequestConfig ) => { const historyMsg = await downloadHistory(msg, options) - return processHistoryMessage(historyMsg, historyCache, recvChats) + return processHistoryMessage(historyMsg) } -export const isHistoryMsg = (message: proto.IMessage) => { +export const getHistoryMsg = (message: proto.IMessage) => { const normalizedContent = !!message ? normalizeMessageContent(message) : undefined - const isAnyHistoryMsg = !!normalizedContent?.protocolMessage?.historySyncNotification + const anyHistoryMsg = normalizedContent?.protocolMessage?.historySyncNotification - return isAnyHistoryMsg + return anyHistoryMsg } \ No newline at end of file diff --git a/src/Utils/process-message.ts b/src/Utils/process-message.ts index acf2eec..5ef9529 100644 --- a/src/Utils/process-message.ts +++ b/src/Utils/process-message.ts @@ -1,14 +1,12 @@ import { AxiosRequestConfig } from 'axios' import type { Logger } from 'pino' import { proto } from '../../WAProto' -import { AuthenticationCreds, BaileysEventEmitter, Chat, GroupMetadata, InitialReceivedChatsState, ParticipantAction, SignalKeyStoreWithTransaction, WAMessageStubType } from '../Types' +import { AuthenticationCreds, BaileysEventEmitter, Chat, GroupMetadata, ParticipantAction, SignalKeyStoreWithTransaction, WAMessageStubType } from '../Types' import { downloadAndProcessHistorySyncNotification, normalizeMessageContent, toNumber } from '../Utils' import { areJidsSameUser, jidNormalizedUser } from '../WABinary' type ProcessMessageContext = { - historyCache: Set - recvChats: InitialReceivedChatsState - downloadHistory: boolean + shouldProcessHistoryMsg: boolean creds: AuthenticationCreds keyStore: SignalKeyStoreWithTransaction ev: BaileysEventEmitter @@ -66,7 +64,14 @@ export const shouldIncrementChatUnread = (message: proto.IWebMessageInfo) => ( const processMessage = async( message: proto.IWebMessageInfo, - { downloadHistory, ev, historyCache, recvChats, creds, keyStore, logger, options }: ProcessMessageContext + { + shouldProcessHistoryMsg, + ev, + creds, + keyStore, + logger, + options + }: ProcessMessageContext ) => { const meId = creds.me!.id const { accountSettings } = creds @@ -92,38 +97,30 @@ const processMessage = async( switch (protocolMsg.type) { case proto.Message.ProtocolMessage.Type.HISTORY_SYNC_NOTIFICATION: const histNotification = protocolMsg!.historySyncNotification! + const process = shouldProcessHistoryMsg + const isLatest = !creds.processedHistoryMessages?.length - logger?.info({ histNotification, id: message.key.id }, 'got history notification') + logger?.info({ + histNotification, + process, + id: message.key.id, + isLatest, + }, 'got history notification') - if(downloadHistory) { - const isLatest = !creds.processedHistoryMessages?.length - const { chats, contacts, messages, didProcess } = await downloadAndProcessHistorySyncNotification( + if(process) { + const data = await downloadAndProcessHistorySyncNotification( histNotification, - historyCache, - recvChats, options ) - if(chats.length) { - ev.emit('chats.set', { chats, isLatest }) - } + ev.emit('messaging-history.set', { ...data, isLatest }) - if(messages.length) { - ev.emit('messages.set', { messages, isLatest }) - } - - if(contacts.length) { - ev.emit('contacts.set', { contacts, isLatest }) - } - - if(didProcess) { - ev.emit('creds.update', { - processedHistoryMessages: [ - ...(creds.processedHistoryMessages || []), - { key: message.key, messageTimestamp: message.messageTimestamp } - ] - }) - } + ev.emit('creds.update', { + processedHistoryMessages: [ + ...(creds.processedHistoryMessages || []), + { key: message.key, messageTimestamp: message.messageTimestamp } + ] + }) } break