Fix/images stream writer error - ERR_STREAM_DESTROYED

src/Socket/messages-recv.ts

@@ -183,7 +183,7 @@
 			return
 		}
 
 		let data: any
 		try {
 			data = JSON.parse(mexNode.content.toString())
 		} catch (error) {
@@ -279,7 +279,7 @@
 			case 'update':
 				const settingsNode = getBinaryNodeChild(child, 'settings')
 				if (settingsNode) {
 					const update: Record<string, any> = {}
 					const nameNode = getBinaryNodeChild(settingsNode, 'name')
 					if (nameNode?.content) update.name = nameNode.content.toString()
 
@@ -884,6 +884,12 @@
 	const handlePrivacyTokenNotification = async (node: BinaryNode) => {
 		const tokensNode = getBinaryNodeChild(node, 'tokens')
 		const from = jidNormalizedUser(node.attrs.from)
+		let lidForPN: string | null = null
+		try {
+			lidForPN = await signalRepository.lidMapping.getLIDForPN(from)
+		} catch (error) {
+			logger.warn({ error, jid: from }, 'Failed to get lid for PN in handlePrivacyTokenNotification.')
+		}
 
 		if (!tokensNode) return
 
@@ -907,6 +913,12 @@
 				await authState.keys.set({
 					tctoken: { [from]: { token: content, timestamp } }
 				})
+
+				if (lidForPN) {
+					await authState.keys.set({
+						tctoken: { [lidForPN]: { token: content, timestamp } }
+					})
+				}
 			}
 		}
 	}

src/Socket/messages-send.ts

@@ -508,8 +508,8 @@
 		const meLid = authState.creds.me?.lid
 		const meLidUser = meLid ? jidDecode(meLid)?.user : null
 
 		const encryptionPromises = (patchedMessages as any).map(
 			async ({ recipientJid: jid, message: patchedMessage }: any) => {
 				if (!jid) return null
 				let msgToEncrypt = patchedMessage
 				if (dsmMessage) {
@@ -1098,7 +1098,7 @@
 								content.url = getUrlFromDirectPath(content.directPath!)
 
 								logger.debug({ directPath: media.directPath, key: result.key }, 'media update successful')
 							} catch (err: any) {
 								error = err
 							}
 						}
@@ -1118,6 +1118,10 @@
 		},
 		sendMessage: async (jid: string, content: AnyMessageContent, options: MiscMessageGenerationOptions = {}) => {
 			const userJid = authState.creds.me!.id
+			if (!options.messageId) {
+				options.messageId =  generateMessageIDV2(sock.user?.id)
+			}
+
 			if (
 				typeof content === 'object' &&
 				'disappearingMessagesInChat' in content &&
@@ -1152,7 +1156,6 @@
 					upload: waUploadToServer,
 					mediaCache: config.mediaCache,
 					options: config.options,
-					messageId: generateMessageIDV2(sock.user?.id),
 					...options
 				})
 				const isEventMsg = 'event' in content && !!content.event

src/Socket/socket.ts

@@ -192,7 +192,7 @@
 
 		const msgId = node.attrs.id
 
 		const result = await promiseTimeout<any>(timeoutMs, async (resolve, reject) => {
 			const result = waitForMessage(msgId, timeoutMs).catch(reject)
 			sendNode(node)
 				.then(async () => resolve(await result))
@@ -420,7 +420,7 @@
 		let node: proto.IClientPayload
 		if (!creds.me) {
 			node = generateRegistrationNode(creds, config)
-			logger.info({ node }, 'not logged in, attempting registration...')
+			//logger.info({ node }, 'not logged in, attempting registration...')
 		} else {
 			node = generateLoginNode(creds.me.id, config)
 			logger.info({ node }, 'logging in...')
@@ -612,7 +612,7 @@
 		}
 
 		closed = true
-		logger.info({ trace: error?.stack }, error ? 'connection errored' : 'connection closed')
+		//logger.info({ trace: error?.stack }, error ? 'connection errored' : 'connection closed')
 
 		clearInterval(keepAliveReq)
 		clearTimeout(qrTimer)
@@ -826,7 +826,7 @@
 	ws.on('open', async () => {
 		try {
 			await validateConnection()
 		} catch (err: any) {
 			logger.error({ err }, 'error in validating connection')
 			end(err)
 		}
@@ -894,7 +894,7 @@
 			ev.emit('connection.update', { isNewLogin: true, qr: undefined })
 
 			await sendNode(reply)
 		} catch (error: any) {
 			logger.info({ trace: error.stack }, 'error in pairing')
 			end(error)
 		}
@@ -986,15 +986,7 @@
 		}
 	})
 
-	let didStartBuffer = false
 	process.nextTick(() => {
-		if (creds.me?.id) {
-			// start buffering important events
-			// if we're logged in
-			ev.buffer()
-			didStartBuffer = true
-		}
-
 		ev.emit('connection.update', { connection: 'connecting', receivedPendingNotifications: false, qr: undefined })
 	})
 
@@ -1004,10 +996,6 @@
 		const offlineNotifs = +(child?.attrs.count || 0)
 
 		logger.info(`handled ${offlineNotifs} offline messages/notifications`)
