WIP add coreOwnership

custom getWinner and initial mapDoc

Add verifyCoreOwnership function

Add mapAndValidateCoreOwnership

move some code around

add mapDoc and getWinner options to indexWriter

Add tests & small fix

Fix types and use defaultGetWinner from @mapeo/sqlite-indexer

package.json

@@ -95,9 +95,9 @@
     "@digidem/types": "^2.0.0",
     "@fastify/type-provider-typebox": "^3.3.0",
     "@hyperswarm/secret-stream": "^6.1.2",
-    "@mapeo/crypto": "^1.0.0-alpha.5",
+    "@mapeo/crypto": "^1.0.0-alpha.7",
     "@mapeo/schema": "^3.0.0-next.8",
-    "@mapeo/sqlite-indexer": "^1.0.0-alpha.5",
+    "@mapeo/sqlite-indexer": "^1.0.0-alpha.6",
     "@sinclair/typebox": "^0.29.6",
     "b4a": "^1.6.3",
     "base32.js": "^0.1.0",

src/core-ownership.js

@@ -0,0 +1,96 @@
+import { verifySignature } from '@mapeo/crypto'
+import { NAMESPACES } from './core-manager/index.js'
+import { parseVersionId } from '@mapeo/schema'
+import { defaultGetWinner } from '@mapeo/sqlite-indexer'
+import assert from 'node:assert'
+import sodium from 'sodium-universal'
+
+/**
+ * @typedef {Extract<ReturnType<import('@mapeo/schema').decode>, { schemaName: 'coreOwnership' }>} CoreOwnershipWithSignatures
+ */
+
+/**
+ * - Validate that the doc is written to the core identified by doc.authCoreId
+ * - Verify the signatures
+ * - Remove the signatures (we don't add them to the indexer)
+ * - Set doc.links to an empty array - this forces the indexer to treat every
+ *   document as a fork, so getWinner is called for every doc, which resolves to
+ *   the doc with the lowest index (e.g. the first)
+ *
+ * @param {CoreOwnershipWithSignatures} doc
+ * @param {import('@mapeo/schema').VersionIdObject} version
+ * @returns {import('@mapeo/schema').CoreOwnership}
+ */
+export function mapAndValidateCoreOwnership(doc, { coreKey }) {
+  if (doc.authCoreId !== coreKey.toString('hex')) {
+    throw new Error('Invalid coreOwnership record: mismatched authCoreId')
+  }
+  if (!verifyCoreOwnership(doc)) {
+    throw new Error('Invalid coreOwnership record: signatures are invalid')
+  }
+  // eslint-disable-next-line no-unused-vars
+  const { identitySignature, coreSignatures, ...docWithoutSignatures } = doc
+  docWithoutSignatures.links = []
+  return docWithoutSignatures
+}
+
+/**
+ * Verify the signatures of a coreOwnership record, which verify that the device
+ * with the identityKey matching the docIds does own (e.g. can write to) cores
+ * with the given core IDs
+ *
+ * @param {CoreOwnershipWithSignatures} doc
+ * @returns {boolean}
+ */
+function verifyCoreOwnership(doc) {
+  const { coreSignatures, identitySignature } = doc
+  for (const namespace of NAMESPACES) {
+    const signature = coreSignatures[namespace]
+    const coreKey = Buffer.from(doc[`${namespace}CoreId`], 'hex')
+    assert.equal(
+      signature.length,
+      sodium.crypto_sign_BYTES,
+      'Invalid core ownership signature'
+    )
+    assert.equal(
+      coreKey.length,
+      sodium.crypto_sign_PUBLICKEYBYTES,
+      'Invalid core ownership coreId'
+    )
+    const isValidSignature = verifySignature(coreKey, signature, coreKey)
+    if (!isValidSignature) return false
+  }
+  const identityPublicKey = Buffer.from(doc.docId, 'hex')
+  assert.equal(identitySignature.length, sodium.crypto_sign_BYTES)
+  assert.equal(identityPublicKey.length, sodium.crypto_sign_PUBLICKEYBYTES)
+  const isValidIdentitySignature = verifySignature(
+    identityPublicKey,
+    identitySignature,
+    identityPublicKey
+  )
+  if (!isValidIdentitySignature) return false
+  return true
+}
+
+/**
+ * For coreOwnership records, we only trust the first record written to the core.
+ *
+ * @type {NonNullable<ConstructorParameters<typeof import('./index-writer/index.js').IndexWriter>[0]['getWinner']>}
+ */
+export function getWinner(docA, docB) {
+  if (
+    'schemaName' in docA &&
+    docA.schemaName === 'coreOwnership' &&
+    'schemaName' in docB &&
+    docB.schemaName === 'coreOwnership'
+  ) {
+    // Assumes docA and docB have same coreKey, so we choose the first one
+    // written to the core
+    const docAindex = parseVersionId(docA.versionId).index
+    const docBindex = parseVersionId(docB.versionId).index
+    if (docAindex < docBindex) return docA
+    return docB
+  } else {
+    return defaultGetWinner(docA, docB)
+  }
+}

