feat: $blobs.getUrl and $blobs.create methods

src/blob-api.js

@@ -0,0 +1,124 @@
+import fs from 'node:fs'
+import { basename } from 'node:path'
+import sodium from 'sodium-universal'
+import b4a from 'b4a'
+
+import { getPort } from './blob-server/index.js'
+
+/** @typedef {import('./types.js').BlobId} BlobId */
+/** @typedef {import('./types.js').BlobType} BlobType  */
+/** @typedef {import('./types.js').BlobVariant<BlobType>} BlobVariant  */
+
+export class BlobApi {
+  /**
+   * @param {object} options
+   * @param {string} options.projectId
+   * @param {import('./blob-store/index.js').BlobStore} options.blobStore
+   * @param {import('fastify').FastifyInstance} options.blobServer
+   */
+  constructor({ projectId, blobStore, blobServer }) {
+    this.projectId = projectId
+    this.blobStore = blobStore
+    this.blobServer = blobServer
+  }
+
+  /**
+   * Get a url for a blob based on its BlobId
+   * @param {import('./types.js').BlobId} blobId
+   * @returns {Promise<string>}
+   */
+  async getUrl(blobId) {
+    const { driveId, type, variant, name } = blobId
+    const port = await getPort(this.blobServer.server)
+    return `http://127.0.0.1:${port}/${this.projectId}/${driveId}/${type}/${variant}/${name}`
+  }
+
+  /**
+   * Write blobs for provided variants of a file
+   * @param {{ original: string, preview?: string, thumbnail?: string }} filepaths
+   * @param {{ mimeType: string }} metadata
+   * @returns {Promise<{ original: Omit<BlobId, 'driveId'>, preview?: Omit<BlobId, 'driveId'>, thumbnail?: Omit<BlobId, 'driveId'> }>}
+   */
+  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 originalBlobId = await this.writeFile(
+      original,
+      {
+        name: `${name}_${basename(original)}`,
+        variant: 'original',
+        type: blobType,
+      },
+      metadata
+    )
+    const previewBlobId = preview
+      ? await this.writeFile(
+          preview,
+          {
+            name: `${name}_${basename(preview)}`,
+            variant: 'preview',
+            type: blobType,
+          },
+          metadata
+        )
+      : null
+    const thumbnailBlobId = thumbnail
+      ? await this.writeFile(
+          thumbnail,
+          {
+            name: `${name}_${basename(thumbnail)}`,
+            variant: 'thumbnail',
+            type: blobType,
+          },
+          metadata
+        )
+      : null
+
+    const blobIds =
+      /** @type {{ original: Omit<BlobId, 'driveId'>, preview?: Omit<BlobId, 'driveId'>, thumbnail?: Omit<BlobId, 'driveId'> }} */ ({
+        original: originalBlobId,
+      })
+
+    if (previewBlobId) blobIds.preview = previewBlobId
+    if (thumbnailBlobId) blobIds.thumbnail = thumbnailBlobId
+
+    return blobIds
+  }
+
+  /**
+   * @param {Omit<BlobId, 'driveId'>} options
+   * @returns {Promise<Omit<BlobId, 'driveId'>>}
+   */
+  async writeFile(filepath, { name, variant, type }, metadata) {
+    return new Promise((resolve, reject) => {
+      fs.createReadStream(filepath)
+        .pipe(
+          this.blobStore.createWriteStream(
+            { type, variant, name },
+            { metadata }
+          )
+        )
+        .on('error', reject)
+        .on('finish', () => {
+          resolve({ type, variant, name })
+        })
+    })
+  }
+}
+
+/**
+ * @param {string} mimeType
+ * @returns {BlobType}
+ */
+function getType(mimeType) {
+  if (mimeType.startsWith('image')) return 'photo'
+  if (mimeType.startsWith('video')) return 'video'
+  if (mimeType.startsWith('audio')) return 'audio'
+
+  throw new Error(`Unsupported mimeType: ${mimeType}`)
+}

src/blob-server/index.js

@@ -1,3 +1,4 @@
+import { once } from 'events'
 import fastify from 'fastify'
 
 import BlobServerPlugin from './fastify-plugin.js'
@@ -23,3 +24,18 @@ export function createBlobServer({ logger, blobStore, prefix, projectId }) {
   })
   return server
 }