-		if (didStartBuffer) {
-			ev.flush()
-			logger.trace('flushed events for initial buffer')
-		}
 
 		ev.emit('connection.update', { receivedPendingNotifications: true })
 	})

src/Utils/messages-media.ts

@@ -57,8 +57,8 @@
 	const hasher = Crypto.createHash('sha256')
 	const filePath = join(tmpdir(), mediaType + generateMessageIDV2())
 	const fileWriteStream = createWriteStream(filePath)
-
 	let fileLength = 0
+
 	try {
 		for await (const data of stream) {
 			fileLength += data.length
@@ -68,24 +68,20 @@
 			}
 		}
 
-		fileWriteStream.end()
-		await once(fileWriteStream, 'finish')
-		stream.destroy()
+		await waitForCloseOrFinishStream(fileWriteStream)
+
 		const fileSha256 = hasher.digest()
 		logger?.debug('hashed data for raw upload')
+
 		return {
 			filePath: filePath,
 			fileSha256,
 			fileLength
 		}
 	} catch (error) {
-		fileWriteStream.destroy()
-		stream.destroy()
-		try {
-			await fs.unlink(filePath)
-		} catch {
-			//
-		}
+		try { stream.destroy() } catch { }
+		await waitForCloseOrFinishStream(fileWriteStream)
+		try { await fs.unlink(filePath) } catch { }
 
 		throw error
 	}
@@ -381,6 +377,13 @@
 	opts?: RequestInit
 }
 
+const waitForCloseOrFinishStream = async (stream?: WriteStream) => {
+	if (!stream) return;
+	try { if (!stream.writableEnded) stream.end(); } catch { }
+	if (stream.destroyed || stream.closed || stream.writableFinished) return;
+	try { await Promise.race([once(stream, 'finish'), once(stream, 'close')]); } catch { }
+}
+
 export const encryptedStream = async (
 	media: WAMediaUpload,
 	mediaType: MediaType,
@@ -410,10 +413,12 @@
 	const sha256Plain = Crypto.createHash('sha256')
 	const sha256Enc = Crypto.createHash('sha256')
 
-	const onChunk = (buff: Buffer) => {
+	const writeChunk = async (buff: Buffer) => {
+		if (!encFileWriteStream.write(buff)) {
+			await once(encFileWriteStream, 'drain')
+		}
 		sha256Enc.update(buff)
 		hmac.update(buff)
-		encFileWriteStream.write(buff)
 	}
 
 	try {
@@ -437,22 +442,25 @@
 			}
 
 			sha256Plain.update(data)
-			onChunk(aes.update(data))
+			await writeChunk(aes.update(data))
 		}
 
-		onChunk(aes.final())
+		await writeChunk(aes.final())
 
 		const mac = hmac.digest().slice(0, 10)
 		sha256Enc.update(mac)
 
 		const fileSha256 = sha256Plain.digest()
 		const fileEncSha256 = sha256Enc.digest()
 
-		encFileWriteStream.write(mac)
+		if (!encFileWriteStream.write(mac)) {
+			await once(encFileWriteStream, 'drain')
+		}
 
-		encFileWriteStream.end()
-		originalFileStream?.end?.()
-		stream.destroy()
+		await Promise.allSettled([		
+			waitForCloseOrFinishStream(encFileWriteStream),
+			waitForCloseOrFinishStream(originalFileStream)
+		])
 
 		logger?.debug('encrypted data successfully')
 
@@ -466,15 +474,13 @@
 			fileLength
 		}
 	} catch (error) {
-		// destroy all streams with error
-		encFileWriteStream.destroy()
-		originalFileStream?.destroy?.()
-		aes.destroy()
-		hmac.destroy()
-		sha256Plain.destroy()
-		sha256Enc.destroy()
-		stream.destroy()
-
+		try { stream.destroy() } catch { }
+		try {
+			await Promise.allSettled([		
+				waitForCloseOrFinishStream(encFileWriteStream),
+				waitForCloseOrFinishStream(originalFileStream)
+			])
+		} catch { }
 		try {
 			await fs.unlink(encFilePath)
 			if (originalFilePath) {
@@ -483,6 +489,10 @@
 		} catch (err) {
 			logger?.error({ err }, 'failed deleting tmp files')
 		}
+		try { aes.destroy() } catch { }
+		try { hmac.destroy() } catch { }
+		try { sha256Plain.destroy() } catch { }
+		try { sha256Enc.destroy() } catch { }
 
 		throw error
 	}

src/Utils/process-message.ts

@@ -264,15 +264,15 @@ const processMessage = async (
 				const process = shouldProcessHistoryMsg
 				const isLatest = !creds.processedHistoryMessages?.length
 
-				logger?.info(
-					{
-						histNotification,
-						process,
-						id: message.key.id,
-						isLatest
-					},
-					'got history notification'
-				)
+				//logger?.info(
+				//	{
+				//		histNotification,
+				//		process,
+				//		id: message.key.id,
+				//		isLatest
+				//	},
+				//	'got history notification'
+				//)
 
 				if (process) {
 					// TODO: investigate