src/index-writer/index.js

@@ -9,20 +9,27 @@ import { getBacklinkTableName } from '../schema/utils.js'
 /**
  * @typedef {import('@mapeo/schema').MapeoDoc} MapeoDoc
  */
+/**
+ * @typedef {ReturnType<import('@mapeo/schema').decode>} MapeoDocInternal
+ */
 
 /**
  * @template {MapeoDocTables} [TTables=MapeoDocTables]
  */
 export class IndexWriter {
   /** @type {Map<TTables['_']['name'], SqliteIndexer>} */
   #indexers = new Map()
+  #mapDoc
   /**
    *
    * @param {object} opts
    * @param {import('better-sqlite3').Database} opts.sqlite
    * @param {TTables[]} opts.tables
+   * @param {(doc: MapeoDocInternal, version: import('@mapeo/schema').VersionIdObject) => MapeoDoc} [opts.mapDoc] optionally transform a document prior to indexing. Can also validate, if an error is thrown then the document will not be indexed
+   * @param {typeof import('@mapeo/sqlite-indexer').defaultGetWinner} [opts.getWinner] custom function to determine the "winner" of two forked documents. Defaults to choosing the document with the most recent `updatedAt`
    */
-  constructor({ tables, sqlite }) {
+  constructor({ tables, sqlite, mapDoc = (d) => d, getWinner }) {
+    this.#mapDoc = mapDoc
     for (const table of tables) {
       const config = getTableConfig(table)
       const schemaName = /** @type {(typeof table)['_']['name']} */ (
@@ -31,6 +38,7 @@ export class IndexWriter {
       const indexer = new SqliteIndexer(sqlite, {
         docTableName: config.name,
         backlinkTableName: getBacklinkTableName(config.name),
+        getWinner,
       })
       this.#indexers.set(schemaName, indexer)
     }
@@ -51,9 +59,10 @@ export class IndexWriter {
     const queued = {}
     for (const { block, key, index } of entries) {
       try {
-        var doc = decode(block, { coreKey: key, index })
+        const version = { coreKey: key, index }
+        var doc = this.#mapDoc(decode(block, version), version)
       } catch (e) {
-        // Unknown entry - silently ignore
+        // Unknown or invalid entry - silently ignore
         continue
       }
       if (queued[doc.schemaName]) {

src/mapeo-project.js

@@ -14,6 +14,7 @@ import RAM from 'random-access-memory'
 import Database from 'better-sqlite3'
 import path from 'path'
 import { RandomAccessFilePool } from './core-manager/random-access-file-pool.js'
+import { getWinner, mapAndValidateCoreOwnership } from './core-ownership.js'
 import { valueOf } from './utils.js'
 
 /** @typedef {Omit<import('@mapeo/schema').ProjectValue, 'schemaName'>} EditableProjectSettings */
@@ -97,6 +98,14 @@ export class MapeoProject {
     const indexWriter = new IndexWriter({
       tables: [observationTable, presetTable, fieldTable],
       sqlite,
+      getWinner,
+      mapDoc: (doc, version) => {
+        if (doc.schemaName === 'coreOwnership') {
+          return mapAndValidateCoreOwnership(doc, version)
+        } else {
+          return doc
+        }
+      },
     })
     this.#dataStores = {
       config: new DataStore({