perf: experimental do not use fs for enc stream

src/Types/Message.ts

@@ -140,7 +140,7 @@ export type MessageGenerationOptionsFromContent = MiscMessageGenerationOptions &
 	userJid: string
 }
 
-export type WAMediaUploadFunction = (readStream: ReadStream, opts: { fileEncSha256B64: string, mediaType: MediaType, timeoutMs?: number }) => Promise<{ mediaUrl: string, directPath: string }>
+export type WAMediaUploadFunction = (readStream: Readable, opts: { fileEncSha256B64: string, mediaType: MediaType, timeoutMs?: number }) => Promise<{ mediaUrl: string, directPath: string }>
 
 export type MediaGenerationOptions = {
 	logger?: Logger

src/Utils/messages-media.ts

@@ -4,7 +4,7 @@ import type { Options, Response } from 'got'
 import { Boom } from '@hapi/boom'
 import * as Crypto from 'crypto'
 import { Readable, Transform } from 'stream'
-import { createReadStream, createWriteStream, promises as fs, WriteStream } from 'fs'
+import { createReadStream, createWriteStream, promises as fs, ReadStream, WriteStream } from 'fs'
 import { exec } from 'child_process'
 import { tmpdir } from 'os'
 import { URL } from 'url'
@@ -219,14 +219,20 @@ export const getGotStream = async(url: string | URL, options: Options & { isStre
     })
     return fetched
 } 
-export const encryptedStream = async(media: WAMediaUpload, mediaType: MediaType, saveOriginalFileIfRequired = true) => {
+export const encryptedStream = async(
+    media: WAMediaUpload, 
+    mediaType: MediaType,
+    saveOriginalFileIfRequired = true
+) => {
     const { stream, type } = await getStream(media)
 
     const mediaKey = Crypto.randomBytes(32)
     const {cipherKey, iv, macKey} = getMediaKeys(mediaKey, mediaType)
     // random name
-    const encBodyPath = join(getTmpFilesDirectory(), mediaType + generateMessageID() + '.enc')
+    //const encBodyPath = join(getTmpFilesDirectory(), mediaType + generateMessageID() + '.enc')
-    const encWriteStream = createWriteStream(encBodyPath)
+    // const encWriteStream = createWriteStream(encBodyPath)
+    const encWriteStream = new Readable({ read: () => {} })
+
     let bodyPath: string
     let writeStream: WriteStream
     let didSaveToTmpPath = false
@@ -247,8 +253,10 @@ export const encryptedStream = async(media: WAMediaUpload, mediaType: MediaType,
     const onChunk = (buff: Buffer) => {
         sha256Enc = sha256Enc.update(buff)
         hmac = hmac.update(buff)
-        encWriteStream.write(buff)
+        encWriteStream.push(buff)
     }
+    
+    try {
         for await(const data of stream) {
             fileLength += data.length
             sha256Plain = sha256Plain.update(data)
@@ -265,14 +273,14 @@ export const encryptedStream = async(media: WAMediaUpload, mediaType: MediaType,
         const fileSha256 = sha256Plain.digest()
         const fileEncSha256 = sha256Enc.digest()
         
-    encWriteStream.write(mac)
+        encWriteStream.push(mac)
-    encWriteStream.end()
+        encWriteStream.push(null)
     
         writeStream && writeStream.end()
     
         return {
             mediaKey,
-        encBodyPath,
+            encWriteStream,
             bodyPath,
             mac,
             fileEncSha256,
@@ -280,6 +288,16 @@ export const encryptedStream = async(media: WAMediaUpload, mediaType: MediaType,
             fileLength,
             didSaveToTmpPath
         }
+    } catch(error) {
+        encWriteStream.destroy(error)
+        writeStream.destroy(error)
+        aes.destroy(error)
+        hmac.destroy(error)
+        sha256Plain.destroy(error)
+        sha256Enc.destroy(error)
+
+        throw error
+    }
 }
 
 const DEF_HOST = 'mmg.whatsapp.net'

src/Utils/messages.ts

@@ -94,7 +94,7 @@ export const prepareWAMessageMedia = async(
 	const requiresOriginalForSomeProcessing = requiresDurationComputation || requiresThumbnailComputation
 	const {
 		mediaKey,
-		encBodyPath,
+		encWriteStream,
 		bodyPath,
 		fileEncSha256,
 		fileSha256,
@@ -108,6 +108,15 @@ export const prepareWAMessageMedia = async(
 		.replace(/\//g, '_')
 		.replace(/\=+$/, '')
 	)
+
+	const [{ mediaUrl, directPath }] = await Promise.all([
+		(() => {
+			return options.upload(
+				encWriteStream,
+				{ fileEncSha256B64, mediaType, timeoutMs: options.mediaUploadTimeoutMs }
+			)
+		})(),
+		(async() => {
 			try {
 				if(requiresThumbnailComputation) {
 					uploadData.jpegThumbnail = await generateThumbnail(bodyPath, mediaType as any, options)
@@ -118,18 +127,16 @@ export const prepareWAMessageMedia = async(
 			} catch (error) {
 				options.logger?.info({ trace: error.stack }, 'failed to obtain extra info')
 			}
-	const {mediaUrl, directPath} = await options.upload(
+		})(),
-		createReadStream(encBodyPath),
+	])
-		{ fileEncSha256B64, mediaType, timeoutMs: options.mediaUploadTimeoutMs }
+	.finally(
-	)
+		async() => {
+			encWriteStream.destroy()
 			// remove tmp files
-	await Promise.all(
+			didSaveToTmpPath && bodyPath && await fs.unlink(bodyPath)
-		[
+		}
-			fs.unlink(encBodyPath),
-			didSaveToTmpPath && bodyPath && fs.unlink(bodyPath)
-		]
-		.filter(Boolean)
 	)
+
 	delete uploadData.media
 
 	const obj = WAProto.Message.fromObject({