Run repository.injectE2ESession in chunks (#881)

package.json

@@ -52,6 +52,7 @@
     "futoin-hkdf": "^1.5.1",
     "libphonenumber-js": "^1.10.20",
     "libsignal": "github:WhiskeySockets/libsignal-node",
+    "lodash": "^4.17.21",
     "music-metadata": "^7.12.3",
     "node-cache": "^5.1.2",
     "pino": "^7.0.0",

src/Utils/signal.ts

@@ -1,3 +1,4 @@
+import { chunk } from 'lodash'
 import { KEY_BUNDLE_TYPE } from '../Defaults'
 import { SignalRepository } from '../Types'
 import { AuthenticationCreds, AuthenticationState, KeyPair, SignalIdentity, SignalKeyStore, SignedKeyPair } from '../Types/Auth'
@@ -81,27 +82,36 @@ export const parseAndInjectE2ESessions = async(
 		assertNodeErrorFree(node)
 	}
 
-	await Promise.all(
-		nodes.map(
-			async node => {
-				const signedKey = getBinaryNodeChild(node, 'skey')!
-				const key = getBinaryNodeChild(node, 'key')!
-				const identity = getBinaryNodeChildBuffer(node, 'identity')!
-				const jid = node.attrs.jid
-				const registrationId = getBinaryNodeChildUInt(node, 'registration', 4)
+	// Most of the work in repository.injectE2ESession is CPU intensive, not IO
+	// So Promise.all doesn't really help here,
+	// but blocks even loop if we're using it inside keys.transaction, and it makes it "sync" actually
+	// This way we chunk it in smaller parts and between those parts we can yield to the event loop
+	// 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 => {
+					const signedKey = getBinaryNodeChild(node, 'skey')!
+					const key = getBinaryNodeChild(node, 'key')!
+					const identity = getBinaryNodeChildBuffer(node, 'identity')!
+					const jid = node.attrs.jid
+					const registrationId = getBinaryNodeChildUInt(node, 'registration', 4)
 
-				await repository.injectE2ESession({
-					jid,
-					session: {
-						registrationId: registrationId!,
-						identityKey: generateSignalPubKey(identity),
-						signedPreKey: extractKey(signedKey)!,
-						preKey: extractKey(key)!
-					}
-				})
-			}
+					await repository.injectE2ESession({
+						jid,
+						session: {
+							registrationId: registrationId!,
+							identityKey: generateSignalPubKey(identity),
+							signedPreKey: extractKey(signedKey)!,
+							preKey: extractKey(key)!
+						}
+					})
+				}
+			)
 		)
-	)
+	}
 }
 
 export const extractDeviceJids = (result: BinaryNode, myJid: string, excludeZeroDevices: boolean) => {
@@ -174,4 +184,4 @@ export const getNextPreKeysNode = async(state: AuthenticationState, count: numbe
 	}
 
 	return { update, node }
-}
+}