perf: reduce DB calls during sync with caching and batching (#2316)
* perf: reduce DB calls during sync with caching and batching * refactor: clean up comments and improve LID-PN mapping storage during history sync
This commit is contained in:
+93
-33
@@ -14,6 +14,9 @@ export class LIDMappingStore {
|
|||||||
|
|
||||||
private pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
|
private pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
|
||||||
|
|
||||||
|
private readonly inflightLIDLookups = new Map<string, Promise<LIDMapping[] | null>>()
|
||||||
|
private readonly inflightPNLookups = new Map<string, Promise<LIDMapping[] | null>>()
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
keys: SignalKeyStoreWithTransaction,
|
keys: SignalKeyStoreWithTransaction,
|
||||||
logger: ILogger,
|
logger: ILogger,
|
||||||
@@ -24,12 +27,10 @@ export class LIDMappingStore {
|
|||||||
this.logger = logger
|
this.logger = logger
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Store LID-PN mapping - USER LEVEL
|
|
||||||
*/
|
|
||||||
async storeLIDPNMappings(pairs: LIDMapping[]): Promise<void> {
|
async storeLIDPNMappings(pairs: LIDMapping[]): Promise<void> {
|
||||||
// Validate inputs
|
if (pairs.length === 0) return
|
||||||
const pairMap: { [_: string]: string } = {}
|
|
||||||
|
const validatedPairs: Array<{ pnUser: string; lidUser: string }> = []
|
||||||
for (const { lid, pn } of pairs) {
|
for (const { lid, pn } of pairs) {
|
||||||
if (!((isLidUser(lid) && isPnUser(pn)) || (isPnUser(lid) && isLidUser(pn)))) {
|
if (!((isLidUser(lid) && isPnUser(pn)) || (isPnUser(lid) && isLidUser(pn)))) {
|
||||||
this.logger.warn(`Invalid LID-PN mapping: ${lid}, ${pn}`)
|
this.logger.warn(`Invalid LID-PN mapping: ${lid}, ${pn}`)
|
||||||
@@ -38,24 +39,43 @@ export class LIDMappingStore {
|
|||||||
|
|
||||||
const lidDecoded = jidDecode(lid)
|
const lidDecoded = jidDecode(lid)
|
||||||
const pnDecoded = jidDecode(pn)
|
const pnDecoded = jidDecode(pn)
|
||||||
|
|
||||||
if (!lidDecoded || !pnDecoded) continue
|
if (!lidDecoded || !pnDecoded) continue
|
||||||
|
|
||||||
const pnUser = pnDecoded.user
|
validatedPairs.push({ pnUser: pnDecoded.user, lidUser: lidDecoded.user })
|
||||||
const lidUser = lidDecoded.user
|
}
|
||||||
|
|
||||||
let existingLidUser = this.mappingCache.get(`pn:${pnUser}`)
|
if (validatedPairs.length === 0) return
|
||||||
if (!existingLidUser) {
|
|
||||||
this.logger.trace(`Cache miss for PN user ${pnUser}; checking database`)
|
const cacheMissSet = new Set<string>()
|
||||||
const stored = await this.keys.get('lid-mapping', [pnUser])
|
const existingMappings = new Map<string, string>()
|
||||||
existingLidUser = stored[pnUser]
|
|
||||||
|
for (const { pnUser } of validatedPairs) {
|
||||||
|
const cached = this.mappingCache.get(`pn:${pnUser}`)
|
||||||
|
if (cached) {
|
||||||
|
existingMappings.set(pnUser, cached)
|
||||||
|
} else {
|
||||||
|
cacheMissSet.add(pnUser)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (cacheMissSet.size > 0) {
|
||||||
|
const cacheMisses = [...cacheMissSet]
|
||||||
|
this.logger.trace(`Batch fetching ${cacheMisses.length} LID mappings from database`)
|
||||||
|
const stored = await this.keys.get('lid-mapping', cacheMisses)
|
||||||
|
|
||||||
|
for (const pnUser of cacheMisses) {
|
||||||
|
const existingLidUser = stored[pnUser]
|
||||||
if (existingLidUser) {
|
if (existingLidUser) {
|
||||||
// Update cache with database value
|
existingMappings.set(pnUser, existingLidUser)
|
||||||
this.mappingCache.set(`pn:${pnUser}`, existingLidUser)
|
this.mappingCache.set(`pn:${pnUser}`, existingLidUser)
|
||||||
this.mappingCache.set(`lid:${existingLidUser}`, pnUser)
|
this.mappingCache.set(`lid:${existingLidUser}`, pnUser)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const pairMap: { [_: string]: string } = {}
|
||||||
|
for (const { pnUser, lidUser } of validatedPairs) {
|
||||||
|
const existingLidUser = existingMappings.get(pnUser)
|
||||||
if (existingLidUser === lidUser) {
|
if (existingLidUser === lidUser) {
|
||||||
this.logger.debug({ pnUser, lidUser }, 'LID mapping already exists, skipping')
|
this.logger.debug({ pnUser, lidUser }, 'LID mapping already exists, skipping')
|
||||||
continue
|
continue
|
||||||
@@ -64,33 +84,55 @@ export class LIDMappingStore {
|
|||||||
pairMap[pnUser] = lidUser
|
pairMap[pnUser] = lidUser
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (Object.keys(pairMap).length === 0) return
|
||||||
|
|
||||||
this.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 () => {
|
const batchData: { [key: string]: string } = {}
|
||||||
for (const [pnUser, lidUser] of Object.entries(pairMap)) {
|
for (const [pnUser, lidUser] of Object.entries(pairMap)) {
|
||||||
await this.keys.set({
|
batchData[pnUser] = lidUser
|
||||||
'lid-mapping': {
|
batchData[`${lidUser}_reverse`] = pnUser
|
||||||
[pnUser]: lidUser,
|
}
|
||||||
[`${lidUser}_reverse`]: pnUser
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
this.mappingCache.set(`pn:${pnUser}`, lidUser)
|
await this.keys.transaction(async () => {
|
||||||
this.mappingCache.set(`lid:${lidUser}`, pnUser)
|
await this.keys.set({ 'lid-mapping': batchData })
|
||||||
}
|
|
||||||
}, 'lid-mapping')
|
}, 'lid-mapping')
|
||||||
|
|
||||||
|
// Update cache after successful DB write
|
||||||
|
for (const [pnUser, lidUser] of Object.entries(pairMap)) {
|
||||||
|
this.mappingCache.set(`pn:${pnUser}`, lidUser)
|
||||||
|
this.mappingCache.set(`lid:${lidUser}`, pnUser)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Get LID for PN - Returns device-specific LID based on user mapping
|
|
||||||
*/
|
|
||||||
async getLIDForPN(pn: string): Promise<string | null> {
|
async getLIDForPN(pn: string): Promise<string | null> {
|
||||||
return (await this.getLIDsForPNs([pn]))?.[0]?.lid || null
|
return (await this.getLIDsForPNs([pn]))?.[0]?.lid || null
|
||||||
}
|
}
|
||||||
|
|
||||||
async getLIDsForPNs(pns: string[]): Promise<LIDMapping[] | null> {
|
async getLIDsForPNs(pns: string[]): Promise<LIDMapping[] | null> {
|
||||||
|
if (pns.length === 0) return null
|
||||||
|
|
||||||
|
const sortedPns = [...new Set(pns)].sort()
|
||||||
|
const cacheKey = sortedPns.join(',')
|
||||||
|
|
||||||
|
const inflight = this.inflightLIDLookups.get(cacheKey)
|
||||||
|
if (inflight) {
|
||||||
|
this.logger.trace(`Coalescing getLIDsForPNs request for ${sortedPns.length} PNs`)
|
||||||
|
return inflight
|
||||||
|
}
|
||||||
|
|
||||||
|
const promise = this._getLIDsForPNsImpl(pns)
|
||||||
|
this.inflightLIDLookups.set(cacheKey, promise)
|
||||||
|
|
||||||
|
try {
|
||||||
|
return await promise
|
||||||
|
} finally {
|
||||||
|
this.inflightLIDLookups.delete(cacheKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async _getLIDsForPNsImpl(pns: string[]): Promise<LIDMapping[] | null> {
|
||||||
const usyncFetch: { [_: string]: number[] } = {}
|
const usyncFetch: { [_: string]: number[] } = {}
|
||||||
// mapped from pn to lid mapping to prevent duplication in results later
|
|
||||||
const successfulPairs: { [_: string]: LIDMapping } = {}
|
const successfulPairs: { [_: string]: LIDMapping } = {}
|
||||||
const pending: Array<{ pn: string; pnUser: string; decoded: ReturnType<typeof jidDecode> }> = []
|
const pending: Array<{ pn: string; pnUser: string; decoded: ReturnType<typeof jidDecode> }> = []
|
||||||
|
|
||||||
@@ -118,7 +160,6 @@ export class LIDMappingStore {
|
|||||||
const decoded = jidDecode(pn)
|
const decoded = jidDecode(pn)
|
||||||
if (!decoded) continue
|
if (!decoded) continue
|
||||||
|
|
||||||
// Check cache first for PN → LID mapping
|
|
||||||
const pnUser = decoded.user
|
const pnUser = decoded.user
|
||||||
const cached = this.mappingCache.get(`pn:${pnUser}`)
|
const cached = this.mappingCache.get(`pn:${pnUser}`)
|
||||||
if (cached && typeof cached === 'string') {
|
if (cached && typeof cached === 'string') {
|
||||||
@@ -199,14 +240,33 @@ export class LIDMappingStore {
|
|||||||
return Object.values(successfulPairs).length > 0 ? Object.values(successfulPairs) : null
|
return Object.values(successfulPairs).length > 0 ? Object.values(successfulPairs) : null
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Get PN for LID - USER LEVEL with device construction
|
|
||||||
*/
|
|
||||||
async getPNForLID(lid: string): Promise<string | null> {
|
async getPNForLID(lid: string): Promise<string | null> {
|
||||||
return (await this.getPNsForLIDs([lid]))?.[0]?.pn || null
|
return (await this.getPNsForLIDs([lid]))?.[0]?.pn || null
|
||||||
}
|
}
|
||||||
|
|
||||||
async getPNsForLIDs(lids: string[]): Promise<LIDMapping[] | null> {
|
async getPNsForLIDs(lids: string[]): Promise<LIDMapping[] | null> {
|
||||||
|
if (lids.length === 0) return null
|
||||||
|
|
||||||
|
const sortedLids = [...new Set(lids)].sort()
|
||||||
|
const cacheKey = sortedLids.join(',')
|
||||||
|
|
||||||
|
const inflight = this.inflightPNLookups.get(cacheKey)
|
||||||
|
if (inflight) {
|
||||||
|
this.logger.trace(`Coalescing getPNsForLIDs request for ${sortedLids.length} LIDs`)
|
||||||
|
return inflight
|
||||||
|
}
|
||||||
|
|
||||||
|
const promise = this._getPNsForLIDsImpl(lids)
|
||||||
|
this.inflightPNLookups.set(cacheKey, promise)
|
||||||
|
|
||||||
|
try {
|
||||||
|
return await promise
|
||||||
|
} finally {
|
||||||
|
this.inflightPNLookups.delete(cacheKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async _getPNsForLIDsImpl(lids: string[]): Promise<LIDMapping[] | null> {
|
||||||
const successfulPairs: { [_: string]: LIDMapping } = {}
|
const successfulPairs: { [_: string]: LIDMapping } = {}
|
||||||
const pending: Array<{ lid: string; lidUser: string; decoded: ReturnType<typeof jidDecode> }> = []
|
const pending: Array<{ lid: string; lidUser: string; decoded: ReturnType<typeof jidDecode> }> = []
|
||||||
|
|
||||||
|
|||||||
+16
-2
@@ -478,6 +478,20 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
const resyncAppState = ev.createBufferedFunction(
|
const resyncAppState = ev.createBufferedFunction(
|
||||||
async (collections: readonly WAPatchName[], isInitialSync: boolean) => {
|
async (collections: readonly WAPatchName[], isInitialSync: boolean) => {
|
||||||
|
const appStateSyncKeyCache = new Map<string, proto.Message.IAppStateSyncKeyData | null>()
|
||||||
|
|
||||||
|
const getCachedAppStateSyncKey = async (
|
||||||
|
keyId: string
|
||||||
|
): Promise<proto.Message.IAppStateSyncKeyData | null | undefined> => {
|
||||||
|
if (appStateSyncKeyCache.has(keyId)) {
|
||||||
|
return appStateSyncKeyCache.get(keyId) ?? undefined
|
||||||
|
}
|
||||||
|
|
||||||
|
const key = await getAppStateSyncKey(keyId)
|
||||||
|
appStateSyncKeyCache.set(keyId, key ?? null)
|
||||||
|
return key
|
||||||
|
}
|
||||||
|
|
||||||
// we use this to determine which events to fire
|
// we use this to determine which events to fire
|
||||||
// otherwise when we resync from scratch -- all notifications will fire
|
// otherwise when we resync from scratch -- all notifications will fire
|
||||||
const initialVersionMap: { [T in WAPatchName]?: number } = {}
|
const initialVersionMap: { [T in WAPatchName]?: number } = {}
|
||||||
@@ -547,7 +561,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
const { state: newState, mutationMap } = await decodeSyncdSnapshot(
|
const { state: newState, mutationMap } = await decodeSyncdSnapshot(
|
||||||
name,
|
name,
|
||||||
snapshot,
|
snapshot,
|
||||||
getAppStateSyncKey,
|
getCachedAppStateSyncKey,
|
||||||
initialVersionMap[name],
|
initialVersionMap[name],
|
||||||
appStateMacVerification.snapshot
|
appStateMacVerification.snapshot
|
||||||
)
|
)
|
||||||
@@ -565,7 +579,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
name,
|
name,
|
||||||
patches,
|
patches,
|
||||||
states[name],
|
states[name],
|
||||||
getAppStateSyncKey,
|
getCachedAppStateSyncKey,
|
||||||
config.options,
|
config.options,
|
||||||
initialVersionMap[name],
|
initialVersionMap[name],
|
||||||
logger,
|
logger,
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ export type BaileysEventMap = {
|
|||||||
chats: Chat[]
|
chats: Chat[]
|
||||||
contacts: Contact[]
|
contacts: Contact[]
|
||||||
messages: WAMessage[]
|
messages: WAMessage[]
|
||||||
|
lidPnMappings?: LIDMapping[]
|
||||||
isLatest?: boolean
|
isLatest?: boolean
|
||||||
progress?: number | null
|
progress?: number | null
|
||||||
syncType?: proto.HistorySync.HistorySyncType | null
|
syncType?: proto.HistorySync.HistorySyncType | null
|
||||||
|
|||||||
@@ -287,14 +287,11 @@ const processMessage = async (
|
|||||||
|
|
||||||
const data = await downloadAndProcessHistorySyncNotification(histNotification, options, logger)
|
const data = await downloadAndProcessHistorySyncNotification(histNotification, options, logger)
|
||||||
|
|
||||||
// Emit LID-PN mappings from history sync
|
|
||||||
// This is how WhatsApp Web learns mappings for chats with non-contacts
|
|
||||||
if (data.lidPnMappings?.length) {
|
if (data.lidPnMappings?.length) {
|
||||||
logger?.debug({ count: data.lidPnMappings.length }, 'processing LID-PN mappings from history sync')
|
logger?.debug({ count: data.lidPnMappings.length }, 'processing LID-PN mappings from history sync')
|
||||||
// eslint-disable-next-line max-depth
|
await signalRepository.lidMapping
|
||||||
for (const mapping of data.lidPnMappings) {
|
.storeLIDPNMappings(data.lidPnMappings)
|
||||||
ev.emit('lid-mapping.update', mapping)
|
.catch(err => logger?.warn({ err }, 'failed to store LID-PN mappings from history sync'))
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
ev.emit('messaging-history.set', {
|
ev.emit('messaging-history.set', {
|
||||||
|
|||||||
Reference in New Issue
Block a user