fix: refactor use natives to drop md5-file and multistream (#888)

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

package-lock.json

@@ -79,8 +79,6 @@
     "json-bigint": "^1.0.0",
     "knex": "^3.1.0",
     "lru-cache": "^10.2.0",
-    "md5-file": "^5.0.0",
-    "multistream": "^4.1.0",
     "object-sizeof": "^2.6.4",
     "pg": "^8.12.0",
     "pg-boss": "github:supabase/pg-boss#feat/exactly_once",
@@ -102,7 +100,6 @@
     "@types/jest": "^29.2.1",
     "@types/js-yaml": "^4.0.5",
     "@types/json-bigint": "^1.0.4",
-    "@types/multistream": "^4.1.3",
     "@types/mustache": "^4.2.2",
     "@types/node": "^22.18.8",
     "@types/pg": "^8.6.4",

package.json

@@ -1,10 +1,8 @@
 import { ERRORS, StorageBackendError } from '@internal/errors'
-import { randomUUID } from 'crypto'
+import { createHash, randomUUID } from 'crypto'
 import fs from 'fs-extra'
 import fsExtra from 'fs-extra'
 import * as xattr from 'fs-xattr'
-import fileChecksum from 'md5-file'
-import MultiStream from 'multistream'
 import path from 'path'
 import stream from 'stream'
 import { promisify } from 'util'
@@ -384,7 +382,7 @@ export class FileBackend implements StorageBackendAdapter {
 
     await pipeline(body, writeStream)
 
-    const etag = await fileChecksum(partPath)
+    const etag = await this.computeMd5(partPath)
 
     const platform = process.platform == 'darwin' ? 'darwin' : 'linux'
     await this.setMetadataAttr(partPath, METADATA_ATTR_KEYS[platform]['etag'], etag)
@@ -432,11 +430,7 @@ export class FileBackend implements StorageBackendAdapter {
     const finalParts = await Promise.all(partsByEtags)
     finalParts.sort((a, b) => parseInt(a.split('-')[1]) - parseInt(b.split('-')[1]))
 
-    const fileStreams = finalParts.map((partPath) => {
-      return fs.createReadStream(partPath)
-    })
-
-    const multistream = new MultiStream(fileStreams)
+    const multipartStream = this.mergePartStreams(finalParts)
     const metadataContent = await fsExtra.readFile(
       this.resolveSecurePath(
         path.join(
@@ -456,7 +450,7 @@ export class FileBackend implements StorageBackendAdapter {
       bucketName,
       key,
       version,
-      multistream,
+      multipartStream,
       metadata.contentType,
       metadata.cacheControl
     )
@@ -524,7 +518,7 @@ export class FileBackend implements StorageBackendAdapter {
     const writePart = fs.createWriteStream(partFilePath)
     await pipeline(partStream, writePart)
 
-    const etag = await fileChecksum(partFilePath)
+    const etag = await this.computeMd5(partFilePath)
     await this.setMetadataAttr(partFilePath, METADATA_ATTR_KEYS[platform]['etag'], etag)
 
     const fileStat = await fs.lstat(partFilePath)
@@ -535,6 +529,30 @@ export class FileBackend implements StorageBackendAdapter {
     }
   }
 
+  private mergePartStreams(partPaths: string[]): stream.Readable {
+    return stream.Readable.from(this.iteratePartChunks(partPaths))
+  }
+
+  private async *iteratePartChunks(partPaths: string[]): AsyncGenerator<Buffer> {
+    for (const partPath of partPaths) {
+      const partStream = fs.createReadStream(partPath)
+      for await (const chunk of partStream) {
+        yield chunk as Buffer
+      }
+    }
+  }
+
+  private async computeMd5(filePath: string): Promise<string> {
+    const hash = createHash('md5')
+    const readStream = fs.createReadStream(filePath)
+
+    for await (const chunk of readStream) {
+      hash.update(chunk)
+    }
+
+    return hash.digest('hex')
+  }
+
   /**
    * Returns a private url that can only be accessed internally by the system
    * @param bucket
@@ -676,7 +694,7 @@ export class FileBackend implements StorageBackendAdapter {
 
   private async etag(file: string, stats: fs.Stats): Promise<string> {
     if (this.etagAlgorithm === 'md5') {
-      const checksum = await fileChecksum(file)
+      const checksum = await this.computeMd5(file)
       return `"${checksum}"`
     } else if (this.etagAlgorithm === 'mtime') {
       return `"${stats.mtimeMs.toString(16)}-${stats.size.toString(16)}"`