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,10 +101,8 @@
     "ts-proto": "^1.156.7",
     "typedoc": "^0.24.8",
     "typedoc-plugin-markdown": "^3.15.3",
-    "typescript": "^5.1.6"
-  },
-  "peerDependencies": {
-    "fastify": ">= 4"
+    "typescript": "^5.1.6",
+    "undici": "^5.27.2"
   },
   "dependencies": {
     "@digidem/types": "^2.2.0",
@@ -125,6 +123,7 @@
     "debug": "^4.3.4",
     "drizzle-orm": "0.28.2",
     "eventemitter3": "^5.0.1",
+    "fastify": ">= 4",
     "fastify-plugin": "^4.5.0",
     "hypercore": "10.17.0",
     "hypercore-crypto": "^3.3.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
+  #projectPublicId
+
   /**
    * @param {object} options
-   * @param {string} options.projectId
+   * @param {string} options.projectPublicId
    * @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({ projectPublicId, blobStore, getMediaBaseUrl }) {
+    this.#blobStore = blobStore
+    this.#getMediaBaseUrl = getMediaBaseUrl
+    this.#projectPublicId = projectPublicId
   }
 
   /**
@@ -30,8 +31,16 @@ 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.#projectPublicId}/${driveId}/${type}/${variant}/${name}`
+    )
   }
 
   /**
@@ -57,8 +66,8 @@ export class BlobApi {
         variant: 'original',
         type: blobType,
       },
-      metadata,
-      contentHash
+      metadata
+      // contentHash
     )
 
     if (preview) {
@@ -86,7 +95,7 @@ export class BlobApi {
     }
 
     return {
-      driveId: this.blobStore.writerDriveId,
+      driveId: this.#blobStore.writerDriveId,
       name,
       type: blobType,
       hash: contentHash.digest('hex'),
@@ -108,7 +117,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 +126,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

@@ -1,21 +1,21 @@
-// @ts-check
 import fp from 'fastify-plugin'
 import { filetypemime } from 'magic-bytes.js'
 import { Type as T } from '@sinclair/typebox'
 
 import { SUPPORTED_BLOB_VARIANTS } from '../blob-store/index.js'
+import { HEX_REGEX_32_BYTES, Z_BASE_32_REGEX_32_BYTES } from './constants.js'
 
 export default fp(blobServerPlugin, {
   fastify: '4.x',
-  name: 'mapeo-blob-server',
+  name: 'mapeo-blobs',
 })
 
 /** @typedef {import('../types.js').BlobId} BlobId */
 
 /**
  * @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'][]} */ (
@@ -24,12 +24,10 @@ const BLOB_TYPES = /** @type {BlobId['type'][]} */ (
 const BLOB_VARIANTS = [
   ...new Set(Object.values(SUPPORTED_BLOB_VARIANTS).flat()),
 ]
-const HEX_REGEX_32_BYTES = '^[0-9a-fA-F]{64}$'
-const HEX_STRING_32_BYTES = T.String({ pattern: HEX_REGEX_32_BYTES })
 
 const PARAMS_JSON_SCHEMA = T.Object({
-  projectId: HEX_STRING_32_BYTES,
-  driveId: HEX_STRING_32_BYTES,
+  projectPublicId: T.String({ pattern: Z_BASE_32_REGEX_32_BYTES }),
+  driveId: T.String({ pattern: HEX_REGEX_32_BYTES }),
   type: T.Union(
     BLOB_TYPES.map((type) => {
       return T.Literal(type)
@@ -57,10 +55,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 +70,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,6 +6,8 @@ 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 { TypedEmitter } from 'tiny-typed-emitter'
+
 import { IndexWriter } from './index-writer/index.js'
 import { MapeoProject, kSetOwnDeviceInfo } from './mapeo-project.js'
 import {
@@ -26,8 +28,8 @@ 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'
 import { LocalDiscovery } from './discovery/local-discovery.js'
-import { TypedEmitter } from 'tiny-typed-emitter'
 
 /** @typedef {import("@mapeo/schema").ProjectSettingsValue} ProjectValue */
 
@@ -67,16 +69,19 @@ export class MapeoManager extends TypedEmitter {
   #deviceId
   #localPeers
   #invite
+  #mediaServer
   #localDiscovery
 
   /**
    * @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 }) {
     super()
+
     this.#dbFolder = dbFolder
     const sqlite = new Database(
       dbFolder === ':memory:'
@@ -127,6 +132,11 @@ export class MapeoManager extends TypedEmitter {
       this.#coreStorage = coreStorage
     }
 
+    this.#mediaServer = new MediaServer({
+      logger: mediaServerOpts?.logger,
+      getProject: this.getProject.bind(this),
+    })
+
     this.#localDiscovery = new LocalDiscovery({
       identityKeypair: this.#keyManager.getIdentityKeypair(),
     })
@@ -328,6 +338,9 @@ export class MapeoManager extends TypedEmitter {
       sharedDb: this.#db,
       sharedIndexWriter: this.#projectSettingsIndexWriter,
       localPeers: this.#localPeers,
+      getMediaBaseUrl: this.#mediaServer.getMediaAddress.bind(
+        this.#mediaServer
+      ),
     })
   }
 
@@ -479,6 +492,17 @@ export class MapeoManager extends TypedEmitter {
     return this.#invite
   }
 
+  /**
+   * @param {import('./media-server.js').StartOpts} [opts]
+   */
+  async start(opts) {
+    await this.#mediaServer.start(opts)
+  }
+
+  async stop() {
+    await this.#mediaServer.stop()
+  }
+
   /**
    * @returns {Promise<PublicPeerInfo[]>}
    */

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 { IconApi } from './icon-api.js'
 import { SyncApi, kSyncReplicate } from './sync/sync-api.js'
@@ -43,6 +47,7 @@ const INDEXER_STORAGE_FOLDER_NAME = 'indexer'
 export const kCoreOwnership = Symbol('coreOwnership')
 export const kCapabilities = Symbol('capabilities')
 export const kSetOwnDeviceInfo = Symbol('kSetOwnDeviceInfo')
+export const kBlobStore = Symbol('blobStore')
 export const kProjectReplicate = Symbol('replicate project')
 
 export class MapeoProject {
@@ -52,11 +57,11 @@ export class MapeoProject {
   #dataStores
   #dataTypes
   #blobStore
-  #blobServer
   #coreOwnership
   #capabilities
   #ownershipWriteDone
   #memberApi
+  #projectPublicId
   #iconApi
   #syncApi
 
@@ -70,6 +75,7 @@ export class MapeoProject {
    * @param {import('drizzle-orm/better-sqlite3').BetterSQLite3Database} opts.sharedDb
    * @param {IndexWriter} opts.sharedIndexWriter
    * @param {import('./types.js').CoreStorage} opts.coreStorage Folder to store all hypercore data
+   * @param {(mediaType: 'blobs' | 'icons') => Promise<string>} opts.getMediaBaseUrl
    * @param {import('./local-peers.js').LocalPeers} opts.localPeers
    *
    */
@@ -82,10 +88,12 @@ export class MapeoProject {
     projectKey,
     projectSecretKey,
     encryptionKeys,
+    getMediaBaseUrl,
     localPeers,
   }) {
     this.#deviceId = getDeviceId(keyManager)
     this.#projectId = projectKeyToId(projectKey)
+    this.#projectPublicId = projectKeyToPublicId(projectKey)
 
     ///////// 1. Setup database
     const sqlite = new Database(dbPath)
@@ -208,18 +216,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,
+      projectPublicId: this.#projectPublicId,
       blobStore: this.#blobStore,
-      blobServer: this.#blobServer,
+      getMediaBaseUrl: async () => getMediaBaseUrl('blobs'),
     })
 
     this.#coreOwnership = new CoreOwnership({
@@ -321,6 +321,10 @@ export class MapeoProject {
     return this.#capabilities
   }
 
+  get [kBlobStore]() {
+    return this.#blobStore
+  }
+
   get deviceId() {
     return this.#deviceId
   }