Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d119cb064a | |||
| b5910c591a | |||
| ae333b2910 |
@@ -40,7 +40,7 @@ import { getKeyAuthor, toNumber } from './generics'
|
|||||||
import { downloadAndProcessHistorySyncNotification } from './history'
|
import { downloadAndProcessHistorySyncNotification } from './history'
|
||||||
import type { ILogger } from './logger'
|
import type { ILogger } from './logger'
|
||||||
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
||||||
import { buildMergedTcTokenIndexWrite, resolveTcTokenJid } from './tc-token-utils'
|
import { buildMergedTcTokenIndexWrite } from './tc-token-utils'
|
||||||
|
|
||||||
type ProcessMessageContext = {
|
type ProcessMessageContext = {
|
||||||
shouldProcessHistoryMsg: boolean
|
shouldProcessHistoryMsg: boolean
|
||||||
@@ -88,27 +88,60 @@ async function storeTcTokensFromHistorySync(
|
|||||||
keyStore: SignalKeyStoreWithTransaction,
|
keyStore: SignalKeyStoreWithTransaction,
|
||||||
logger?: ILogger
|
logger?: ILogger
|
||||||
) {
|
) {
|
||||||
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
// Cheap filter first — most chats in a sync chunk don't carry tcToken at all,
|
||||||
|
// and we want to avoid spinning up promises for them.
|
||||||
const candidates: { storageJid: string; token: Buffer; ts: number; senderTs?: number }[] = []
|
const tokenChats = chats.filter(chat => {
|
||||||
for (const chat of chats) {
|
|
||||||
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
||||||
if (chat.tcToken?.length && ts > 0) {
|
return !!chat.tcToken?.length && ts > 0
|
||||||
const jid = jidNormalizedUser(chat.id!)
|
|
||||||
const storageJid = await resolveTcTokenJid(jid, getLIDForPN)
|
|
||||||
candidates.push({
|
|
||||||
storageJid,
|
|
||||||
token: Buffer.from(chat.tcToken),
|
|
||||||
ts,
|
|
||||||
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
|
||||||
})
|
})
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!candidates.length) {
|
if (!tokenChats.length) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Pre-normalize so the rest of the pipeline is a synchronous join.
|
||||||
|
const normalized = tokenChats.map(chat => ({
|
||||||
|
chat,
|
||||||
|
ts: toNumber(chat.tcTokenTimestamp!),
|
||||||
|
jid: jidNormalizedUser(chat.id!)
|
||||||
|
}))
|
||||||
|
|
||||||
|
// BATCHED LID resolution. The previous shape called getLIDForPN once per
|
||||||
|
// chat (sequential await inside a for-of), which became the bottleneck
|
||||||
|
// during heavy history sync — every cold-cache hit was a DB round-trip,
|
||||||
|
// stalling messaging-history.set and spilling into the event-buffer.
|
||||||
|
// `getLIDsForPNs` resolves a deduped list in ONE batched query (and shares
|
||||||
|
// USync retry across PNs that miss cache), turning O(N) round-trips into 1.
|
||||||
|
//
|
||||||
|
// LID inputs (and `@hosted.lid`) skip the lookup entirely — they're already
|
||||||
|
// the storage form. Failures degrade gracefully: a missing mapping just
|
||||||
|
// stores under the original jid, matching `resolveTcTokenJid`'s null branch.
|
||||||
|
const pnsToResolve = [...new Set(normalized.filter(({ jid }) => !isLidUser(jid)).map(({ jid }) => jid))]
|
||||||
|
const pnToLid = new Map<string, string>()
|
||||||
|
|
||||||
|
if (pnsToResolve.length) {
|
||||||
|
try {
|
||||||
|
const mappings = await signalRepository.lidMapping.getLIDsForPNs(pnsToResolve)
|
||||||
|
// Flat loop (continue-on-skip) keeps max nesting depth at 4 for lint.
|
||||||
|
for (const { pn, lid } of mappings ?? []) {
|
||||||
|
if (!pn || !lid) continue
|
||||||
|
pnToLid.set(jidNormalizedUser(pn), lid)
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
// Per-chat fallback below (storageJid := jid). Don't abort the chunk —
|
||||||
|
// CodeRabbit noted that all-or-nothing rejection here would drop every
|
||||||
|
// tctoken in the batch AND prevent messaging-history.set from firing.
|
||||||
|
logger?.warn({ err }, 'storeTcTokensFromHistorySync: getLIDsForPNs batch failed; falling back to per-chat jid')
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const candidates = normalized.map(({ chat, ts, jid }) => ({
|
||||||
|
storageJid: pnToLid.get(jid) ?? jid,
|
||||||
|
token: Buffer.from(chat.tcToken!),
|
||||||
|
ts,
|
||||||
|
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
||||||
|
}))
|
||||||
|
|
||||||
const jids = candidates.map(c => c.storageJid)
|
const jids = candidates.map(c => c.storageJid)
|
||||||
const existing = await keyStore.get('tctoken', jids)
|
const existing = await keyStore.get('tctoken', jids)
|
||||||
const entries: Record<string, { token: Buffer; timestamp?: string; senderTimestamp?: number }> = {}
|
const entries: Record<string, { token: Buffer; timestamp?: string; senderTimestamp?: number }> = {}
|
||||||
|
|||||||
Reference in New Issue
Block a user