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",
@@ -101,7 +101,8 @@
     "ts-proto": "^1.156.7",
     "typedoc": "^0.24.8",
     "typedoc-plugin-markdown": "^3.15.3",
-    "typescript": "^5.1.6"
+    "typescript": "^5.1.6",
+    "undici": "^5.27.2"
   },
   "peerDependencies": {
     "fastify": ">= 4"

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}`
   }
 
   /**
@@ -57,8 +64,8 @@ export class BlobApi {
         variant: 'original',
         type: blobType,
       },
-      metadata,
-      contentHash
+      metadata
+      // contentHash
     )
 
     if (preview) {
@@ -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

@@ -24,6 +24,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 { MediaServer } from './media-server.js'
 
 /** @typedef {import("@mapeo/schema").ProjectSettingsValue} ProjectValue */
 
@@ -50,14 +51,16 @@ export class MapeoManager {
   #deviceId
   #rpc
   #invite
+  #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 +106,11 @@ export class MapeoManager {
     } else {
       this.#coreStorage = coreStorage
     }
+
+    this.#mediaServer = new MediaServer({
+      logger: mediaServerOpts?.logger,
+      getProject: this.getProject.bind(this),
+    })
   }
 
   /**
@@ -214,6 +222,9 @@ export class MapeoManager {
       sharedDb: this.#db,
       sharedIndexWriter: this.#projectSettingsIndexWriter,
       rpc: this.#rpc,
+      getMediaBaseUrl: this.#mediaServer.getMediaAddress.bind(
+        this.#mediaServer
+      ),
     })
 
     // 5. Write project name and any other relevant metadata to project instance
@@ -270,6 +281,7 @@ export class MapeoManager {
       sharedDb: this.#db,
       sharedIndexWriter: this.#projectSettingsIndexWriter,
       rpc: this.#rpc,
+      getMediaBaseUrl: this.#mediaServer.getMediaAddress,
     })
 
     // 3. Keep track of project instance as we know it's a properly existing project
@@ -425,4 +437,15 @@ export class MapeoManager {
   get invite() {
     return this.#invite
   }
+
+  /**
+   * @param {import('./media-server.js').StartOpts} [opts]
+   */
+  async start(opts) {
+    await this.#mediaServer.start(opts)
+  }
+
+  async stop() {
+    await this.#mediaServer.stop()
+  }
 }

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'
@@ -30,7 +29,12 @@ import {
   mapAndValidateCoreOwnership,
 } from './core-ownership.js'
 import { Capabilities } from './capabilities.js'
-import { getDeviceId, projectKeyToId, valueOf } from './utils.js'
+import {
+  getDeviceId,
+  projectKeyToId,
+  projectKeyToPublicId,
+  valueOf,
+} from './utils.js'
 import { MemberApi } from './member-api.js'
 import { SyncController } from './sync/sync-controller.js'
 
@@ -42,6 +46,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,12 +55,12 @@ export class MapeoProject {
   #dataStores
   #dataTypes
   #blobStore
-  #blobServer
   #coreOwnership
   #capabilities
   #ownershipWriteDone
   #memberApi
   #syncController
+  #projectPublicId
 
   /**
    * @param {Object} opts
@@ -68,6 +73,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,9 +86,11 @@ export class MapeoProject {
     projectSecretKey,
     encryptionKeys,
     rpc,
+    getMediaBaseUrl,
   }) {
     this.#deviceId = getDeviceId(keyManager)
     this.#projectId = projectKeyToId(projectKey)
+    this.#projectPublicId = projectKeyToPublicId(projectKey)
 
     ///////// 1. Setup database
     const sqlite = new Database(dbPath)
@@ -205,18 +213,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,
+      projectId: this.#projectPublicId,
       blobStore: this.#blobStore,
-      blobServer: this.#blobServer,
+      getMediaBaseUrl: async () => getMediaBaseUrl('blobs'),
     })
 
     this.#coreOwnership = new CoreOwnership({
@@ -286,6 +286,10 @@ export class MapeoProject {
     return this.#capabilities
   }
 
+  get [kBlobStore]() {
+    return this.#blobStore
+  }
+
   get deviceId() {
     return this.#deviceId
   }