Per user mutex

src/Socket/chats.ts

@@ -1,4 +1,4 @@
 import NodeCache from '@cacheable/node-cache'
 import { Boom } from '@hapi/boom'
 import { proto } from '../../WAProto/index.js'
 import { DEFAULT_CACHE_TTLS, PROCESSABLE_HISTORY_TYPES } from '../Defaults'
@@ -28,7 +28,7 @@
 import { ALL_WA_PATCH_NAMES } from '../Types'
 import type { QuickReplyAction } from '../Types/Bussines.js'
 import type { LabelActionBody } from '../Types/Label'
-import { SyncState } from '../Types/State'
+import { SyncState } from '../Types'
 import {
 	chatModificationToAppPatch,
 	type ChatMutationMap,
@@ -41,7 +41,7 @@
 	newLTHashState,
 	processSyncAction
 } from '../Utils'
-import { makeMutex } from '../Utils/make-mutex'
+import { makeMutex , makeKeyedMutex} from '../Utils/make-mutex'
 import processMessage from '../Utils/process-message'
 import {
 	type BinaryNode,
@@ -74,16 +74,16 @@
 	let syncState: SyncState = SyncState.Connecting
 
 	/** this mutex ensures that messages are processed in order */
-	const messageMutex = makeMutex()
+	const messageMutex = makeKeyedMutex()
 
 	/** this mutex ensures that receipts are processed in order */
-	const receiptMutex = makeMutex()
+	const receiptMutex = makeKeyedMutex()
 
 	/** this mutex ensures that app state patches are processed in order */
 	const appStatePatchMutex = makeMutex()
 
 	/** this mutex ensures that notifications are processed in order */
-	const notificationMutex = makeMutex()
+	const notificationMutex = makeKeyedMutex()
 
 	// Timeout for AwaitingInitialSync state
 	let awaitingSyncTimeout: NodeJS.Timeout | undefined
@@ -575,7 +575,7 @@
 								// collection is done with sync
 								collectionsToHandle.delete(name)
 							}
 						} catch (error: any) {
 							// if retry attempts overshoot
 							// or key not found
 							const isIrrecoverableError =

src/Socket/messages-recv.ts

@@ -2,7 +2,7 @@
 import { Boom } from '@hapi/boom'
 import { randomBytes } from 'crypto'
 import Long from 'long'
-import { proto } from '../../WAProto/index.js'
+import { proto } from '../../WAProto'
 import { DEFAULT_CACHE_TTLS, KEY_BUNDLE_TYPE, MIN_PREKEY_COUNT } from '../Defaults'
 import type {
 	GroupParticipant,
@@ -41,7 +41,7 @@
 	xmppPreKey,
 	xmppSignedPreKey
 } from '../Utils'
-import { makeMutex } from '../Utils/make-mutex'
+import { makeKeyedMutex } from '../Utils/make-mutex'
 import {
 	areJidsSameUser,
 	type BinaryNode,
@@ -89,7 +89,7 @@
 	} = sock
 
 	/** this mutex ensures that each retryRequest will wait for the previous one to finish */
-	const retryMutex = makeMutex()
+	const retryMutex = makeKeyedMutex();
 
 	const msgRetryCache =
 		config.msgRetryCounterCache ||
@@ -185,7 +185,7 @@
 			return
 		}
 
 		let data: any
 		try {
 			data = JSON.parse(mexNode.content.toString())
 		} catch (error) {
@@ -281,7 +281,7 @@
 			case 'update':
 				const settingsNode = getBinaryNodeChild(child, 'settings')
 				if (settingsNode) {
 					const update: Record<string, any> = {}
 					const nameNode = getBinaryNodeChild(settingsNode, 'name')
 					if (nameNode?.content) update.name = nameNode.content.toString()
 
@@ -1045,7 +1045,7 @@
 			attrs.participant || attrs.from,
 			isLid ? authState.creds.me?.lid : authState.creds.me?.id
 		)
-		const remoteJid = !isNodeFromMe || isJidGroup(attrs.from) ? attrs.from : attrs.recipient
+		const remoteJid = !isNodeFromMe || isJidGroup(attrs.from) ? attrs.from : attrs.recipient;
 		const fromMe = !attrs.recipient || ((attrs.type === 'retry' || attrs.type === 'sender') && isNodeFromMe)
 
 		const key: proto.IMessageKey = {
@@ -1069,7 +1069,7 @@
 
 		try {
 			await Promise.all([
-				receiptMutex.mutex(async () => {
+				receiptMutex.mutex(remoteJid!,async () => {
 					const status = getStatusFromReceiptType(attrs.type)
 					if (
 						typeof status !== 'undefined' &&
@@ -1143,7 +1143,7 @@
 
 		try {
 			await Promise.all([
-				notificationMutex.mutex(async () => {
+				notificationMutex.mutex(remoteJid!,async () => {
 					const msg = await processNotification(node)
 					if (msg) {
 						const fromMe = areJidsSameUser(node.attrs.participant || remoteJid, authState.creds.me!.id)
@@ -1208,8 +1208,11 @@
 			}
 		}
 
-		if (msg.key?.remoteJid && msg.key?.id && messageRetryManager) {
-			messageRetryManager.addRecentMessage(msg.key.remoteJid, msg.key.id, msg.message!)
+		const remoteJid = msg.key.remoteJid!;
+		const msgId = msg.key.id!;
+
+		if (messageRetryManager && remoteJid && msgId ) {
+			messageRetryManager.addRecentMessage(remoteJid, msgId, msg.message!)
 			logger.debug(
 				{
 					jid: msg.key.remoteJid,
@@ -1220,7 +1223,7 @@
 		}
 
 		try {
-			await messageMutex.mutex(async () => {
+			await messageMutex.mutex(remoteJid,async () => {
 				await decrypt()
 				// message failed to decrypt
 				if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT && msg.category !== 'peer') {
@@ -1237,7 +1240,7 @@
 					logger.debug(`[handleMessage] Attempting retry request for failed decryption`)
 
 					// Handle both pre-key and normal retries in single mutex
-					await retryMutex.mutex(async () => {
+					await retryMutex.mutex(msgId,async () => {
 						try {
 							if (!ws.isOpen) {
 								logger.debug({ node }, 'Connection closed, skipping retry')

src/Socket/messages-send.ts

@@ -508,8 +508,8 @@
 		const meLid = authState.creds.me?.lid
 		const meLidUser = meLid ? jidDecode(meLid)?.user : null
 
 		const encryptionPromises = (patchedMessages as any).map(
 			async ({ recipientJid: jid, message: patchedMessage }: any) => {
 				if (!jid) return null
 				let msgToEncrypt = patchedMessage
 				if (dsmMessage) {
@@ -1098,7 +1098,7 @@
 								content.url = getUrlFromDirectPath(content.directPath!)
 
 								logger.debug({ directPath: media.directPath, key: result.key }, 'media update successful')
 							} catch (err: any) {
 								error = err
 							}
 						}
@@ -1199,7 +1199,7 @@
 				})
 				if (config.emitOwnEvents) {
 					process.nextTick(async () => {
-						await messageMutex.mutex(() => upsertMessage(fullMsg, 'append'))
+						await messageMutex.mutex(jid,() => upsertMessage(fullMsg, 'append'))
 					})
 				}