fix: send message speed, lid logic, remove messages-send useless functions (#1794)
* fix: remove redundant migration * fix: remove deduplicatelidpnjids * fix: assertsessions complexity * fix: assertsessions call * fix: remove getencryptionjid * fix: add wirejid to injecte2esession, remove libsignal lid migration * fix: logger lid-mapping * fix: injecte2esession decoded approach * fix: changes made by @AstroX11 * fix: lint * fix: lint * fix: lint * Revert "fix: injecte2esession decoded approach" This reverts commit 4e368296ed084398b8a173ec117dc2478e481748. * fix: centralize getlidforpn calls in messages-send * fix: add resolvechunklids to signal * fix: remove useless arrays * fix: assertsessions logic * fix: lint * Revert "fix: assertsessions logic" This reverts commit 65b0730891c91573edcab1c96af010db54382fba. * fix: remove onwhatsapp fallback to prevent rate limit, migrate sessions when lid-map is created, new migration logic and user devices key * fix: migrate session devices, socket creation * fix: add back decryptionjid validation * fix: add resolveSignalAddress to get lid * fix: type error migration * fix: lint * fix: device level cache, proposed changes
This commit is contained in:
Gustavo Quadri
+136
-107
@@ -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<string, true>({
|
||||
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<string> => {
|
||||
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
|
||||
|
||||
+19
-47
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+142
-336
@@ -84,6 +84,11 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
||||
useClones: false
|
||||
})
|
||||
|
||||
const peerSessionsCache = new NodeCache<boolean>({
|
||||
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<string>()
|
||||
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<Map<string, string>> => {
|
||||
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<string, string>()
|
||||
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<string, string[]>()
|
||||
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<string>()
|
||||
const pnUsersBeingFetched = new Set<string>()
|
||||
|
||||
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<string, string>,
|
||||
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<string, Array<{ recipientJid: string; patchedMessage: proto.IMessage }>>()
|
||||
|
||||
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,
|
||||
|
||||
+19
-6
@@ -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) {
|
||||
|
||||
@@ -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 } }
|
||||
|
||||
+1
-12
@@ -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<void>
|
||||
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<void>
|
||||
}
|
||||
|
||||
@@ -145,6 +145,7 @@ export type SocketConfig = {
|
||||
|
||||
makeSignalRepository: (
|
||||
auth: SignalAuthState,
|
||||
logger: ILogger,
|
||||
onWhatsAppFunc?: (...jids: string[]) => Promise<
|
||||
| {
|
||||
jid: string
|
||||
|
||||
@@ -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<void> => {
|
||||
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 {
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in New Issue
Block a user