+
+/**
+ * @param {import('node:http').Server} server
+ * @returns {Promise<number>}
+ */
+export async function getPort(server) {
+  const address = server.address()
+
+  if (!address || !(typeof address === 'object') || !address.port) {
+    await once(server, 'listening')
+    return getPort(server)
+  }
+
+  return address.port
+}

src/mapeo-project.js

@@ -5,6 +5,8 @@ import { migrate } from 'drizzle-orm/better-sqlite3/migrator'
 import { CoreManager } from './core-manager/index.js'
 import { DataStore } from './datastore/index.js'
 import { DataType } from './datatype/index.js'
+import { BlobStore } from './blobstore/index.js'
+import { BlobApi } from './blob-api.js'
 import { IndexWriter } from './index-writer/index.js'
 import { fieldTable, observationTable, presetTable } from './schema/project.js'
 import RandomAccessFile from 'random-access-file'
@@ -23,9 +25,11 @@ const INDEXER_STORAGE_FOLDER_NAME = 'indexer'
 const MAX_FILE_DESCRIPTORS = 768
 
 export class MapeoProject {
+  #projectId
   #coreManager
   #dataStores
   #dataTypes
+  #blobStore
 
   /**
    * @param {Object} opts
@@ -36,6 +40,8 @@ export class MapeoProject {
    * @param {Partial<Record<import('./core-manager/index.js').Namespace, Buffer>>} [opts.encryptionKeys] Encryption keys for each namespace
    */
   constructor({ storagePath, ...coreManagerOpts }) {
+    this.#projectId = coreManagerOpts.projectKey.toString('hex') // TODO: update based on outcome of https://github.com/digidem/mapeo-core-next/issues/171
+
     ///////// 1. Setup database
 
     const dbPath =
@@ -113,6 +119,15 @@ export class MapeoProject {
         db,
       }),
     }
