franpcode/Baileys
1
0
Fork 0
mirror of https://github.com/FranP-code/Baileys.git synced 2025-10-13 00:32:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor!: cleaner message history sync

Browse Source 
This is a breaking change,
1. three events (chats.set, contacts.set, messages.set) are now just one `messaging-history.set` event
2. no need to debounce for app state sync
3. added a new "conditional" chat update to allow for correct app state sync despite not having the chat available on hand
This commit is contained in:
Adhiraj Singh
2022-09-29 16:32:57 +05:30
parent e08dd10198
commit d0330d1863
16 changed files with 600 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/Defaults/index.ts
View File

@@ -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,

127
src/Socket/chats.ts
View File

@@ -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<string>()
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
}
}

7
src/Socket/messages-recv.ts
View File

@@ -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 })
})

2
src/Socket/socket.ts
View File

@@ -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! }

23
src/Store/make-in-memory-store.ts
View File

@@ -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!]) {

121
src/Tests/test.event-buffer.ts
View File

@@ -8,7 +8,9 @@ describe('Event Buffer Tests', () => {
let ev: ReturnType<typeof makeEventBuffer>
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 = {

20
src/Types/Chat.ts
View File

@@ -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<Chat & {
/**
* if specified in the update,
* the EV buffer will check if the condition gets fulfilled before applying the update
* Right now, used to determine when to release an app state sync event
*
* @returns true, if the update should be applied;
* false if it can be discarded;
* undefined if the condition is not yet fulfilled
* */
conditional: (bufferedData: BufferedEventData) => 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
}

26
src/Types/Events.ts
View File

@@ -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<ConnectionState>
/** credentials updated -- some metadata, keys or something */
'creds.update': Partial<AuthenticationCreds>
/** 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<Chat>[]
'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<Chat> }
chatUpdates: { [jid: string]: ChatUpdate }
chatDeletes: Set<string>
contactUpserts: { [jid: string]: Contact }
contactUpdates: { [jid: string]: Partial<Contact> }

4
src/Types/Socket.ts
View File

@@ -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 */

102
src/Utils/chat-utils.ts
View File

@@ -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
}
}

231
src/Utils/event-buffer.ts
View File

@@ -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<string>()
let data = makeBufferData()
let isBuffering = false
let preBufferTask: Promise<any> = 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<T extends BaileysEvent>(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<E extends BufferableEvent>(
data: BufferedEventData,
historyCache: Set<string>,
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<Chat>[]) {
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<E extends BufferableEvent>(
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<C extends Partial<Chat>>(a: C, b: C) {
function concatChats<C extends Partial<Chat>>(a: C, b: Partial<Chat>) {
if(b.unreadCount === null) {
// neutralize unread counter
if(a.unreadCount! < 0) {

10
src/Utils/generics.ts
View File

@@ -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
}

88
src/Utils/history.ts
View File

@@ -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<string>,
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<string>,
recvChats: InitialReceivedChatsState,
options: AxiosRequestConfig<any>
) => {
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
}

57
src/Utils/process-message.ts
View File

@@ -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<string>
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