implement media server in MapeoManager

package.json

@@ -85,7 +85,7 @@
     "depcheck": "^1.4.3",
     "drizzle-kit": "^0.19.12",
     "eslint": "^8.39.0",
-    "fastify": "^4.20.0",
+    "fastify": "^4.24.3",
     "husky": "^8.0.0",
     "light-my-request": "^5.10.0",
     "lint-staged": "^14.0.1",

src/blob-api.js

@@ -4,23 +4,24 @@ import { createHash } from 'node:crypto'
 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 {
+  #blobStore
+  #getMediaBaseUrl
+  #projectId
+
   /**
    * @param {object} options
    * @param {string} options.projectId
    * @param {import('./blob-store/index.js').BlobStore} options.blobStore
-   * @param {import('fastify').FastifyInstance} options.blobServer
+   * @param {() => Promise<string>} options.getMediaBaseUrl
    */
-  constructor({ projectId, blobStore, blobServer }) {
-    this.projectId = projectId
-    this.blobStore = blobStore
-    this.blobServer = blobServer
+  constructor({ projectId, blobStore, getMediaBaseUrl }) {
+    this.#blobStore = blobStore
+    this.#getMediaBaseUrl = getMediaBaseUrl
+    this.#projectId = projectId
   }
 
   /**
@@ -30,8 +31,14 @@ export class BlobApi {
    */
   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}`
+
+    let base = await this.#getMediaBaseUrl()
+
+    if (!base.endsWith('/')) {
+      base += '/'
+    }
+
+    return base + `${this.#projectId}/${driveId}/${type}/${variant}/${name}`
   }
 
   /**
@@ -86,7 +93,7 @@ export class BlobApi {
     }
 
     return {
-      driveId: this.blobStore.writerDriveId,
+      driveId: this.#blobStore.writerDriveId,
       name,
       type: blobType,
       hash: contentHash.digest('hex'),
@@ -108,7 +115,7 @@ export class BlobApi {
         hash,
 
         // @ts-ignore TODO: remove driveId property from createWriteStream
-        this.blobStore.createWriteStream({ type, variant, name }, { metadata })
+        this.#blobStore.createWriteStream({ type, variant, name }, { metadata })
       )
 
       return { name, variant, type, hash }
@@ -117,7 +124,7 @@ export class BlobApi {
     // @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 })
+      this.#blobStore.createWriteStream({ type, variant, name }, { metadata })
     )
 
     return { name, variant, type }

src/blob-server/fastify-plugin.js → src/fastify-plugins/blobs.js

@@ -15,7 +15,7 @@ export default fp(blobServerPlugin, {
 /**
  * @typedef {Object} BlobServerPluginOpts
  *
- * @property {(projectId: string) => import('../blob-store/index.js').BlobStore} getBlobStore
+ * @property {(projectPublicId: string) => Promise<import('../blob-store/index.js').BlobStore>} getBlobStore
  */
 
 const BLOB_TYPES = /** @type {BlobId['type'][]} */ (
@@ -27,8 +27,14 @@ const BLOB_VARIANTS = [
 const HEX_REGEX_32_BYTES = '^[0-9a-fA-F]{64}$'
 const HEX_STRING_32_BYTES = T.String({ pattern: HEX_REGEX_32_BYTES })
 
+const Z_BASE_32_REGEX_32_BYTES = '^[0-9a-zA-Z]{52}$'
+const Z_BASE_32_STRING_32_BYTES = T.String({
+  pattern: Z_BASE_32_REGEX_32_BYTES,
+})
+
 const PARAMS_JSON_SCHEMA = T.Object({
-  projectId: HEX_STRING_32_BYTES,
+  // the projectPublicId is encoded to a z-base-32 52-character string (32 bytes)
+  projectPublicId: Z_BASE_32_STRING_32_BYTES,
   driveId: HEX_STRING_32_BYTES,
   type: T.Union(
     BLOB_TYPES.map((type) => {
@@ -57,10 +63,10 @@ async function routes(fastify, options) {
   const { getBlobStore } = options
 
   fastify.get(
-    '/:projectId/:driveId/:type/:variant/:name',
+    '/:projectPublicId/:driveId/:type/:variant/:name',
     { schema: { params: PARAMS_JSON_SCHEMA } },
     async (request, reply) => {
-      const { projectId, ...blobId } = request.params
+      const { projectPublicId, ...blobId } = request.params
 
       if (!isValidBlobId(blobId)) {
         reply.code(400)
@@ -72,7 +78,7 @@ async function routes(fastify, options) {
 
       let blobStore
       try {
-        blobStore = getBlobStore(projectId)
+        blobStore = await getBlobStore(projectPublicId)
       } catch (e) {
         reply.code(404)
         throw e

src/mapeo-manager.js

@@ -6,8 +6,11 @@ import { eq } from 'drizzle-orm'
 import { drizzle } from 'drizzle-orm/better-sqlite3'
 import { migrate } from 'drizzle-orm/better-sqlite3/migrator'
 import Hypercore from 'hypercore'
+import fastify from 'fastify'
+import pDefer from 'p-defer'
+
 import { IndexWriter } from './index-writer/index.js'
-import { MapeoProject, kSetOwnDeviceInfo } from './mapeo-project.js'
+import { MapeoProject, kBlobStore, kSetOwnDeviceInfo } from './mapeo-project.js'
 import {
   localDeviceInfoTable,
   projectKeysTable,
@@ -24,6 +27,7 @@ import {
 import { RandomAccessFilePool } from './core-manager/random-access-file-pool.js'
 import { LocalPeers } from './local-peers.js'
 import { InviteApi } from './invite-api.js'
+import BlobServerPlugin from './fastify-plugins/blobs.js'
 
 /** @typedef {import("@mapeo/schema").ProjectSettingsValue} ProjectValue */
 
@@ -35,7 +39,11 @@ const CLIENT_SQLITE_FILE_NAME = 'client.db'
 // other things e.g. SQLite and other parts of the app.
 const MAX_FILE_DESCRIPTORS = 768
 
+const MEDIA_SERVER_BLOBS_PREFIX = 'blobs'
+const MEDIA_SERVER_ICONS_PREFIX = 'icons'
+
 export const kRPC = Symbol('rpc')
+export const kClose = Symbol('close')
 
 export class MapeoManager {
   #keyManager
@@ -50,14 +58,18 @@ export class MapeoManager {
   #deviceId
   #rpc
   #invite
+  /** @type {import('p-defer').DeferredPromise<string>} */
+  #deferredMediaServerListen
+  #mediaServer
 
   /**
    * @param {Object} opts
    * @param {Buffer} opts.rootKey 16-bytes of random data that uniquely identify the device, used to derive a 32-byte master key, which is used to derive all the keypairs used for Mapeo
    * @param {string} opts.dbFolder Folder for sqlite Dbs. Folder must exist. Use ':memory:' to store everything in-memory
    * @param {string | import('./types.js').CoreStorage} opts.coreStorage Folder for hypercore storage or a function that returns a RandomAccessStorage instance
+   * @param {{port?: number, logger: import('fastify').FastifyServerOptions['logger'] }} [opts.mediaServerOpts]
    */
-  constructor({ rootKey, dbFolder, coreStorage }) {
+  constructor({ rootKey, dbFolder, coreStorage, mediaServerOpts }) {
     this.#dbFolder = dbFolder
     const sqlite = new Database(
       dbFolder === ':memory:'
@@ -103,6 +115,25 @@ export class MapeoManager {
     } else {
       this.#coreStorage = coreStorage
     }
+
+    this.#mediaServer = fastify({ logger: mediaServerOpts?.logger })
+
+    this.#mediaServer.register(BlobServerPlugin, {
+      prefix: MEDIA_SERVER_BLOBS_PREFIX,
+      getBlobStore: async (projectPublicId) => {
+        const project = await this.getProject(projectPublicId)
+        return project[kBlobStore]
+      },
+    })
+
+    this.#deferredMediaServerListen = pDefer()
+    this.#mediaServer
+      .listen({ port: mediaServerOpts?.port, host: '127.0.0.1' })
+      .then(this.#deferredMediaServerListen.resolve)
+      .catch((err) => {
+        console.error('Could not start media server', err)
+        this.#deferredMediaServerListen.reject(err)
+      })
   }
 
   /**
@@ -112,6 +143,32 @@ export class MapeoManager {
     return this.#rpc
   }
 
+  /**
+   * @param {'blobs' | 'icons'} mediaType
+   * @returns
+   */
+  async #getMediaBaseUrl(mediaType) {
+    await this.#deferredMediaServerListen.promise
+
+    let address = this.#mediaServer.server.address()
+
+    // Should happen but just in case
+    if (!address) throw new Error('Could not get address')
+
+    if (typeof address !== 'string') {
+      address = address.address
+    }
+
+    switch (mediaType) {
+      case 'blobs': {
+        return `${address}/${MEDIA_SERVER_BLOBS_PREFIX}`
+      }
+      case 'icons': {
+        return `${address}/${MEDIA_SERVER_ICONS_PREFIX}`
+      }
+    }
+  }
+
   /**
    * @param {Buffer} keysCipher
    * @param {string} projectId
@@ -214,6 +271,7 @@ export class MapeoManager {
       sharedDb: this.#db,
       sharedIndexWriter: this.#projectSettingsIndexWriter,
       rpc: this.#rpc,
+      getMediaBaseUrl: this.#getMediaBaseUrl,
     })
 
     // 5. Write project name and any other relevant metadata to project instance
@@ -270,6 +328,7 @@ export class MapeoManager {
       sharedDb: this.#db,
       sharedIndexWriter: this.#projectSettingsIndexWriter,
       rpc: this.#rpc,
+      getMediaBaseUrl: this.#getMediaBaseUrl,
     })
 
     // 3. Keep track of project instance as we know it's a properly existing project
@@ -425,4 +484,10 @@ export class MapeoManager {
   get invite() {
     return this.#invite
   }
+
+  async [kClose]() {
+    // Needs to be called to ensure that the server.listen() finished fully
+    await this.#deferredMediaServerListen.promise
+    await this.#mediaServer.close()
+  }
 }

src/mapeo-project.js

@@ -11,7 +11,6 @@ import { CoreManager, NAMESPACES } from './core-manager/index.js'
 import { DataStore } from './datastore/index.js'
 import { DataType, kCreateWithDocId } from './datatype/index.js'
 import { BlobStore } from './blob-store/index.js'
-import { createBlobServer } from './blob-server/index.js'
 import { BlobApi } from './blob-api.js'
 import { IndexWriter } from './index-writer/index.js'
 import { projectSettingsTable } from './schema/client.js'
@@ -42,6 +41,7 @@ export const kCoreOwnership = Symbol('coreOwnership')
 export const kCapabilities = Symbol('capabilities')
 export const kSetOwnDeviceInfo = Symbol('kSetOwnDeviceInfo')
 export const kReplicate = Symbol('replicate')
+export const kBlobStore = Symbol('blobStore')
 
 export class MapeoProject {
   #projectId
@@ -50,7 +50,6 @@ export class MapeoProject {
   #dataStores
   #dataTypes
   #blobStore
-  #blobServer
   #coreOwnership
   #capabilities
   #ownershipWriteDone
@@ -68,6 +67,7 @@ export class MapeoProject {
    * @param {IndexWriter} opts.sharedIndexWriter
    * @param {import('./types.js').CoreStorage} opts.coreStorage Folder to store all hypercore data
    * @param {import('./local-peers.js').LocalPeers} opts.rpc
+   * @param {(mediaType: 'blobs' | 'icons') => Promise<string>} opts.getMediaBaseUrl
    *
    */
   constructor({
@@ -80,6 +80,7 @@ export class MapeoProject {
     projectSecretKey,
     encryptionKeys,
     rpc,
+    getMediaBaseUrl,
   }) {
     this.#deviceId = getDeviceId(keyManager)
     this.#projectId = projectKeyToId(projectKey)
@@ -205,18 +206,10 @@ export class MapeoProject {
       coreManager: this.#coreManager,
     })
 
-    this.#blobServer = createBlobServer({
-      logger: true,
-      blobStore: this.#blobStore,
-      prefix: '/blobs/',
-      projectId: this.#projectId,
-    })
-
-    // @ts-ignore TODO: pass in blobServer
     this.$blobs = new BlobApi({
       projectId: this.#projectId,
       blobStore: this.#blobStore,
-      blobServer: this.#blobServer,
+      getMediaBaseUrl: async () => getMediaBaseUrl('blobs'),
     })
 
     this.#coreOwnership = new CoreOwnership({
@@ -286,6 +279,10 @@ export class MapeoProject {
     return this.#capabilities
   }
 
+  get [kBlobStore]() {
+    return this.#blobStore
+  }
+
   get deviceId() {
     return this.#deviceId
   }