Updates

lib/mesh-registry.js

@@ -3,6 +3,7 @@
 /**
  * Mesh region registry (ADR-0007).
  * Autobase multi-writer log + Hyperbee view of regions and spatial cells.
+ * Supports open-membership (optimistic) mode for the experimental global mesh.
  */
 
 const path = require('path')
@@ -18,6 +19,9 @@ const { encodeKey, decodeKey } = require('./invite')
 
 const CELL = 512 // blocks per spatial index cell
 
+/** @type {Map<string, MeshRegistry>} */
+const openMeshStores = new Map()
+
 function ensureDir(dir) {
   fs.mkdirSync(dir, { recursive: true })
 }
@@ -56,53 +60,6 @@ function openView(store) {
   })
 }
 
-async function applyOps(nodes, view, host) {
-  for (const node of nodes) {
-    let op = node.value
-    if (typeof op === 'string') {
-      try {
-        op = JSON.parse(op)
-      } catch {
-        continue
-      }
-    }
-    if (!op || typeof op !== 'object') continue
-
-    if (op.addWriter) {
-      const writerKey =
-        typeof op.addWriter === 'string' ? decodeKeyMaybe(op.addWriter) : b4a.from(op.addWriter)
-      await host.addWriter(writerKey, { indexer: true })
-      continue
-    }
-
-    if (op.type === 'enroll' && op.region) {
-      const r = op.region
-      const batch = view.batch()
-      await batch.put('region:' + r.regionId, r)
-      if (r.bounds) {
-        for (const cell of cellsCovering(r.bounds)) {
-          await batch.put('cell:' + cell, r.regionId)
-        }
-      }
-      await batch.flush()
-      continue
-    }
-
-    if (op.type === 'unenroll' && op.regionId) {
-      const batch = view.batch()
-      const existing = await view.get('region:' + op.regionId)
-      if (existing && existing.value && existing.value.bounds) {
-        for (const cell of cellsCovering(existing.value.bounds)) {
-          const cur = await view.get('cell:' + cell)
-          if (cur && cur.value === op.regionId) await batch.del('cell:' + cell)
-        }
-      }
-      await batch.del('region:' + op.regionId)
-      await batch.flush()
-    }
-  }
-}
-
 function decodeKeyMaybe(s) {
   try {
     return decodeKey(s)
@@ -115,6 +72,67 @@ function decodeKeyMaybe(s) {
   }
 }
 
+/**
+ * @param {boolean} openMembership
+ */
+function makeApply(openMembership) {
+  return async function applyOps(nodes, view, host) {
+    for (const node of nodes) {
+      let op = node.value
+      if (typeof op === 'string') {
+        try {
+          op = JSON.parse(op)
+        } catch {
+          continue
+        }
+      }
+      if (!op || typeof op !== 'object') continue
+
+      if (openMembership && node.from && node.from.key) {
+        try {
+          await host.ackWriter(node.from.key)
+          await host.addWriter(node.from.key, { indexer: true })
+        } catch {
+          /* already admitted or host unavailable */
+        }
+      }
+
+      if (op.addWriter) {
+        const writerKey =
+          typeof op.addWriter === 'string' ? decodeKeyMaybe(op.addWriter) : b4a.from(op.addWriter)
+        await host.addWriter(writerKey, { indexer: true })
+        continue
+      }
+
+      if (op.type === 'enroll' && op.region) {
+        const r = op.region
+        const batch = view.batch()
+        await batch.put('region:' + r.regionId, r)
+        if (r.bounds) {
+          for (const cell of cellsCovering(r.bounds)) {
+            await batch.put('cell:' + cell, r.regionId)
+          }
+        }
+        await batch.flush()
+        continue
+      }
+
+      if (op.type === 'unenroll' && op.regionId) {
+        const batch = view.batch()
+        const existing = await view.get('region:' + op.regionId)
+        if (existing && existing.value && existing.value.bounds) {
+          for (const cell of cellsCovering(existing.value.bounds)) {
+            const cur = await view.get('cell:' + cell)
+            if (cur && cur.value === op.regionId) await batch.del('cell:' + cell)
+          }
+        }
+        await batch.del('region:' + op.regionId)
+        await batch.flush()
+      }
+    }
+  }
+}
+
 async function closeView(view) {
   await view.close()
 }
@@ -126,6 +144,10 @@ class MeshRegistry extends ReadyResource {
    * @param {string} [opts.meshId] local folder id (default from key)
    * @param {Buffer|string|null} [opts.bootstrap] mesh key to join; null = create
    * @param {string} [opts.name]
+   * @param {boolean} [opts.openMembership] optimistic open writers (global mesh)
+   * @param {Buffer} [opts.topic] Hyperswarm topic override
+   * @param {{ publicKey: Buffer, secretKey: Buffer }} [opts.keyPair] Autobase bootstrap keyPair
+   * @param {boolean} [opts.global] mark as global mesh in meta
    */
   constructor(opts) {
     super()
@@ -137,6 +159,12 @@ class MeshRegistry extends ReadyResource {
       : null
     this.name = opts.name || 'mesh'
     this.meshId = opts.meshId || null
+    this.openMembership = Boolean(opts.openMembership)
+    this.topic = opts.topic ? b4a.from(opts.topic) : null
+    this.keyPair = opts.keyPair || null
+    this.global = Boolean(opts.global)
+    /** When false, skip Hyperswarm (unit tests that pipe Autobase directly). */
+    this.enableSwarm = opts.swarm !== false
     this.store = null
     this.base = null
     this.swarm = null
@@ -159,6 +187,30 @@ class MeshRegistry extends ReadyResource {
     return this.base && this.base.local && this.base.local.key
   }
 
+  async _append(value) {
+    if (this.writable) {
+      await this.base.append(value)
+    } else if (this.openMembership) {
+      await this.base.append(value, { optimistic: true })
+    } else {
+      throw new Error('Not a mesh writer')
+    }
+    // Don't block forever on DHT convergence — local append is enough for enroll UX
+    try {
+      await Promise.race([
+        this.base.update(),
+        new Promise((resolve) => setTimeout(resolve, 2500))
+      ])
+    } catch {
+      /* ignore update errors; view may still catch up via swarm */
+    }
+  }
+
+  /** True if this peer can enroll / append (writer or open global mesh). */
+  get canEnroll() {
+    return this.writable || this.openMembership
+  }
+
   async _open() {
     const id =
       this.meshId ||
@@ -167,15 +219,41 @@ class MeshRegistry extends ReadyResource {
     const dir = meshStorePath(this.storageDir, id)
     ensureDir(dir)
 
-    this.store = new Corestore(path.join(dir, 'corestore'))
-    this.base = new Autobase(this.store, this.bootstrap, {
+    const storePath = path.resolve(dir, 'corestore')
+    const existing = openMeshStores.get(storePath)
+    if (existing && existing !== this) {
+      throw new Error(
+        'Mesh store already open in this process. Close the other mesh before opening again.'
+      )
+    }
+    openMeshStores.set(storePath, this)
+    this._storePath = storePath
+
+    this.store = new Corestore(storePath)
+    const handlers = {
       open: openView,
-      apply: applyOps,
+      apply: makeApply(this.openMembership),
       close: closeView,
       valueEncoding: 'json',
-      ackInterval: 1000
-    })
-    await this.base.ready()
+      ackInterval: 1000,
+      optimistic: this.openMembership
+    }
+    if (this.keyPair) handlers.keyPair = this.keyPair
+
+    try {
+      this.base = new Autobase(this.store, this.bootstrap, handlers)
+      await this.base.ready()
+    } catch (err) {
+      openMeshStores.delete(storePath)
+      this._storePath = null
+      try {
+        await this.store.close()
+      } catch {
+        /* ignore */
+      }
+      this.store = null
+      throw err
+    }
 
     // Persist mesh meta
     const metaPath = path.join(dir, 'meta.json')
@@ -187,6 +265,7 @@ class MeshRegistry extends ReadyResource {
             meshId: this.meshId,
             key: this.keyZ32,
             name: this.name,
+            global: this.global,
             createdAt: Date.now()
           },
           null,
@@ -196,26 +275,29 @@ class MeshRegistry extends ReadyResource {
     }
 
     // Replicate
-    this.swarm = new Hyperswarm()
-    this.swarm.on('connection', (conn) => {
-      this.base.replicate(conn)
-    })
-    const topic = crypto.discoveryKey(this.base.key)
-    this._discovery = this.swarm.join(topic, { server: true, client: true })
-    await this.swarm.flush()
+    if (this.enableSwarm) {
+      this.swarm = new Hyperswarm()
+      this.swarm.on('connection', (conn) => {
+        this.base.replicate(conn)
+      })
+      const topic = this.topic || crypto.discoveryKey(this.base.key)
+      this._discovery = this.swarm.join(topic, { server: true, client: true })
+      await this.swarm.flush()
+    }
   }
 
   /**
-   * Admit another peer's writer key (must be writable).
+   * Admit another peer's writer key (must be writable or open-membership).
    * @param {Buffer|string} writerKey
    */
   async addWriter(writerKey) {
-    if (!this.writable) throw new Error('Not a mesh writer; cannot addWriter')
+    if (!this.writable && !this.openMembership) {
+      throw new Error('Not a mesh writer; cannot addWriter')
+    }
     const key = typeof writerKey === 'string' ? decodeKeyMaybe(writerKey) : b4a.from(writerKey)
-    await this.base.append({
+    await this._append({
       addWriter: encodeKey(key)
     })
-    await this.base.update()
   }
 
   /**
@@ -223,7 +305,9 @@ class MeshRegistry extends ReadyResource {
    * @param {object} region RegionRecord
    */
   async enroll(region) {
-    if (!this.writable) throw new Error('Not a mesh writer; cannot enroll')
+    if (!this.writable && !this.openMembership) {
+      throw new Error('Not a mesh writer; cannot enroll')
+    }
     if (!region.regionId || !region.worldKey || !region.bounds) {
       throw new Error('region requires regionId, worldKey, bounds')
     }
@@ -240,16 +324,16 @@ class MeshRegistry extends ReadyResource {
       name: region.name || region.regionId,
       updatedAt: Date.now()
     }
-    await this.base.append({ type: 'enroll', region: record })
-    await this.base.update()
+    await this._append({ type: 'enroll', region: record })
     this.emit('enroll', record)
     return record
   }
 
   async unenroll(regionId) {
-    if (!this.writable) throw new Error('Not a mesh writer')
-    await this.base.append({ type: 'unenroll', regionId })
-    await this.base.update()
+    if (!this.writable && !this.openMembership) {
+      throw new Error('Not a mesh writer')
+    }
+    await this._append({ type: 'unenroll', regionId })
   }
 
   /**
@@ -293,6 +377,12 @@ class MeshRegistry extends ReadyResource {
   }
 
   async _close() {
+    if (this._storePath) {
+      if (openMeshStores.get(this._storePath) === this) {
+        openMeshStores.delete(this._storePath)
+      }
+      this._storePath = null
+    }
     if (this._discovery) {
       try {
         await this._discovery.destroy()