diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index a0965f04..ea9cd7bb 100644 --- a/src/Signal/libsignal.ts +++ b/src/Signal/libsignal.ts @@ -1,8 +1,10 @@ /* @ts-ignore */ import * as libsignal from 'libsignal' +import { LRUCache } from 'lru-cache' import type { SignalAuthState, SignalKeyStoreWithTransaction } from '../Types' import type { SignalRepositoryWithLIDStore } from '../Types/Signal' import { generateSignalPubKey } from '../Utils' +import type { ILogger } from '../Utils/logger' import { jidDecode, transferDevice } from '../WABinary' import type { SenderKeyStore } from './Group/group_cipher' import { SenderKeyName } from './Group/sender-key-name' @@ -10,21 +12,16 @@ import { SenderKeyRecord } from './Group/sender-key-record' import { GroupCipher, GroupSessionBuilder, SenderKeyDistributionMessage } from './Group' import { LIDMappingStore } from './lid-mapping' -export function makeLibSignalRepository( - auth: SignalAuthState, - onWhatsAppFunc?: (...jids: string[]) => Promise< - | { - jid: string - exists: boolean - lid: string - }[] - | undefined - > -): SignalRepositoryWithLIDStore { - const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction, onWhatsAppFunc) +export function makeLibSignalRepository(auth: SignalAuthState, logger: ILogger): SignalRepositoryWithLIDStore { + const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction, logger) const storage = signalStorage(auth, lidMapping) const parsedKeys = auth.keys as SignalKeyStoreWithTransaction + const migratedSessionCache = new LRUCache({ + ttl: 7 * 24 * 60 * 60 * 1000, // 7 days + ttlAutopurge: true, + updateAgeOnGet: true + }) function isLikelySyncMessage(addr: libsignal.ProtocolAddress): boolean { const key = addr.toString() @@ -110,34 +107,7 @@ export function makeLibSignalRepository( }, async encryptMessage({ jid, data }) { - // LID SINGLE SOURCE OF TRUTH: Always prefer LID when available - let encryptionJid = jid - - // Check for LID mapping and use it if session exists - if (jid.includes('@s.whatsapp.net')) { - const lidForPN = await lidMapping.getLIDForPN(jid) - if (lidForPN?.includes('@lid')) { - const lidAddr = jidToSignalProtocolAddress(lidForPN) - const { [lidAddr.toString()]: lidSession } = await auth.keys.get('session', [lidAddr.toString()]) - - if (lidSession) { - // LID session exists, use it - encryptionJid = lidForPN - } else { - // Try to migrate if PN session exists - const pnAddr = jidToSignalProtocolAddress(jid) - const { [pnAddr.toString()]: pnSession } = await auth.keys.get('session', [pnAddr.toString()]) - - if (pnSession) { - // Migrate PN to LID - await repository.migrateSession([jid], lidForPN) - encryptionJid = lidForPN - } - } - } - } - - const addr = jidToSignalProtocolAddress(encryptionJid) + const addr = jidToSignalProtocolAddress(jid) const cipher = new libsignal.SessionCipher(storage, addr) // Use transaction to ensure atomicity @@ -218,27 +188,80 @@ export function makeLibSignalRepository( }, async migrateSession( - fromJids: string[], + fromJid: string, toJid: string ): Promise<{ migrated: number; skipped: number; total: number }> { - if (!fromJids.length || !toJid.includes('@lid')) return { migrated: 0, skipped: 0, total: 0 } + if (!fromJid || !toJid.includes('@lid')) return { migrated: 0, skipped: 0, total: 0 } - // Filter valid PN JIDs - const validJids = fromJids.filter(jid => jid.includes('@s.whatsapp.net')) - if (!validJids.length) return { migrated: 0, skipped: 0, total: fromJids.length } + // Only support PN to LID migration + if (!fromJid.includes('@s.whatsapp.net')) { + return { migrated: 0, skipped: 0, total: 1 } + } - // Single optimized transaction for all migrations + const { user } = jidDecode(fromJid)! + + logger.debug({ fromJid }, 'bulk device migration - loading all user devices') + + // Get user's device list from storage + const { [user]: userDevices } = await parsedKeys.get('device-list', [user]) + if (!userDevices) { + return { migrated: 0, skipped: 0, total: 0 } + } + + const { device: fromDevice } = jidDecode(fromJid)! + const fromDeviceStr = fromDevice?.toString() || '0' + if (!userDevices.includes(fromDeviceStr)) { + userDevices.push(fromDeviceStr) + } + + // Filter out cached devices before database fetch + const uncachedDevices = userDevices.filter(device => { + const deviceKey = `${user}.${device}` + return !migratedSessionCache.has(deviceKey) + }) + + // Bulk check session existence only for uncached devices + const deviceSessionKeys = uncachedDevices.map(device => `${user}.${device}`) + const existingSessions = await parsedKeys.get('session', deviceSessionKeys) + + // Step 3: Convert existing sessions to JIDs (only migrate sessions that exist) + const deviceJids: string[] = [] + for (const [sessionKey, sessionData] of Object.entries(existingSessions)) { + if (sessionData) { + // Session exists in storage + const deviceStr = sessionKey.split('.')[1] + if (!deviceStr) continue + const deviceNum = parseInt(deviceStr) + const jid = deviceNum === 0 ? `${user}@s.whatsapp.net` : `${user}:${deviceNum}@s.whatsapp.net` + deviceJids.push(jid) + } + } + + logger.info( + { + fromJid, + totalDevices: userDevices.length, + devicesWithSessions: deviceJids.length, + devices: deviceJids + }, + 'bulk device migration complete - all user devices processed' + ) + + // Single transaction for all migrations return parsedKeys.transaction( async (): Promise<{ migrated: number; skipped: number; total: number }> => { - // 1. Batch store all LID mappings - const mappings = validJids.map(jid => ({ - lid: transferDevice(jid, toJid), - pn: jid - })) - await lidMapping.storeLIDPNMappings(mappings) + // Prepare migration operations with addressing metadata + type MigrationOp = { + fromJid: string + toJid: string + pnUser: string + lidUser: string + deviceId: number + fromAddr: libsignal.ProtocolAddress + toAddr: libsignal.ProtocolAddress + } - // 2. Prepare migration operations - const migrationOps = validJids.map(jid => { + const migrationOps: MigrationOp[] = deviceJids.map(jid => { const lidWithDevice = transferDevice(jid, toJid) const fromDecoded = jidDecode(jid)! const toDecoded = jidDecode(lidWithDevice)! @@ -254,44 +277,53 @@ export function makeLibSignalRepository( } }) - // 3. Batch check which LID sessions already exist - const lidAddrs = migrationOps.map(op => op.toAddr.toString()) - const existingSessions = await auth.keys.get('session', lidAddrs) + const totalOps = migrationOps.length + let migratedCount = 0 - // 4. Filter out sessions that already have LID sessions - const opsToMigrate = migrationOps.filter(op => !existingSessions[op.toAddr.toString()]) - const skippedCount = migrationOps.length - opsToMigrate.length + // Bulk fetch PN sessions - already exist (verified during device discovery) + const pnAddrStrings = Array.from(new Set(migrationOps.map(op => op.fromAddr.toString()))) + const pnSessions = await parsedKeys.get('session', pnAddrStrings) - if (!opsToMigrate.length) { - return { migrated: 0, skipped: skippedCount, total: validJids.length } + // Prepare bulk session updates (PN → LID migration + deletion) + const sessionUpdates: { [key: string]: Uint8Array | null } = {} + + for (const op of migrationOps) { + const pnAddrStr = op.fromAddr.toString() + const lidAddrStr = op.toAddr.toString() + + const pnSession = pnSessions[pnAddrStr] + if (pnSession) { + // Session exists (guaranteed from device discovery) + const fromSession = libsignal.SessionRecord.deserialize(pnSession) + if (fromSession.haveOpenSession()) { + // Queue for bulk update: copy to LID, delete from PN + sessionUpdates[lidAddrStr] = fromSession.serialize() + sessionUpdates[pnAddrStr] = null + + migratedCount++ + } + } } - // 5. Execute all migrations in parallel - await Promise.all( - opsToMigrate.map(async op => { - const fromSession = await storage.loadSession(op.fromAddr.toString()) + // Single bulk session update for all migrations + if (Object.keys(sessionUpdates).length > 0) { + await parsedKeys.set({ session: sessionUpdates }) + logger.debug({ migratedSessions: migratedCount }, 'bulk session migration complete') - if (fromSession?.haveOpenSession()) { - // Copy session to LID address - const sessionBytes = fromSession.serialize() - const copiedSession = libsignal.SessionRecord.deserialize(sessionBytes) - await storage.storeSession(op.toAddr.toString(), copiedSession) - - // Delete PN session - await auth.keys.set({ session: { [op.fromAddr.toString()]: null } }) + // Cache device-level migrations + for (const op of migrationOps) { + if (sessionUpdates[op.toAddr.toString()]) { + const deviceKey = `${op.pnUser}.${op.deviceId}` + migratedSessionCache.set(deviceKey, true) } - }) - ) + } + } - return { migrated: opsToMigrate.length, skipped: skippedCount, total: validJids.length } + const skippedCount = totalOps - migratedCount + return { migrated: migratedCount, skipped: skippedCount, total: totalOps } }, - `migrate-${validJids.length}-sessions-${jidDecode(toJid)?.user}` + `migrate-${deviceJids.length}-sessions-${jidDecode(toJid)?.user}` ) - }, - - async encryptMessageWithWire({ encryptionJid, wireJid, data }) { - const result = await repository.encryptMessage({ jid: encryptionJid, data }) - return { ...result, wireJid } } } @@ -323,32 +355,28 @@ function signalStorage( { creds, keys }: SignalAuthState, lidMapping: LIDMappingStore ): SenderKeyStore & libsignal.SignalStorage { + // Shared function to resolve PN signal address to LID if mapping exists + const resolveSignalAddress = async (id: string): Promise => { + if (id.includes('.') && !id.includes('_1')) { + const parts = id.split('.') + const device = parts[1] || '0' + const pnJid = device === '0' ? `${parts[0]}@s.whatsapp.net` : `${parts[0]}:${device}@s.whatsapp.net` + + const lidForPN = await lidMapping.getLIDForPN(pnJid) + if (lidForPN?.includes('@lid')) { + const lidAddr = jidToSignalProtocolAddress(lidForPN) + return lidAddr.toString() + } + } + + return id + } + return { loadSession: async (id: string) => { try { - // LID SINGLE SOURCE OF TRUTH: Auto-redirect PN to LID if mapping exists - let actualId = id - if (id.includes('.') && !id.includes('_1')) { - // This is a PN signal address format (e.g., "1234567890.0") - // Convert back to JID to check for LID mapping - const parts = id.split('.') - const device = parts[1] || '0' - const pnJid = device === '0' ? `${parts[0]}@s.whatsapp.net` : `${parts[0]}:${device}@s.whatsapp.net` - - const lidForPN = await lidMapping.getLIDForPN(pnJid) - if (lidForPN?.includes('@lid')) { - const lidAddr = jidToSignalProtocolAddress(lidForPN) - const lidId = lidAddr.toString() - - // Check if LID session exists - const { [lidId]: lidSession } = await keys.get('session', [lidId]) - if (lidSession) { - actualId = lidId - } - } - } - - const { [actualId]: sess } = await keys.get('session', [actualId]) + const wireJid = await resolveSignalAddress(id) + const { [wireJid]: sess } = await keys.get('session', [wireJid]) if (sess) { return libsignal.SessionRecord.deserialize(sess) @@ -360,7 +388,8 @@ function signalStorage( return null }, storeSession: async (id: string, session: libsignal.SessionRecord) => { - await keys.set({ session: { [id]: session.serialize() } }) + const wireJid = await resolveSignalAddress(id) + await keys.set({ session: { [wireJid]: session.serialize() } }) }, isTrustedIdentity: () => { return true diff --git a/src/Signal/lid-mapping.ts b/src/Signal/lid-mapping.ts index 5cdaa398..4f1adc53 100644 --- a/src/Signal/lid-mapping.ts +++ b/src/Signal/lid-mapping.ts @@ -1,6 +1,6 @@ import { LRUCache } from 'lru-cache' import type { SignalKeyStoreWithTransaction } from '../Types' -import logger from '../Utils/logger' +import type { ILogger } from '../Utils/logger' import { isLidUser, isPnUser, jidDecode } from '../WABinary' export class LIDMappingStore { @@ -10,28 +10,11 @@ export class LIDMappingStore { updateAgeOnGet: true }) private readonly keys: SignalKeyStoreWithTransaction - private onWhatsAppFunc?: (...jids: string[]) => Promise< - | { - jid: string - exists: boolean - lid: string - }[] - | undefined - > + private readonly logger: ILogger - constructor( - keys: SignalKeyStoreWithTransaction, - onWhatsAppFunc?: (...jids: string[]) => Promise< - | { - jid: string - exists: boolean - lid: string - }[] - | undefined - > - ) { + constructor(keys: SignalKeyStoreWithTransaction, logger: ILogger) { this.keys = keys - this.onWhatsAppFunc = onWhatsAppFunc // needed to get LID from PN if not found + this.logger = logger } /** @@ -42,7 +25,7 @@ export class LIDMappingStore { const pairMap: { [_: string]: string } = {} for (const { lid, pn } of pairs) { if (!((isLidUser(lid) && isPnUser(pn)) || (isPnUser(lid) && isLidUser(pn)))) { - logger.warn(`Invalid LID-PN mapping: ${lid}, ${pn}`) + this.logger.warn(`Invalid LID-PN mapping: ${lid}, ${pn}`) continue } @@ -54,10 +37,9 @@ export class LIDMappingStore { const pnUser = pnDecoded.user const lidUser = lidDecoded.user - // Check if mapping already exists (cache first, then database) let existingLidUser = this.mappingCache.get(`pn:${pnUser}`) if (!existingLidUser) { - // Cache miss - check database + this.logger.trace(`Cache miss for PN user ${pnUser}; checking database`) const stored = await this.keys.get('lid-mapping', [pnUser]) existingLidUser = stored[pnUser] if (existingLidUser) { @@ -68,25 +50,24 @@ export class LIDMappingStore { } if (existingLidUser === lidUser) { - logger.debug({ pnUser, lidUser }, 'LID mapping already exists, skipping') + this.logger.debug({ pnUser, lidUser }, 'LID mapping already exists, skipping') continue } pairMap[pnUser] = lidUser } - logger.trace({ pairMap }, `Storing ${Object.keys(pairMap).length} pn mappings`) + this.logger.trace({ pairMap }, `Storing ${Object.keys(pairMap).length} pn mappings`) await this.keys.transaction(async () => { for (const [pnUser, lidUser] of Object.entries(pairMap)) { await this.keys.set({ 'lid-mapping': { - [pnUser]: lidUser, // "554396160286" -> "102765716062358" - [`${lidUser}_reverse`]: pnUser // "102765716062358_reverse" -> "554396160286" + [pnUser]: lidUser, + [`${lidUser}_reverse`]: pnUser } }) - // Update cache with both directions this.mappingCache.set(`pn:${pnUser}`, lidUser) this.mappingCache.set(`lid:${lidUser}`, pnUser) } @@ -112,26 +93,17 @@ export class LIDMappingStore { lidUser = stored[pnUser] if (lidUser) { - // Cache the database result this.mappingCache.set(`pn:${pnUser}`, lidUser) + this.mappingCache.set(`lid:${lidUser}`, pnUser) } else { - // Not in database - try USync - logger.trace(`No LID mapping found for PN user ${pnUser}; getting from USync`) - const { exists, lid } = (await this.onWhatsAppFunc?.(pn))?.[0]! // this function already adds LIDs to mapping - if (exists && lid) { - lidUser = jidDecode(lid)?.user - if (lidUser) { - // Cache the USync result - this.mappingCache.set(`pn:${pnUser}`, lidUser) - } - } else { - return null - } + this.logger.trace(`No LID mapping found for PN user ${pnUser}`) + return null } } - if (typeof lidUser !== 'string' || !lidUser) { - logger.warn(`Invalid or empty LID user for PN ${pn}: lidUser = "${lidUser}"`) + lidUser = lidUser.toString() + if (!lidUser) { + this.logger.warn(`Invalid or empty LID user for PN ${pn}: lidUser = "${lidUser}"`) return null } @@ -139,7 +111,7 @@ export class LIDMappingStore { const pnDevice = decoded.device !== undefined ? decoded.device : 0 const deviceSpecificLid = `${lidUser}:${pnDevice}@lid` - logger.trace(`getLIDForPN: ${pn} → ${deviceSpecificLid} (user mapping with device ${pnDevice})`) + this.logger.trace(`getLIDForPN: ${pn} → ${deviceSpecificLid} (user mapping with device ${pnDevice})`) return deviceSpecificLid } @@ -162,7 +134,7 @@ export class LIDMappingStore { pnUser = stored[`${lidUser}_reverse`] if (!pnUser || typeof pnUser !== 'string') { - logger.trace(`No reverse mapping found for LID user: ${lidUser}`) + this.logger.trace(`No reverse mapping found for LID user: ${lidUser}`) return null } @@ -173,7 +145,7 @@ export class LIDMappingStore { const lidDevice = decoded.device !== undefined ? decoded.device : 0 const pnJid = `${pnUser}:${lidDevice}@s.whatsapp.net` - logger.trace(`Found reverse mapping: ${lid} → ${pnJid}`) + this.logger.trace(`Found reverse mapping: ${lid} → ${pnJid}`) return pnJid } } diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index b1d227e7..ac58d1d5 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -939,7 +939,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { } } - await assertSessions([participant], shouldRecreateSession) + await assertSessions([participant]) if (isJidGroup(remoteJid)) { await authState.keys.set({ 'sender-key-memory': { [remoteJid]: null } }) @@ -1130,23 +1130,23 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { pn = jidNormalizedUser(node.attrs.sender_pn) ev.emit('lid-mapping.update', { lid, pn }) await signalRepository.lidMapping.storeLIDPNMappings([{ lid, pn }]) + await signalRepository.migrateSession(pn, lid) } const alt = msg.key.participantAlt || msg.key.remoteJidAlt // store new mappings we didn't have before if (!!alt) { const altServer = jidDecode(alt)?.server + const primaryJid = msg.key.participant || msg.key.remoteJid! if (altServer === 'lid') { if (typeof (await signalRepository.lidMapping.getPNForLID(alt)) === 'string') { - await signalRepository.lidMapping.storeLIDPNMappings([ - { lid: alt, pn: msg.key.participant || msg.key.remoteJid! } - ]) + await signalRepository.lidMapping.storeLIDPNMappings([{ lid: alt, pn: primaryJid }]) + await signalRepository.migrateSession(primaryJid, alt) } } else { if (typeof (await signalRepository.lidMapping.getLIDForPN(alt)) === 'string') { - await signalRepository.lidMapping.storeLIDPNMappings([ - { lid: msg.key.participant || msg.key.remoteJid!, pn: alt } - ]) + await signalRepository.lidMapping.storeLIDPNMappings([{ lid: primaryJid, pn: alt }]) + await signalRepository.migrateSession(alt, primaryJid) } } } diff --git a/src/Socket/messages-send.ts b/src/Socket/messages-send.ts index 0d6abfc9..b9848c78 100644 --- a/src/Socket/messages-send.ts +++ b/src/Socket/messages-send.ts @@ -84,6 +84,11 @@ export const makeMessagesSocket = (config: SocketConfig) => { useClones: false }) + const peerSessionsCache = new NodeCache({ + stdTTL: DEFAULT_CACHE_TTLS.USER_DEVICES, + useClones: false + }) + // Initialize message retry manager if enabled const messageRetryManager = enableRecentMessageCache ? new MessageRetryManager(logger, maxMsgRetryCount) : null @@ -197,39 +202,35 @@ export const makeMessagesSocket = (config: SocketConfig) => { /** Device info with wire JID format for envelope addressing */ type DeviceWithWireJid = JidWithDevice & { - wireJid: string // The exact JID format that should be used in wire protocol (envelope addressing) + wireJid: string } - /** - * Deduplicate JIDs when both LID and PN versions exist for same user - * Prefers LID over PN to maintain single encryption layer - */ - const deduplicateLidPnJids = (jids: string[]): string[] => { - const lidUsers = new Set() - const filteredJids: string[] = [] - - // Collect all LID users - for (const jid of jids) { - if (jid.includes('@lid')) { - const user = jidDecode(jid)?.user - if (user) lidUsers.add(user) - } + const resolveSessionJids = async (jids: string[]): Promise> => { + const uniquePnJids = Array.from(new Set(jids.filter(isPnUser))) + if (!uniquePnJids.length) { + return new Map() } - // Filter out PN versions when LID exists - for (const jid of jids) { - if (jid.includes('@s.whatsapp.net')) { - const user = jidDecode(jid)?.user - if (user && lidUsers.has(user)) { - logger.debug({ jid }, 'Skipping PN - LID version exists') - continue + const lookups = await Promise.all( + uniquePnJids.map(async pnJid => { + try { + const resolved = await signalRepository.lidMapping.getLIDForPN(pnJid) + return resolved ? ([pnJid, resolved] as const) : [pnJid, pnJid] + } catch (error) { + logger.warn({ pnJid, error }, 'Failed to resolve LID mapping for PN JID') + return [pnJid, pnJid] } - } + }) + ) - filteredJids.push(jid) + const sessionMap = new Map() + for (const entry of lookups) { + if (entry) { + sessionMap.set(entry[0], entry[1]) + } } - return filteredJids + return sessionMap } /** Fetch all the devices we've to send a message to */ @@ -245,7 +246,7 @@ export const makeMessagesSocket = (config: SocketConfig) => { } const toFetch: string[] = [] - jids = deduplicateLidPnJids(Array.from(new Set(jids))) + const jidsWithUser = jids .map(jid => { const decoded = jidDecode(jid) @@ -360,170 +361,57 @@ export const makeMessagesSocket = (config: SocketConfig) => { if (deviceMap[key]) await userDevicesCache.set(key, deviceMap[key]) } } + + const userDeviceUpdates: { [userId: string]: string[] } = {} + for (const [userId, devices] of Object.entries(deviceMap)) { + if (devices && devices.length > 0) { + userDeviceUpdates[userId] = devices.map(d => d.device?.toString() || '0') + } + } + + if (Object.keys(userDeviceUpdates).length > 0) { + try { + await authState.keys.set({ 'device-list': userDeviceUpdates }) + logger.debug( + { userCount: Object.keys(userDeviceUpdates).length }, + 'stored user device lists for bulk migration' + ) + } catch (error) { + logger.warn({ error }, 'failed to store user device lists') + } + } } return deviceResults } - const assertSessions = async (jids: string[], force: boolean) => { + const assertSessions = async (jids: string[]) => { let didFetchNewSession = false + const uniqueJids = [...new Set(jids)] // Deduplicate JIDs const jidsRequiringFetch: string[] = [] - // Apply same deduplication as in getUSyncDevices - jids = deduplicateLidPnJids(jids) - - if (force) { - // Check which sessions are missing (with LID migration check) - const addrs = jids.map(jid => signalRepository.jidToSignalProtocolAddress(jid)) - const sessions = await authState.keys.get('session', addrs) - - // Simplified: Check session existence directly - const checkJidSession = (jid: string) => { - const signalId = signalRepository.jidToSignalProtocolAddress(jid) + // Check peerSessionsCache and authState.keys + for (const jid of uniqueJids) { + const signalId = signalRepository.jidToSignalProtocolAddress(jid) + const cachedSession = peerSessionsCache.get(signalId) + if (cachedSession !== undefined) { + if (cachedSession) { + continue // Session exists in cache + } + } else { + const sessions = await authState.keys.get('session', [signalId]) const hasSession = !!sessions[signalId] - - // Add to fetch list if no session exists - // Session type selection (LID vs PN) is handled in encryptMessage - if (!hasSession) { - if (jid.includes('@lid')) { - logger.debug({ jid }, 'No LID session found, will create new LID session') - } - - jidsRequiringFetch.push(jid) + peerSessionsCache.set(signalId, hasSession) + if (hasSession) { + continue } } - // Process all JIDs - for (const jid of jids) { - checkJidSession(jid) - } - } else { - const addrs = jids.map(jid => signalRepository.jidToSignalProtocolAddress(jid)) - const sessions = await authState.keys.get('session', addrs) - - // Group JIDs by user for bulk migration - const userGroups = new Map() - for (const jid of jids) { - const user = jidNormalizedUser(jid) - if (!userGroups.has(user)) { - userGroups.set(user, []) - } - - userGroups.get(user)!.push(jid) - } - - // Helper to check LID mapping for a user - const checkUserLidMapping = async (user: string, userJids: string[]) => { - if (!userJids.some(jid => jid.includes('@s.whatsapp.net'))) { - return { shouldMigrate: false, lidForPN: undefined } - } - - try { - // Convert user to proper PN JID format for getLIDForPN - const pnJid = `${user}@s.whatsapp.net` - const mapping = await signalRepository.lidMapping.getLIDForPN(pnJid) - if (mapping?.includes('@lid')) { - logger.debug( - { user, lidForPN: mapping, deviceCount: userJids.length }, - 'User has LID mapping - preparing bulk migration' - ) - return { shouldMigrate: true, lidForPN: mapping } - } - } catch (error) { - logger.debug({ user, error }, 'Failed to check LID mapping for user') - } - - return { shouldMigrate: false, lidForPN: undefined } - } - - // Process each user group for potential bulk LID migration - for (const [user, userJids] of userGroups) { - const mappingResult = await checkUserLidMapping(user, userJids) - const shouldMigrateUser = mappingResult.shouldMigrate - const lidForPN = mappingResult.lidForPN - - // Migrate all devices for this user if LID mapping exists - if (shouldMigrateUser && lidForPN) { - // Bulk migrate all user devices in single transaction - const migrationResult = await signalRepository.migrateSession(userJids, lidForPN) - - if (migrationResult.migrated > 0) { - logger.info( - { - user, - lidMapping: lidForPN, - migrated: migrationResult.migrated, - skipped: migrationResult.skipped, - total: migrationResult.total - }, - 'Completed bulk migration for user devices' - ) - } else { - logger.debug( - { - user, - lidMapping: lidForPN, - skipped: migrationResult.skipped, - total: migrationResult.total - }, - 'All user device sessions already migrated' - ) - } - } - - // Direct bulk session check with LID single source of truth - const addMissingSessionsToFetchList = (jid: string) => { - const signalId = signalRepository.jidToSignalProtocolAddress(jid) - if (sessions[signalId]) return - - // Determine correct JID to fetch (LID if mapping exists, otherwise original) - if (jid.includes('@s.whatsapp.net') && shouldMigrateUser && lidForPN) { - const decoded = jidDecode(jid)! - const lidDeviceJid = - decoded.device !== undefined ? `${jidDecode(lidForPN)!.user}:${decoded.device}@lid` : lidForPN - jidsRequiringFetch.push(lidDeviceJid) - logger.debug({ pnJid: jid, lidJid: lidDeviceJid }, 'Adding LID JID to fetch list (conversion)') - } else { - jidsRequiringFetch.push(jid) - logger.debug({ jid }, 'Adding JID to fetch list') - } - } - - userJids.forEach(addMissingSessionsToFetchList) - } + jidsRequiringFetch.push(jid) } if (jidsRequiringFetch.length) { logger.debug({ jidsRequiringFetch }, 'fetching sessions') - - // DEBUG: Check if there are PN versions of LID users being fetched - const lidUsersBeingFetched = new Set() - const pnUsersBeingFetched = new Set() - - for (const jid of jidsRequiringFetch) { - const user = jidDecode(jid)?.user - if (user) { - if (jid.includes('@lid')) { - lidUsersBeingFetched.add(user) - } else if (jid.includes('@s.whatsapp.net')) { - pnUsersBeingFetched.add(user) - } - } - } - - // Find overlaps - const overlapping = Array.from(pnUsersBeingFetched).filter(user => lidUsersBeingFetched.has(user)) - if (overlapping.length > 0) { - logger.warn( - { - overlapping, - lidUsersBeingFetched: Array.from(lidUsersBeingFetched), - pnUsersBeingFetched: Array.from(pnUsersBeingFetched) - }, - 'Fetching both LID and PN sessions for same users' - ) - } - const result = await query({ tag: 'iq', attrs: { @@ -543,8 +431,13 @@ export const makeMessagesSocket = (config: SocketConfig) => { ] }) await parseAndInjectE2ESessions(result, signalRepository) - didFetchNewSession = true + + // Cache fetched sessions + for (const jid of jidsRequiringFetch) { + const signalId = signalRepository.jidToSignalProtocolAddress(jid) + peerSessionsCache.set(signalId, true) + } } return didFetchNewSession @@ -578,133 +471,58 @@ export const makeMessagesSocket = (config: SocketConfig) => { return msgId } - const createParticipantNodes = async ( - jids: string[], + const createParticipantNodesWithSessionMap = async ( + recipientWireJids: string[], + sessionMap: Map, message: proto.IMessage, extraAttrs?: BinaryNode['attrs'], dsmMessage?: proto.IMessage ) => { - let patched = await patchMessageBeforeSending(message, jids) - if (!Array.isArray(patched)) { - patched = jids ? jids.map(jid => ({ recipientJid: jid, ...patched })) : [patched] + if (!recipientWireJids.length) { + return { nodes: [] as BinaryNode[], shouldIncludeDeviceIdentity: false } } - let shouldIncludeDeviceIdentity = false + const patched = await patchMessageBeforeSending(message, recipientWireJids) + const patchedMessages = Array.isArray(patched) + ? patched + : recipientWireJids.map(jid => ({ recipientJid: jid, message: patched })) + let shouldIncludeDeviceIdentity = false const meId = authState.creds.me!.id const meLid = authState.creds.me?.lid const meLidUser = meLid ? jidDecode(meLid)?.user : null - const devicesByUser = new Map>() - - for (const patchedMessageWithJid of patched) { - const { recipientJid: wireJid, ...patchedMessage } = patchedMessageWithJid - if (!wireJid) continue - - // Extract user from JID for grouping - const decoded = jidDecode(wireJid) - const user = decoded?.user - - if (!user) continue - - if (!devicesByUser.has(user)) { - devicesByUser.set(user, []) - } - - devicesByUser.get(user)!.push({ recipientJid: wireJid, patchedMessage }) - } - - // Process each user's devices sequentially, but different users in parallel - const userEncryptionPromises = Array.from(devicesByUser.entries()).map(([user, userDevices]) => - encryptionMutex.mutex(user, async () => { - logger.debug({ user, deviceCount: userDevices.length }, 'Acquiring encryption lock for user devices') - - const userNodes: BinaryNode[] = [] - - // Helper to get encryption JID with LID migration - const getEncryptionJid = async (wireJid: string) => { - if (!wireJid.includes('@s.whatsapp.net')) return wireJid - - try { - const lidForPN = await signalRepository.lidMapping.getLIDForPN(wireJid) - - if (!lidForPN?.includes('@lid')) return wireJid - - // Preserve device ID from original wire JID - const wireDecoded = jidDecode(wireJid) - const deviceId = wireDecoded?.device || 0 - const lidDecoded = jidDecode(lidForPN) - const lidWithDevice = jidEncode(lidDecoded?.user!, 'lid', deviceId) - - // Migrate session to LID for unified encryption layer - try { - const migrationResult = await signalRepository.migrateSession([wireJid], lidWithDevice) - const recipientUser = jidNormalizedUser(wireJid) - const ownPnUser = jidNormalizedUser(meId) - const isOwnDevice = recipientUser === ownPnUser - logger.info({ wireJid, lidWithDevice, isOwnDevice }, 'Migrated to LID encryption') - - // Delete PN session after successful migration - try { - if (migrationResult.migrated) { - await signalRepository.deleteSession([wireJid]) - logger.debug({ deletedPNSession: wireJid }, 'Deleted PN session') - } - } catch (deleteError) { - logger.warn({ wireJid, error: deleteError }, 'Failed to delete PN session') - } - - return lidWithDevice - } catch (migrationError) { - logger.warn({ wireJid, error: migrationError }, 'Failed to migrate session') - return wireJid - } - } catch (error) { - logger.debug({ wireJid, error }, 'Failed to check LID mapping') - return wireJid + const encryptionPromises = (patchedMessages as any).map( + async ({ recipientJid: wireJid, message: patchedMessage }: any) => { + if (!wireJid) return null + wireJid = sessionMap.get(wireJid) ?? wireJid + let msgToEncrypt = patchedMessage + if (dsmMessage) { + const { user: targetUser } = jidDecode(wireJid)! + const { user: ownPnUser } = jidDecode(meId)! + const ownLidUser = meLidUser + const isOwnUser = targetUser === ownPnUser || (ownLidUser && targetUser === ownLidUser) + const isExactSenderDevice = wireJid === meId || (meLid && wireJid === meLid) + if (isOwnUser && !isExactSenderDevice) { + msgToEncrypt = dsmMessage + logger.debug({ wireJid, targetUser }, 'Using DSM for own device') } } - // Encrypt to this user's devices sequentially to prevent session corruption - for (const { recipientJid: wireJid, patchedMessage } of userDevices) { - // DSM logic: Use DSM for own other devices (following whatsmeow implementation) - let messageToEncrypt = patchedMessage - if (dsmMessage) { - const { user: targetUser } = jidDecode(wireJid)! - const { user: ownPnUser } = jidDecode(meId)! - const ownLidUser = meLidUser - - // Check if this is our device (same user, different device) - const isOwnUser = targetUser === ownPnUser || (ownLidUser && targetUser === ownLidUser) - - // Exclude exact sender device (whatsmeow: if jid == ownJID || jid == ownLID { continue }) - const isExactSenderDevice = - wireJid === meId || (authState.creds.me?.lid && wireJid === authState.creds.me.lid) - - if (isOwnUser && !isExactSenderDevice) { - messageToEncrypt = dsmMessage - logger.debug({ wireJid, targetUser }, 'Using DSM for own device') - } - } - - const bytes = encodeWAMessage(messageToEncrypt) - - // Get encryption JID with LID migration - const encryptionJid = await getEncryptionJid(wireJid) - - // ENCRYPT: Use the determined encryption identity (prefers migrated LID) + const bytes = encodeWAMessage(msgToEncrypt) + const mutexKey = wireJid + const node = await encryptionMutex.mutex(mutexKey, async () => { const { type, ciphertext } = await signalRepository.encryptMessage({ - jid: encryptionJid, // Unified encryption layer (LID when available) + jid: wireJid, data: bytes }) - if (type === 'pkmsg') { shouldIncludeDeviceIdentity = true } - const node: BinaryNode = { + return { tag: 'to', - attrs: { jid: wireJid }, // Always use original wire identity in envelope + attrs: { jid: wireJid }, content: [ { tag: 'enc', @@ -717,20 +535,25 @@ export const makeMessagesSocket = (config: SocketConfig) => { } ] } - userNodes.push(node) - } - - logger.debug({ user, nodesCreated: userNodes.length }, 'Releasing encryption lock for user devices') - return userNodes - }) + }) + return node + } ) - // Wait for all users to complete (users are processed in parallel) - const userNodesArrays = await Promise.all(userEncryptionPromises) - const nodes = userNodesArrays.flat() + const nodes = (await Promise.all(encryptionPromises)).filter(node => node !== null) as BinaryNode[] return { nodes, shouldIncludeDeviceIdentity } } + const createParticipantNodes = async ( + recipientWireJids: string[], + message: proto.IMessage, + extraAttrs?: BinaryNode['attrs'], + dsmMessage?: proto.IMessage + ) => { + const sessionMap = await resolveSessionJids(recipientWireJids) + return createParticipantNodesWithSessionMap(recipientWireJids, sessionMap, message, extraAttrs, dsmMessage) + } + const relayMessage = async ( jid: string, message: proto.IMessage, @@ -746,19 +569,14 @@ export const makeMessagesSocket = (config: SocketConfig) => { ) => { const meId = authState.creds.me!.id const meLid = authState.creds.me?.lid - - // ADDRESSING CONSISTENCY: Keep envelope addressing as user provided, handle LID migration in encryption - let shouldIncludeDeviceIdentity = false + const statusJid = 'status@broadcast' const { user, server } = jidDecode(jid)! - const statusJid = 'status@broadcast' const isGroup = server === 'g.us' const isStatus = jid === statusJid const isLid = server === 'lid' const isNewsletter = server === 'newsletter' - - // Keep user's original JID choice for envelope addressing const finalJid = jid // ADDRESSING CONSISTENCY: Match own identity to conversation context @@ -777,7 +595,6 @@ export const makeMessagesSocket = (config: SocketConfig) => { const participants: BinaryNode[] = [] const destinationJid = !isStatus ? finalJid : statusJid - const binaryNodeContent: BinaryNode[] = [] const devices: DeviceWithWireJid[] = [] @@ -792,18 +609,15 @@ export const makeMessagesSocket = (config: SocketConfig) => { const extraAttrs: BinaryNodeAttributes = {} if (participant) { - // when the retry request is not for a group - // only send to the specific device that asked for a retry - // otherwise the message is sent out to every device that should be a recipient if (!isGroup && !isStatus) { additionalAttributes = { ...additionalAttributes, device_fanout: 'false' } } - const { user, device } = jidDecode(participant.jid)! // rajeh: how does this even make sense TODO check out + const { user, device } = jidDecode(participant.jid)! devices.push({ user, device, - wireJid: participant.jid // Use the participant JID as wire JID + wireJid: participant.jid }) } @@ -814,7 +628,6 @@ export const makeMessagesSocket = (config: SocketConfig) => { } if (isNewsletter) { - // Patch message if needed, then encode as plaintext const patched = patchMessageBeforeSending ? await patchMessageBeforeSending(message, []) : message const bytes = encodeNewsletterMessage(patched as proto.IMessage) binaryNodeContent.push({ @@ -883,14 +696,11 @@ export const makeMessagesSocket = (config: SocketConfig) => { } const patched = await patchMessageBeforeSending(message) - if (Array.isArray(patched)) { throw new Boom('Per-jid patching is not supported in groups') } const bytes = encodeWAMessage(patched) - - // This should match the group's addressing mode and conversation context const groupAddressingMode = groupData?.addressingMode || (isLid ? 'lid' : 'pn') const groupSenderIdentity = groupAddressingMode === 'lid' && meLid ? meLid : meId @@ -900,23 +710,19 @@ export const makeMessagesSocket = (config: SocketConfig) => { meId: groupSenderIdentity }) - const senderKeyJids: string[] = [] - // ensure a connection is established with every device + const deviceSessionMap = await resolveSessionJids(devices.map(d => d.wireJid)) + const senderKeyRecipients: string[] = [] for (const device of devices) { - // This preserves the LID migration results from getUSyncDevices const deviceJid = device.wireJid const hasKey = !!senderKeyMap[deviceJid] if (!hasKey || !!participant) { - senderKeyJids.push(deviceJid) - // store that this person has had the sender keys sent to them + senderKeyRecipients.push(deviceJid) senderKeyMap[deviceJid] = true } } - // if there are some participants with whom the session has not been established - // if there are, we re-send the senderkey - if (senderKeyJids.length) { - logger.debug({ senderKeyJids }, 'sending new sender key') + if (senderKeyRecipients.length) { + logger.debug({ senderKeyJids: senderKeyRecipients }, 'sending new sender key') const senderKeyMsg: proto.IMessage = { senderKeyDistributionMessage: { @@ -925,9 +731,15 @@ export const makeMessagesSocket = (config: SocketConfig) => { } } - await assertSessions(senderKeyJids, false) + const senderKeySessionTargets = senderKeyRecipients.map(jid => deviceSessionMap.get(jid) ?? jid) + await assertSessions(senderKeySessionTargets) - const result = await createParticipantNodes(senderKeyJids, senderKeyMsg, extraAttrs) + const result = await createParticipantNodesWithSessionMap( + senderKeyRecipients, + deviceSessionMap, + senderKeyMsg, + extraAttrs + ) shouldIncludeDeviceIdentity = shouldIncludeDeviceIdentity || result.shouldIncludeDeviceIdentity participants.push(...result.nodes) @@ -974,7 +786,7 @@ export const makeMessagesSocket = (config: SocketConfig) => { : jidEncode(jidDecode(meId)?.user!, 's.whatsapp.net', undefined) // Enumerate devices for sender and target with consistent addressing - const sessionDevices = await getUSyncDevices([senderIdentity, jid], false, false) + const sessionDevices = await getUSyncDevices([senderIdentity, jid], true, false) devices.push(...sessionDevices) logger.debug( @@ -987,9 +799,9 @@ export const makeMessagesSocket = (config: SocketConfig) => { } } - const allJids: string[] = [] - const meJids: string[] = [] - const otherJids: string[] = [] + const allRecipients: string[] = [] + const meRecipients: string[] = [] + const otherRecipients: string[] = [] const { user: mePnUser } = jidDecode(meId)! const { user: meLidUser } = meLid ? jidDecode(meLid)! : { user: null } @@ -1003,32 +815,32 @@ export const makeMessagesSocket = (config: SocketConfig) => { // Check if this is our device (could match either PN or LID user) const isMe = user === mePnUser || (meLidUser && user === meLidUser) - const jid = wireJid - if (isMe) { - meJids.push(jid) + meRecipients.push(wireJid) } else { - otherJids.push(jid) + otherRecipients.push(wireJid) } - allJids.push(jid) + allRecipients.push(wireJid) } - await assertSessions([...otherJids, ...meJids], false) + const deviceSessionMap = await resolveSessionJids(devices.map(d => d.wireJid)) + const sessionTargets = allRecipients.map(jid => deviceSessionMap.get(jid) ?? jid) + await assertSessions(sessionTargets) const [ { nodes: meNodes, shouldIncludeDeviceIdentity: s1 }, { nodes: otherNodes, shouldIncludeDeviceIdentity: s2 } ] = await Promise.all([ // For own devices: use DSM if available (1:1 chats only) - createParticipantNodes(meJids, meMsg || message, extraAttrs), - createParticipantNodes(otherJids, message, extraAttrs, meMsg) + createParticipantNodesWithSessionMap(meRecipients, deviceSessionMap, meMsg || message, extraAttrs), + createParticipantNodesWithSessionMap(otherRecipients, deviceSessionMap, message, extraAttrs, meMsg) ]) participants.push(...meNodes) participants.push(...otherNodes) - if (meJids.length > 0 || otherJids.length > 0) { - extraAttrs['phash'] = generateParticipantHashV2([...meJids, ...otherJids]) + if (meRecipients.length > 0 || otherRecipients.length > 0) { + extraAttrs['phash'] = generateParticipantHashV2([...meRecipients, ...otherRecipients]) } shouldIncludeDeviceIdentity = shouldIncludeDeviceIdentity || s1 || s2 @@ -1325,12 +1137,6 @@ export const makeMessagesSocket = (config: SocketConfig) => { } as BinaryNode) } - if ('cachedGroupMetadata' in options) { - console.warn( - 'cachedGroupMetadata in sendMessage are deprecated, now cachedGroupMetadata is part of the socket config.' - ) - } - await relayMessage(jid, fullMsg.message!, { messageId: fullMsg.key.id!, useCachedGroupMetadata: options.useCachedGroupMetadata, diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index bf45d3ab..dfa9303d 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -42,6 +42,7 @@ import { getBinaryNodeChild, getBinaryNodeChildren, isLidUser, + jidDecode, jidEncode, S_WHATSAPP_NET } from '../WABinary' @@ -272,9 +273,13 @@ export const makeSocket = (config: SocketConfig) => { const results = await executeUSyncQuery(usyncQuery) if (results) { - if (results.list.filter(a => !!a.lid).length > 0) { - const lidOnly = results.list.filter(a => !!a.lid) - await signalRepository.lidMapping.storeLIDPNMappings(lidOnly.map(a => ({ pn: a.id, lid: a.lid as string }))) + const lidOnly = results.list.filter(a => !!a.lid) + if (lidOnly.length > 0) { + const pairs = lidOnly.map(a => ({ pn: a.id, lid: a.lid as string })) + await signalRepository.lidMapping.storeLIDPNMappings(pairs) + for (const { pn, lid } of pairs) { + await signalRepository.migrateSession(pn, lid) + } } return results.list @@ -288,7 +293,7 @@ export const makeSocket = (config: SocketConfig) => { const { creds } = authState // add transaction capability const keys = addTransactionCapability(authState.keys, logger, transactionOpts) - const signalRepository = makeSignalRepository({ creds, keys }, onWhatsApp) + const signalRepository = makeSignalRepository({ creds, keys }, logger, onWhatsApp) let lastDateRecv: Date let epoch = 1 @@ -853,8 +858,16 @@ export const makeSocket = (config: SocketConfig) => { // Store our own LID-PN mapping await signalRepository.lidMapping.storeLIDPNMappings([{ lid: myLID, pn: myPN }]) - // Create LID session for ourselves (whatsmeow pattern) - await signalRepository.migrateSession([myPN], myLID) + // Create device list for our own user (needed for bulk migration) + const { user, device } = jidDecode(myPN)! + await authState.keys.set({ + 'device-list': { + [user]: [device?.toString() || '0'] + } + }) + + // migrate our own session + await signalRepository.migrateSession(myPN, myLID) logger.info({ myPN, myLID }, 'Own LID session created successfully') } catch (error) { diff --git a/src/Types/Auth.ts b/src/Types/Auth.ts index a8918bdd..f47c691f 100644 --- a/src/Types/Auth.ts +++ b/src/Types/Auth.ts @@ -73,6 +73,7 @@ export type SignalDataTypeMap = { 'app-state-sync-key': proto.Message.IAppStateSyncKeyData 'app-state-sync-version': LTHashState 'lid-mapping': string + 'device-list': string[] } export type SignalDataSet = { [T in keyof SignalDataTypeMap]?: { [id: string]: SignalDataTypeMap[T] | null } } diff --git a/src/Types/Signal.ts b/src/Types/Signal.ts index 52cadcf0..e46cd425 100644 --- a/src/Types/Signal.ts +++ b/src/Types/Signal.ts @@ -23,12 +23,6 @@ type EncryptMessageOpts = { data: Uint8Array } -type EncryptMessageWithWireOpts = { - encryptionJid: string // JID used for session lookup (LID) - wireJid: string // JID used for envelope (PN) - data: Uint8Array -} - type EncryptGroupMessageOpts = { group: string data: Uint8Array @@ -64,11 +58,6 @@ export type SignalRepository = { type: 'pkmsg' | 'msg' ciphertext: Uint8Array }> - encryptMessageWithWire(opts: EncryptMessageWithWireOpts): Promise<{ - type: 'pkmsg' | 'msg' - ciphertext: Uint8Array - wireJid: string // Return the wire JID for envelope - }> encryptGroupMessage(opts: EncryptGroupMessageOpts): Promise<{ senderKeyDistributionMessage: Uint8Array ciphertext: Uint8Array @@ -76,7 +65,7 @@ export type SignalRepository = { injectE2ESession(opts: E2ESessionOpts): Promise validateSession(jid: string): Promise<{ exists: boolean; reason?: string }> jidToSignalProtocolAddress(jid: string): string - migrateSession(fromJids: string[], toJid: string): Promise<{ migrated: number; skipped: number; total: number }> + migrateSession(fromJid: string, toJid: string): Promise<{ migrated: number; skipped: number; total: number }> validateSession(jid: string): Promise<{ exists: boolean; reason?: string }> deleteSession(jids: string[]): Promise } diff --git a/src/Types/Socket.ts b/src/Types/Socket.ts index 3a14100f..bafbb3c8 100644 --- a/src/Types/Socket.ts +++ b/src/Types/Socket.ts @@ -145,6 +145,7 @@ export type SocketConfig = { makeSignalRepository: ( auth: SignalAuthState, + logger: ILogger, onWhatsAppFunc?: (...jids: string[]) => Promise< | { jid: string diff --git a/src/Utils/decode-wa-message.ts b/src/Utils/decode-wa-message.ts index 22647483..10306ade 100644 --- a/src/Utils/decode-wa-message.ts +++ b/src/Utils/decode-wa-message.ts @@ -23,14 +23,15 @@ const getDecryptionJid = async (sender: string, repository: SignalRepositoryWith return sender } - return (await repository.lidMapping.getLIDForPN(sender))! + const mapped = await repository.lidMapping.getLIDForPN(sender) + return mapped || sender } const storeMappingFromEnvelope = async ( stanza: BinaryNode, sender: string, - decryptionJid: string, repository: SignalRepositoryWithLIDStore, + decryptionJid: string, logger: ILogger ): Promise => { const { senderAlt } = extractAddressingContext(stanza) @@ -38,6 +39,7 @@ const storeMappingFromEnvelope = async ( if (senderAlt && isLidUser(senderAlt) && isPnUser(sender) && decryptionJid === sender) { try { await repository.lidMapping.storeLIDPNMappings([{ lid: senderAlt, pn: sender }]) + await repository.migrateSession(sender, senderAlt) logger.debug({ sender, senderAlt }, 'Stored LID mapping from envelope') } catch (error) { logger.warn({ sender, senderAlt, error }, 'Failed to store LID mapping') @@ -243,9 +245,11 @@ export const decryptMessageNode = ( let msgBuffer: Uint8Array const user = isPnUser(sender) ? sender : author // TODO: flaky logic + const decryptionJid = await getDecryptionJid(user, repository) + if (tag !== 'plaintext') { - await storeMappingFromEnvelope(stanza, user, decryptionJid, repository, logger) + await storeMappingFromEnvelope(stanza, user, repository, decryptionJid, logger) } try { diff --git a/src/Utils/process-message.ts b/src/Utils/process-message.ts index e10d4a70..4d8ab548 100644 --- a/src/Utils/process-message.ts +++ b/src/Utils/process-message.ts @@ -318,6 +318,11 @@ const processMessage = async ( } await signalRepository.lidMapping.storeLIDPNMappings(pairs) + if (pairs.length) { + for (const { pn, lid } of pairs) { + await signalRepository.migrateSession(pn, lid) + } + } } } else if (content?.reactionMessage) { const reaction: proto.IReaction = { diff --git a/src/Utils/signal.ts b/src/Utils/signal.ts index a755ae25..aadb1e4f 100644 --- a/src/Utils/signal.ts +++ b/src/Utils/signal.ts @@ -106,6 +106,7 @@ export const parseAndInjectE2ESessions = async (node: BinaryNode, repository: Si // It's rare case when you need to E2E sessions for so many users, but it's possible const chunkSize = 100 const chunks = chunk(nodes, chunkSize) + for (const nodesChunk of chunks) { await Promise.all( nodesChunk.map(async (node: BinaryNode) => { @@ -113,6 +114,7 @@ export const parseAndInjectE2ESessions = async (node: BinaryNode, repository: Si const key = getBinaryNodeChild(node, 'key')! const identity = getBinaryNodeChildBuffer(node, 'identity')! const jid = node.attrs.jid! + const registrationId = getBinaryNodeChildUInt(node, 'registration', 4) await repository.injectE2ESession({