+
+    this.#blobStore = new BlobStore({
+      coreManager: this.#coreManager,
+    })
+
+    this.$blobs = new BlobApi({
+      projectId: this.#projectId,
+      blobStore: this.#blobStore,
+    })
   }
 
   get observation() {

src/types.ts

@@ -16,8 +16,8 @@ import Corestore from 'corestore'
 import Hypercore from 'hypercore'
 
 type SupportedBlobVariants = typeof SUPPORTED_BLOB_VARIANTS
-type BlobType = keyof SupportedBlobVariants
-type BlobVariant<TBlobType extends BlobType> = TupleToUnion<
+export type BlobType = keyof SupportedBlobVariants
+export type BlobVariant<TBlobType extends BlobType> = TupleToUnion<
   SupportedBlobVariants[TBlobType]
 >
 

tests/blob-api.js

@@ -0,0 +1,122 @@
+import { join, basename } from 'node:path'
+import { fileURLToPath } from 'url'
+import test from 'brittle'
+import { BlobApi } from '../src/blob-api.js'
+import { createBlobServer, getPort } from '../src/blob-server/index.js'
+import { createBlobStore } from './helpers/blob-store.js'
+import { timeoutException } from './helpers/index.js'
+
+test('get port after listening event with explicit port', async (t) => {
+  const blobStore = createBlobStore()
+  const server = await createBlobServer({ blobStore })
+
+  t.ok(await timeoutException(getPort(server.server)))
+
+  await new Promise((resolve) => {
+    server.listen({ port: 3456 }, (err, address) => {
+      resolve(address)
+    })
+  })
+
+  const port = await getPort(server.server)
+
+  t.is(typeof port, 'number')
+  t.is(port, 3456)
+
+  t.teardown(async () => {
+    await server.close()
+  })
+})
+
+test('get port after listening event with unset port', async (t) => {
+  const blobStore = createBlobStore()
+  const server = await createBlobServer({ blobStore })
+
+  t.ok(await timeoutException(getPort(server.server)))
+
+  await new Promise((resolve) => {
+    server.listen({ port: 0 }, (err, address) => {
+      resolve(address)
+    })
+  })
+
+  const port = await getPort(server.server)
+
+  t.is(typeof port, 'number', 'port is a number')
+  t.teardown(async () => {
+    await server.close()
+  })
+})
+
+test('get url from blobId', async (t) => {
+  const projectId = '1234'
+  const driveId = '1234'
+  const type = 'image'
+  const variant = 'original'
+  const name = '1234'
+
+  const blobStore = createBlobStore()
+  const blobServer = await createBlobServer({ blobStore })
+  const blobApi = new BlobApi({ projectId: '1234', blobStore, blobServer })
+
+  await new Promise((resolve) => {
+    blobServer.listen({ port: 0 }, (err, address) => {
+      resolve(address)
+    })
+  })
+
+  const url = await blobApi.getUrl({ driveId, type, variant, name })
+
+  t.is(
+    url,
+    `http://127.0.0.1:${
+      blobServer.server.address().port
+    }/${projectId}/${driveId}/${type}/${variant}/${name}`
+  )
+  t.teardown(async () => {
+    await blobServer.close()
+  })
+})
+
+test('create blobs', async (t) => {
+  const { blobStore } = createBlobStore()
+  const blobServer = await createBlobServer({ blobStore })
+  const blobApi = new BlobApi({ projectId: '1234', blobStore, blobServer })
+
+  await new Promise((resolve) => {
+    blobServer.listen({ port: 0 }, (err, address) => {
+      resolve(address)
+    })
+  })
+
+  const directory = fileURLToPath(
+    new URL('./fixtures/blob-api/', import.meta.url)
+  )
+
+  const blobIds = await blobApi.create(
+    {
+      original: join(directory, 'original.png'),
+      preview: join(directory, 'preview.png'),
+      thumbnail: join(directory, 'thumbnail.png'),
+    },
+    {
+      mimeType: 'image/png',
+    }
+  )
+
+  t.is(blobIds.original.type, 'photo')
+  t.is(blobIds.original.variant, 'original')
+  t.ok(blobIds.original.name.includes('_original.png'))
+
+  t.is(blobIds.preview.type, 'photo')
+  t.is(blobIds.preview.variant, 'preview')
+  t.ok(blobIds.preview.name.includes('_preview.png'))
+
+  t.is(blobIds.thumbnail.type, 'photo')
+  t.is(blobIds.thumbnail.variant, 'thumbnail')
+  t.ok(blobIds.thumbnail.name.includes('_thumbnail.png'))
+
+  t.teardown(async () => {
+    await blobServer.close()
+  })
+})

tests/blob-server.js

@@ -3,13 +3,11 @@ import test from 'brittle'
 import { readdirSync } from 'fs'
 import { readFile } from 'fs/promises'
 import path from 'path'
-import { createCoreManager } from './helpers/core-manager.js'
-import { BlobStore } from '../src/blob-store/index.js'
 import { createBlobServer } from '../src/blob-server/index.js'
 import BlobServerPlugin from '../src/blob-server/fastify-plugin.js'
 import fastify from 'fastify'
 
-import { replicateBlobs } from './helpers/blob-store.js'
+import { replicateBlobs, createBlobStore } from './helpers/blob-store.js'
 
 test('Plugin throws error if missing getBlobStore option', async (t) => {
   const server = fastify()
@@ -217,12 +215,6 @@ test('GET photo returns 404 when trying to get non-replicated blob', async (t) =
   t.is(res.statusCode, 404)
 })
 
-function createBlobStore(opts) {
-  const coreManager = createCoreManager(opts)
-  const blobStore = new BlobStore({ coreManager })
-  return { blobStore, coreManager }
-}
-
 async function testenv({ prefix, logger } = {}) {
   const projectKey = randomBytes(32)
   const projectId = projectKey.toString('hex')

tests/helpers/blob-store.js

@@ -1,6 +1,20 @@
 import { replicate } from './core-manager.js'
 import { pipelinePromise as pipeline, Writable } from 'streamx'
 
+import { BlobStore } from '../../src/blob-store/index.js'
+import { createCoreManager } from './core-manager.js'
+
+/**
+ *
+ * @param {Object} options
+ * @returns
+ */
+export function createBlobStore(options = {}) {
+  const coreManager = createCoreManager(options)
+  const blobStore = new BlobStore({ coreManager })
+  return { blobStore, coreManager }
+}
+
 /**
  *
  * @param {import('../../src/core-manager/index.js').CoreManager} cm1

tests/helpers/index.js

@@ -73,3 +73,13 @@ export async function waitForIndexing(stores) {
     })
   )
 }
+
+export async function timeoutException(promise, timeout = 100) {
+  const timer = new Promise((resolve) => {
+    setTimeout(() => {
+      resolve('timeout')
+    }, timeout)
+  })
+
+  return (await Promise.race([promise, timer])) === 'timeout'
+}