449 lines
12 KiB
JavaScript
449 lines
12 KiB
JavaScript
'use strict'
|
|
|
|
/**
|
|
* 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')
|
|
const fs = require('fs')
|
|
const ReadyResource = require('ready-resource')
|
|
const Corestore = require('corestore')
|
|
const Autobase = require('autobase')
|
|
const Hyperbee = require('hyperbee')
|
|
const Hyperswarm = require('hyperswarm')
|
|
const b4a = require('b4a')
|
|
const crypto = require('hypercore-crypto')
|
|
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 })
|
|
}
|
|
|
|
function meshStorePath(storageDir, meshId) {
|
|
return path.join(storageDir, 'mesh', meshId)
|
|
}
|
|
|
|
/**
|
|
* @param {{ minX:number, maxX:number, minZ:number, maxZ:number }} bounds
|
|
* @returns {string[]}
|
|
*/
|
|
function cellsCovering(bounds) {
|
|
const cells = []
|
|
const x0 = Math.floor(bounds.minX / CELL)
|
|
const x1 = Math.floor(bounds.maxX / CELL)
|
|
const z0 = Math.floor(bounds.minZ / CELL)
|
|
const z1 = Math.floor(bounds.maxZ / CELL)
|
|
for (let gx = x0; gx <= x1; gx++) {
|
|
for (let gz = z0; gz <= z1; gz++) {
|
|
cells.push(`${gx}:${gz}`)
|
|
}
|
|
}
|
|
return cells
|
|
}
|
|
|
|
function cellAt(x, z) {
|
|
return `${Math.floor(x / CELL)}:${Math.floor(z / CELL)}`
|
|
}
|
|
|
|
function openView(store) {
|
|
const core = store.get('mesh-regions-view')
|
|
return new Hyperbee(core, {
|
|
keyEncoding: 'utf-8',
|
|
valueEncoding: 'json'
|
|
})
|
|
}
|
|
|
|
function decodeKeyMaybe(s) {
|
|
try {
|
|
return decodeKey(s)
|
|
} catch {
|
|
try {
|
|
return b4a.from(s, 'hex')
|
|
} catch {
|
|
return b4a.from(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()
|
|
}
|
|
|
|
class MeshRegistry extends ReadyResource {
|
|
/**
|
|
* @param {object} opts
|
|
* @param {string} opts.storageDir app storage root
|
|
* @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()
|
|
this.storageDir = opts.storageDir
|
|
this.bootstrap = opts.bootstrap
|
|
? typeof opts.bootstrap === 'string'
|
|
? decodeKeyMaybe(opts.bootstrap)
|
|
: b4a.from(opts.bootstrap)
|
|
: 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
|
|
this._discovery = null
|
|
}
|
|
|
|
get key() {
|
|
return this.base && this.base.key
|
|
}
|
|
|
|
get keyZ32() {
|
|
return this.key ? encodeKey(this.key) : null
|
|
}
|
|
|
|
get writable() {
|
|
return !!(this.base && this.base.writable)
|
|
}
|
|
|
|
get localWriterKey() {
|
|
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 ||
|
|
(this.bootstrap ? encodeKey(this.bootstrap).slice(0, 16) : 'local-' + Date.now().toString(36))
|
|
this.meshId = id
|
|
const dir = meshStorePath(this.storageDir, id)
|
|
ensureDir(dir)
|
|
|
|
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: makeApply(this.openMembership),
|
|
close: closeView,
|
|
valueEncoding: 'json',
|
|
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')
|
|
if (!fs.existsSync(metaPath)) {
|
|
fs.writeFileSync(
|
|
metaPath,
|
|
JSON.stringify(
|
|
{
|
|
meshId: this.meshId,
|
|
key: this.keyZ32,
|
|
name: this.name,
|
|
global: this.global,
|
|
createdAt: Date.now()
|
|
},
|
|
null,
|
|
2
|
|
)
|
|
)
|
|
}
|
|
|
|
// Replicate
|
|
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 or open-membership).
|
|
* @param {Buffer|string} writerKey
|
|
*/
|
|
async addWriter(writerKey) {
|
|
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._append({
|
|
addWriter: encodeKey(key)
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Enroll a region into the mesh.
|
|
* @param {object} region RegionRecord
|
|
*/
|
|
async enroll(region) {
|
|
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')
|
|
}
|
|
const record = {
|
|
regionId: region.regionId,
|
|
ownerKey: region.ownerKey || null,
|
|
worldKey: region.worldKey,
|
|
cap: region.cap || null, // mesh-shared tunnel capability (z32)
|
|
offset: region.offset || { x: 0, z: 0 },
|
|
bounds: region.bounds,
|
|
seed: region.seed != null ? region.seed : null,
|
|
mcVersion: region.mcVersion || null,
|
|
portalPolicy: region.portalPolicy || 'teleport',
|
|
name: region.name || region.regionId,
|
|
updatedAt: Date.now()
|
|
}
|
|
await this._append({ type: 'enroll', region: record })
|
|
this.emit('enroll', record)
|
|
return record
|
|
}
|
|
|
|
async unenroll(regionId) {
|
|
if (!this.writable && !this.openMembership) {
|
|
throw new Error('Not a mesh writer')
|
|
}
|
|
await this._append({ type: 'unenroll', regionId })
|
|
}
|
|
|
|
/**
|
|
* @returns {Promise<object[]>}
|
|
*/
|
|
async listRegions() {
|
|
await this.base.update()
|
|
const view = this.base.view
|
|
const out = []
|
|
for await (const entry of view.createReadStream({ gte: 'region:', lt: 'region;' })) {
|
|
if (entry.value) out.push(entry.value)
|
|
}
|
|
return out
|
|
}
|
|
|
|
/**
|
|
* Resolve region owning world coordinates (x,z).
|
|
* @param {number} x
|
|
* @param {number} z
|
|
*/
|
|
async regionAt(x, z) {
|
|
await this.base.update()
|
|
const cell = cellAt(x, z)
|
|
const hit = await this.base.view.get('cell:' + cell)
|
|
if (!hit || !hit.value) {
|
|
// fallback scan (sparse enrollments)
|
|
const all = await this.listRegions()
|
|
return (
|
|
all.find(
|
|
(r) =>
|
|
r.bounds &&
|
|
x >= r.bounds.minX &&
|
|
x <= r.bounds.maxX &&
|
|
z >= r.bounds.minZ &&
|
|
z <= r.bounds.maxZ
|
|
) || null
|
|
)
|
|
}
|
|
const reg = await this.base.view.get('region:' + hit.value)
|
|
return reg ? reg.value : null
|
|
}
|
|
|
|
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()
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
this._discovery = null
|
|
}
|
|
if (this.swarm) {
|
|
try {
|
|
await this.swarm.destroy()
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
this.swarm = null
|
|
}
|
|
if (this.base) {
|
|
try {
|
|
await this.base.close()
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
this.base = null
|
|
}
|
|
if (this.store) {
|
|
try {
|
|
await this.store.close()
|
|
} catch {
|
|
/* ignore */
|
|
}
|
|
this.store = null
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* List locally known meshes (from disk).
|
|
* @param {string} storageDir
|
|
*/
|
|
function listLocalMeshes(storageDir) {
|
|
const root = path.join(storageDir, 'mesh')
|
|
if (!fs.existsSync(root)) return []
|
|
const out = []
|
|
for (const name of fs.readdirSync(root)) {
|
|
const meta = path.join(root, name, 'meta.json')
|
|
if (!fs.existsSync(meta)) continue
|
|
try {
|
|
out.push(JSON.parse(fs.readFileSync(meta, 'utf8')))
|
|
} catch {
|
|
/* skip */
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
module.exports = {
|
|
MeshRegistry,
|
|
cellsCovering,
|
|
cellAt,
|
|
CELL,
|
|
listLocalMeshes,
|
|
meshStorePath
|
|
}
|