wip: Upgrade to hypercore11 cinematic universe

package.json

@@ -171,23 +171,23 @@
     "@mapeo/sqlite-indexer": "^1.0.2",
     "@sinclair/typebox": "^0.33.17",
     "@sindresorhus/merge-streams": "^4.0.0",
-    "b4a": "^1.6.3",
+    "b4a": "^1.7.2",
     "bcp-47": "^2.1.0",
     "better-sqlite3": "^11.10.0",
     "big-sparse-array": "^1.0.3",
     "bogon": "^1.1.0",
     "compact-encoding": "^2.12.0",
-    "corestore": "6.8.4",
+    "corestore": "^7.4.7",
     "debug": "^4.3.4",
     "dot-prop": "^9.0.0",
     "drizzle-orm": "^0.30.8",
     "ensure-error": "^4.0.0",
     "fastify": "^4.0.0",
     "fastify-plugin": "^4.5.1",
-    "hyperblobs": "2.3.0",
-    "hypercore": "10.19.0",
-    "hypercore-crypto": "3.4.2",
-    "hyperdrive": "11.5.3",
+    "hyperblobs": "^2.8.0",
+    "hypercore": "^11.16.2",
+    "hypercore-crypto": "^3.6.1",
+    "hyperdrive": "^13.0.1",
     "json-stable-stringify": "^1.1.1",
     "magic-bytes.js": "^1.10.0",
     "map-obj": "^5.0.2",
@@ -197,11 +197,11 @@
     "p-event": "^6.0.1",
     "p-timeout": "^6.1.2",
     "protobufjs": "^7.2.3",
-    "protomux": "^3.4.1",
+    "protomux": "^3.10.1",
     "quickbit-universal": "^2.2.0",
-    "sodium-universal": "^4.0.0",
+    "sodium-universal": "^5.0.1",
     "start-stop-state-machine": "^1.2.0",
-    "streamx": "^2.19.0",
+    "streamx": "^2.23.0",
     "string-timing-safe-equal": "^0.1.0",
     "styled-map-package": "^3.0.0",
     "sub-encoder": "^2.1.1",

src/blob-store/hyperdrive-index.js

@@ -89,6 +89,10 @@ class PretendCorestore {
       // @ts-ignore
       opts.publicKey = opts.key
     }
+    if ('keyPair' in opts) {
+      // @ts-ignore
+      opts.publicKey = opts.keyPair.publicKey
+    }
     if ('publicKey' in opts) {
       // NB! We should always add blobIndex (Hyperbee) cores to the core manager
       // before we use them here. We would only reach the addCore path if the

src/core-manager/index.js

@@ -64,7 +64,7 @@
    * @param {Buffer} options.projectKey 32-byte public key of the project creator core
    * @param {Buffer} [options.projectSecretKey] 32-byte secret key of the project creator core
    * @param {Partial<Record<Namespace, Buffer>>} [options.encryptionKeys] Encryption keys for each namespace
-   * @param {import('hypercore').HypercoreStorage} options.storage Folder to store all hypercore data
+   * @param {string} options.storage Folder to store all hypercore data
    * @param {boolean} [options.autoDownload=true] Immediately start downloading cores - should only be set to false for tests
    * @param {Logger} [options.logger]
    */
@@ -290,20 +290,22 @@
     const existingCore = this.#coreIndex.getByCoreKey(keyPair.publicKey)
     if (existingCore) return existingCore
 
+    const encryptionKey = this.#encryptionKeys[namespace]
+
     const { publicKey: key, secretKey } = keyPair
     const writer = !!secretKey
     const core = this.#corestore.get({
       keyPair,
-      encryptionKey: this.#encryptionKeys[namespace],
+      encryptionKey: encryptionKey ? { key: encryptionKey } : undefined,
     })
     if (this.#autoDownload && namespace !== 'blob') {
       // Blob downloads are managed by BlobStore
       core.download({ start: 0, end: -1 })
     }
     // Every peer adds a listener, so could have many peers
     core.setMaxListeners(0)
-    // @ts-ignore - ensure key is defined before hypercore is ready
-    core.key = key
+    // // @ts-ignore - ensure key is defined before hypercore is ready
+    // core.key = key
     this.#coreIndex.add({ core, key, namespace, writer })
 
     // **Hack** As soon as a peer is added, eagerly send a "want" for the entire

src/datastore/index.js

@@ -146,6 +146,7 @@ export class DataStore extends TypedEmitter {
       )
     }
     const block = encode(doc)
+    console.log({ block })
     // The indexer batch can sometimes complete before the append below
     // resolves, so in the batch function we await any pending appends. We can't
     // know the versionId before the append, because docs can be written in the
