Merge branch 'master' into fix/pn-lid-session-reuse

fix: unify PN and LID session reuse for 1:1 sends (#293)

src/Socket/messages-send.ts

@@ -417,6 +417,39 @@ export const makeMessagesSocket = (config: SocketConfig) => {
 		)
 	}
 
+	const resolveSessionJid = async (jid: string) => {
+		if (isLidUser(jid) || isHostedLidUser(jid)) {
+			return jid
+		}
+
+		if (isPnUser(jid) || isHostedPnUser(jid)) {
+			return await signalRepository.lidMapping.getLIDForPN(jid) || jid
+		}
+
+		return jid
+	}
+
+	const canonicalizeSessionRecipients = async (recipientJids: string[]) => {
+		const uniqueRecipients = [...new Set(recipientJids)]
+		const canonicalRecipients: string[] = []
+
+		for (const jid of uniqueRecipients) {
+			const canonicalJid = await resolveSessionJid(jid)
+			if (!canonicalRecipients.includes(canonicalJid)) {
+				canonicalRecipients.push(canonicalJid)
+			}
+		}
+
+		if (canonicalRecipients.length !== uniqueRecipients.length || canonicalRecipients.some((jid, index) => jid !== uniqueRecipients[index])) {
+			logger.debug(
+				{ before: uniqueRecipients, after: canonicalRecipients },
+				'[SESSION] Canonicalized recipient addressing for session reuse'
+			)
+		}
+
+		return canonicalRecipients
+	}
+
 	const assertSessions = async (jids: string[], force?: boolean) => {
 		let didFetchNewSession = false
 		const uniqueJids = [...new Set(jids)] // Deduplicate JIDs
@@ -426,22 +459,32 @@ export const makeMessagesSocket = (config: SocketConfig) => {
 
 		// Check peerSessionsCache and validate sessions using libsignal loadSession
 		for (const jid of uniqueJids) {
-			const signalId = signalRepository.jidToSignalProtocolAddress(jid)
+			const canonicalJid = await resolveSessionJid(jid)
-			const cachedSession = peerSessionsCache.get(signalId)
+			const signalIds = [
+				signalRepository.jidToSignalProtocolAddress(jid),
+				signalRepository.jidToSignalProtocolAddress(canonicalJid)
+			]
+			const cachedSession = signalIds
+				.map(signalId => peerSessionsCache.get(signalId))
+				.find(session => session !== undefined)
+
 			if (cachedSession !== undefined) {
 				if (cachedSession && !force) {
 					continue // Session exists in cache
 				}
 			} else {
-				const sessionValidation = await signalRepository.validateSession(jid)
+				const sessionValidation = await signalRepository.validateSession(canonicalJid)
 				const hasSession = sessionValidation.exists
-				peerSessionsCache.set(signalId, hasSession)
+				for (const signalId of signalIds) {
+					peerSessionsCache.set(signalId, hasSession)
+				}
+
 				if (hasSession && !force) {
 					continue
 				}
 			}
 
-			jidsRequiringFetch.push(jid)
+			jidsRequiringFetch.push(canonicalJid)
 		}
 
 		if (jidsRequiringFetch.length) {
@@ -682,33 +725,6 @@ export const makeMessagesSocket = (config: SocketConfig) => {
 		return false
 	}
 
-	const canonicalizeCarouselRecipients = async (recipientJids: string[]) => {
-		const uniqueRecipients = [...new Set(recipientJids)]
-		const pnRecipients = uniqueRecipients.filter(jid => isPnUser(jid) || isHostedPnUser(jid))
-		const lidMappings = pnRecipients.length > 0
-			? await signalRepository.lidMapping.getLIDsForPNs(pnRecipients)
-			: []
-
-		const lidByPn = new Map((lidMappings || []).map(({ pn, lid }) => [pn, lid]))
-		const canonicalRecipients: string[] = []
-
-		for (const jid of uniqueRecipients) {
-			const canonicalJid = lidByPn.get(jid) || jid
-			if (!canonicalRecipients.includes(canonicalJid)) {
-				canonicalRecipients.push(canonicalJid)
-			}
-		}
-
-		if (canonicalRecipients.length !== uniqueRecipients.length || canonicalRecipients.some((jid, index) => jid !== uniqueRecipients[index])) {
-			logger.debug(
-				{ before: uniqueRecipients, after: canonicalRecipients },
-				'[CAROUSEL] Canonicalized recipient addressing for session reuse'
-			)
-		}
-
-		return canonicalRecipients
-	}
-
 	const relayMessage = async (
 		jid: string,
 		message: proto.IMessage,
@@ -995,13 +1011,10 @@ export const makeMessagesSocket = (config: SocketConfig) => {
 					allRecipients.push(jid)
 				}
 
-				const isCarouselFanout = isCarouselMessage(message)
+
-				const effectiveMeRecipients = isCarouselFanout
+				await assertSessions(allRecipients)
-					? await canonicalizeCarouselRecipients(meRecipients)
+				const effectiveMeRecipients = await canonicalizeSessionRecipients(meRecipients)
-					: meRecipients
+				const effectiveOtherRecipients = await canonicalizeSessionRecipients(otherRecipients)
-				const effectiveOtherRecipients = isCarouselFanout
-					? await canonicalizeCarouselRecipients(otherRecipients)
-					: otherRecipients
 				const effectiveAllRecipients = [...effectiveMeRecipients, ...effectiveOtherRecipients]
 
 				await assertSessions(effectiveAllRecipients)
@@ -1011,14 +1024,14 @@ export const makeMessagesSocket = (config: SocketConfig) => {
 					{ nodes: otherNodes, shouldIncludeDeviceIdentity: s2 }
 				] = await Promise.all([
 					// For own devices: use DSM if available (1:1 chats only)
-					createParticipantNodes(effectiveMeRecipients, meMsg || message, extraAttrs),
+					createParticipantNodes(meRecipients, meMsg || message, extraAttrs),
-					createParticipantNodes(effectiveOtherRecipients, message, extraAttrs, meMsg)
+					createParticipantNodes(otherRecipients, message, extraAttrs, meMsg)
 				])
 				participants.push(...meNodes)
 				participants.push(...otherNodes)
 
-				if (effectiveMeRecipients.length > 0 || effectiveOtherRecipients.length > 0) {
+				if (meRecipients.length > 0 || otherRecipients.length > 0) {
-					extraAttrs['phash'] = generateParticipantHashV2([...effectiveMeRecipients, ...effectiveOtherRecipients])
+					extraAttrs['phash'] = generateParticipantHashV2([...meRecipients, ...otherRecipients])
 				}
 
 				shouldIncludeDeviceIdentity = shouldIncludeDeviceIdentity || s1 || s2