fix: fix writing to blob store when also calculating content hash

src/blob-api.js

@@ -1,8 +1,7 @@
 import fs from 'node:fs'
-import { pipeline } from 'node:stream/promises'
-import { createHash } from 'node:crypto'
-import sodium from 'sodium-universal'
-import b4a from 'b4a'
+// @ts-expect-error - pipelinePromise missing from streamx types
+import { Transform, pipelinePromise as pipeline } from 'streamx'
+import { createHash, randomBytes } from 'node:crypto'
 
 /** @typedef {import('./types.js').BlobId} BlobId */
 /** @typedef {import('./types.js').BlobType} BlobType  */
@@ -47,85 +46,55 @@ export class BlobApi {
   async create(filepaths, metadata) {
     const { original, preview, thumbnail } = filepaths
     const { mimeType } = metadata
-    const blobType = getType(mimeType)
-    const hash = b4a.alloc(8)
-    sodium.randombytes_buf(hash)
-    const name = hash.toString('hex')
-
-    const contentHash = createHash('sha256')
-
-    await this.writeFile(
-      original,
-      {
-        name: `${name}`,
-        variant: 'original',
-        type: blobType,
-      },
-      metadata
-      // contentHash
+    const type = getType(mimeType)
+    const name = randomBytes(8).toString('hex')
+    const hash = createHash('sha256')
+
+    const ws = this.#blobStore.createWriteStream(
+      { type, variant: 'original', name },
+      { metadata }
     )
+    const writePromises = [
+      pipeline(fs.createReadStream(original), hashTransform(hash), ws),
+    ]
 
     if (preview) {
-      await this.writeFile(
-        preview,
-        {
-          name: `${name}`,
-          variant: 'preview',
-          type: blobType,
-        },
-        metadata
+      const ws = this.#blobStore.createWriteStream(
+        { type, variant: 'preview', name },
+        { metadata }
       )
+      writePromises.push(pipeline(fs.createReadStream(preview), ws))
     }
 
     if (thumbnail) {
-      await this.writeFile(
-        thumbnail,
-        {
-          name: `${name}`,
-          variant: 'thumbnail',
-          type: blobType,
-        },
-        metadata
+      const ws = this.#blobStore.createWriteStream(
+        { type, variant: 'thumbnail', name },
+        { metadata }
       )
+      writePromises.push(pipeline(fs.createReadStream(thumbnail), ws))
     }
 
+    await Promise.all(writePromises)
+
     return {
       driveId: this.#blobStore.writerDriveId,
       name,
-      type: blobType,
-      hash: contentHash.digest('hex'),
+      type,
+      hash: hash.digest('hex'),
     }
   }
+}
 
-  /**
-   * @param {string} filepath
-   * @param {Omit<BlobId, 'driveId'>} options
-   * @param {object} metadata
-   * @param {string} metadata.mimeType
-   * @param {import('node:crypto').Hash} [hash]
-   */
-  async writeFile(filepath, { name, variant, type }, metadata, hash) {
-    if (hash) {
-      // @ts-ignore TODO: return value types don't match pipeline's expectations, though they should
-      await pipeline(
-        fs.createReadStream(filepath),
-        hash,
-
-        // @ts-ignore TODO: remove driveId property from createWriteStream
-        this.#blobStore.createWriteStream({ type, variant, name }, { metadata })
-      )
-
-      return { name, variant, type, hash }
-    }
-
-    // @ts-ignore TODO: return value types don't match pipeline's expectations, though they should
-    await pipeline(
-      fs.createReadStream(filepath),
-      this.#blobStore.createWriteStream({ type, variant, name }, { metadata })
-    )
-
-    return { name, variant, type }
-  }
+/**
+ * @param {import('node:crypto').Hash} hash
+ */
+function hashTransform(hash) {
+  return new Transform({
+    transform: (data, cb) => {
+      hash.update(data)
+      cb(null, data)
+    },
+  })
 }
 
 /**

tests/blob-api.js

@@ -22,6 +22,7 @@ test('create blobs', async (t) => {
 
   const hash = createHash('sha256')
   const originalContent = await fs.readFile(join(directory, 'original.png'))
+
   hash.update(originalContent)
 
   const attachment = await blobApi.create(
@@ -35,9 +36,7 @@ test('create blobs', async (t) => {
 
   t.is(attachment.driveId, blobStore.writerDriveId)
   t.is(attachment.type, 'photo')
-  // TODO: Need to fix BlobApi implementation
-  // https://github.com/digidem/mapeo-core-next/pull/365#pullrequestreview-1716846341
-  // t.alike(attachment.hash, hash.digest('hex'))
+  t.alike(attachment.hash, hash.digest('hex'))
 })
 
 test('get url from blobId', async (t) => {