@@ -166,6 +167,7 @@ export class DataStore extends TypedEmitter {
     const deferred = pDefer()
     this.#pendingIndex.set(versionId, deferred)
     await deferred.promise
+    console.log('wrote', doc)
     this.#pendingIndex.delete(versionId)
 
     return /** @type {Extract<MapeoDoc, TDoc>} */ (

src/index-writer/index-writer.js

@@ -0,0 +1,106 @@
+import { decode } from '@comapeo/schema'
+import SqliteIndexer from '@mapeo/sqlite-indexer'
+import { getBacklinkTableName } from '../schema/utils.js'
+import { discoveryKey } from 'hypercore-crypto'
+import { Logger } from '../logger.js'
+import { mapDoc } from './map-doc.js'
+import { getWinner } from './get-winner.js'
+/** @import { MapeoDoc } from '@comapeo/schema' */
+/**
+ * @typedef {{ [K in MapeoDoc['schemaName']]?: string[] }} IndexedDocIds
+ */
+/**
+ * @typedef {MapeoDoc['schemaName']} SchemaName
+ */
+
+export class IndexWriter {
+  /** @type {Map<SchemaName, SqliteIndexer>} */
+  #indexers = new Map()
+  #mapDoc
+  #l
+  /**
+   *
+   * @param {object} opts
+   * @param {import('better-sqlite3').Database} opts.sqlite
+   * @param {Iterable<SchemaName>} opts.schemas
+   * @param {Logger} [opts.logger]
+   */
+  constructor({ schemas, sqlite, logger }) {
+    this.#l = Logger.create('indexWriter', logger)
+    this.#mapDoc = mapDoc
+
+    for (const schemaName of schemas) {
+      const indexer = new SqliteIndexer(sqlite, {
+        docTableName: schemaName,
+        backlinkTableName: getBacklinkTableName(schemaName),
+        getWinner,
+      })
+      this.#indexers.set(schemaName, indexer)
+    }
+  }
+
+  /**
+   * @param {import('multi-core-indexer').Entry[]} entries
+   * @returns {Promise<IndexedDocIds>} map of indexed docIds by schemaName
+   */
+  async batch(entries) {
+    // sqlite-indexer is _significantly_ faster when batching even <10 at a
+    // time, so best to queue docs here before calling sliteIndexer.batch()
+    /** @type {Record<string, MapeoDoc[]>} */
+    const queued = {}
+    /** @type {IndexedDocIds} */
+    const indexed = {}
+    for (const { block, key, index } of entries) {
+      /** @type {MapeoDoc} */ let doc
+      try {
+        const version = { coreDiscoveryKey: discoveryKey(key), index }
+        doc = this.#mapDoc(decode(block, version), version)
+      } catch (e) {
+        this.#l.log('Could not decode entry %d of %h', index, key)
+        console.trace('Could not decode entry %d of %h', index, key, e.stack)
+        // Unknown or invalid entry - silently ignore
+        continue
+      }
+      // Don't have an indexer for this type - silently ignore
+      if (!this.#indexers.has(doc.schemaName)) continue
+      if (queued[doc.schemaName]) {
+        queued[doc.schemaName].push(doc)
+        // @ts-expect-error - we know this is defined, TS doesn't
+        indexed[doc.schemaName].push(doc.docId)
+      } else {
+        queued[doc.schemaName] = [doc]
+        indexed[doc.schemaName] = [doc.docId]
+      }
+    }
+    for (const [schemaName, docs] of Object.entries(queued)) {
+      // @ts-expect-error
+      const indexer = this.#indexers.get(schemaName)
+      if (!indexer) continue // Won't happen, but TS doesn't know that
+      indexer.batch(docs)
+      // TODO: selectively turn this on when log level is 'trace' or 'debug'
+      // Otherwise this has a big performance overhead because this is all synchronous
+      // if (this.#l.log.enabled) {
+      //   for (const doc of docs) {
+      //     this.#l.log(
+      //       'Indexed %s %S @ %S',
+      //       doc.schemaName,
+      //       doc.docId,
+      //       doc.versionId
+      //     )
+      //   }
+      // }
+    }
+    return indexed
+  }
+
+  /**
+   * @param {SchemaName} schemaName
+   */
+  async deleteSchema(schemaName) {
+    const indexer = this.#indexers.get(schemaName)
+    if (!indexer) {
+      throw new Error(`IndexWriter doesn't know a schema named "${schemaName}"`)
+    }
+    await indexer.deleteAll()
+  }
+}

src/mapeo-manager.js

@@ -5,7 +5,6 @@
 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 pTimeout from 'p-timeout'
 import { createRequire } from 'module'
@@ -41,7 +40,6 @@
 import { UNIX_EPOCH_DATE } from './constants.js'
 import { openedNoiseSecretStream } from './lib/noise-secret-stream-helpers.js'
 import { omit } from './lib/omit.js'
-import { RandomAccessFilePool } from './core-manager/random-access-file-pool.js'
 import BlobServerPlugin from './fastify-plugins/blobs.js'
 import IconServerPlugin from './fastify-plugins/icons.js'
 import { plugin as MapServerPlugin } from './fastify-plugins/maps.js'
@@ -71,12 +69,6 @@
 
 const CLIENT_SQLITE_FILE_NAME = 'client.db'
 
-// Max file descriptors that RandomAccessFile should use for hypercore storage
-// and index bitfield persistence (used by MultiCoreIndexer). Android has a
-// limit of 1024 per process, so choosing 768 to leave 256 descriptors free for
-// other things e.g. SQLite and other parts of the app.
-const MAX_FILE_DESCRIPTORS = 768
-
 // Prefix names for routes registered with http server
 const BLOBS_PREFIX = 'blobs'
 const ICONS_PREFIX = 'icons'
@@ -133,7 +125,7 @@
    * @param {string} opts.dbFolder Folder for sqlite Dbs. Folder must exist. Use ':memory:' to store everything in-memory
    * @param {string} opts.projectMigrationsFolder path for drizzle migrations folder for project database
    * @param {string} opts.clientMigrationsFolder path for drizzle migrations folder for client database
-   * @param {string | CoreStorage} opts.coreStorage Folder for hypercore storage or a function that returns a RandomAccessStorage instance
+   * @param {string} opts.coreStorage Folder for hypercore storage or a function that returns a RandomAccessStorage instance
    * @param {import('fastify').FastifyInstance} opts.fastify Fastify server instance
    * @param {String} [opts.defaultConfigPath]
    * @param {string} [opts.customMapPath] File path to a locally stored Styled Map Package (SMP).
@@ -164,6 +156,9 @@
     this.#makeWebsocket = makeWebsocket
     const logger = (this.#loggerBase = new Logger({ deviceId: this.#deviceId }))
     this.#l = Logger.create('manager', logger)
+    if (dbFolder === ':memory:') {
+      throw new Error('In-Memory storage not supported as of 5.0.0')
+    }
     this.#dbFolder = dbFolder
     this.#projectMigrationsFolder = projectMigrationsFolder
     const sqlite = new Database(
@@ -206,13 +201,8 @@
       logger,
     })
 
-    if (typeof coreStorage === 'string') {
-      const pool = new RandomAccessFilePool(MAX_FILE_DESCRIPTORS)
-      // @ts-expect-error
-      this.#coreStorage = Hypercore.defaultStorage(coreStorage, { pool })
-    } else {
-      this.#coreStorage = coreStorage
-    }
+    // Must be a string now!
+    this.#coreStorage = coreStorage
 
     this.#fastify = fastify
     this.#fastify.register(BlobServerPlugin, {
@@ -329,11 +319,8 @@
    */
   #projectStorage(projectId) {
     return {
-      dbPath:
-        this.#dbFolder === ':memory:'
-          ? ':memory:'
-          : path.join(this.#dbFolder, projectId + '.db'),
-      coreStorage: (name) => this.#coreStorage(path.join(projectId, name)),
+      dbPath: path.join(this.#dbFolder, projectId + '.db'),
+      coreStorage: path.join(this.#coreStorage, projectId),
     }
   }
 

src/mapeo-project.js

@@ -10,6 +10,7 @@
 import mime from 'mime/lite'
 // @ts-expect-error
 import { Readable, pipelinePromise } from 'streamx'
+import RandomAccessFile from 'random-access-file'
 
 import { NAMESPACES, NAMESPACE_SCHEMAS, UNIX_EPOCH_DATE } from './constants.js'
 import { CoreManager } from './core-manager/index.js'
@@ -220,13 +221,16 @@
 
     ///////// 3. Setup random-access-storage functions
 
-    /** @type {ConstructorParameters<typeof CoreManager>[0]['storage']} */
-    const coreManagerStorage = (name) =>
-      coreStorage(path.join(CORESTORE_STORAGE_FOLDER_NAME, name))
+    const coreManagerStorage = path.join(
+      coreStorage,
+      CORESTORE_STORAGE_FOLDER_NAME
+    )
 
     /** @type {ConstructorParameters<typeof DataStore>[0]['storage']} */
     const indexerStorage = (name) =>
-      coreStorage(path.join(INDEXER_STORAGE_FOLDER_NAME, name))
+      new RandomAccessFile(
+        path.join(coreStorage, INDEXER_STORAGE_FOLDER_NAME, name)
+      )
 
     ///////// 4. Create instances