/**
* metadata-base.js — Autobase "smart contract" for extensible metadata
*
* A dedicated Autobase instance for metadata that keet-core's contract
* doesn't handle (room topics, discoverability, bridges, governance, etc.)
*
* Design principles:
* - Namespace-based operations: add new namespaces without changing apply
* - Field-level LWW: per-field timestamps for granular conflict resolution
* - Forward compatibility: unknown namespaces stored verbatim in default case
* - Secondary indexes: topic search via Hyperbee range queries
* - Single instance per identity (not per-room)
*
* The apply function IS the smart contract. Every peer runs it on the same
* inputs and gets the same state via Autobase linearization.
*/
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hyperbee from 'hyperbee'
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import c from 'compact-encoding'
import fs from 'bare-fs'
import path from 'bare-path'
export class MetadataBase {
/**
* @param {string} storagePath - Per-identity storage directory
* @param {object} [opts]
* @param {Buffer} [opts.bootstrapKey] - Existing Autobase key to join
* @param {boolean} [opts.network=true] - Enable Hyperswarm networking
*/
constructor (storagePath, opts = {}) {
this._storagePath = path.join(storagePath, 'metadata')
this._keyPath = path.join(storagePath, 'metadata-key')
this._network = opts.network !== false
this._bootstrapKey = opts.bootstrapKey || null
this.store = null
this.base = null
this.swarm = null
this._ready = false
this._closing = false
}
async ready () {
if (this._ready) return
if (this._closing) throw new Error('MetadataBase is closing')
// Create corestore for this metadata instance
this.store = new Corestore(this._storagePath)
await this.store.ready()
// Load or create bootstrap key
const bootstrapKey = this._bootstrapKey || this._loadKey()
// Create Autobase with metadata contract
this.base = new Autobase(this.store, bootstrapKey, {
open: this._open.bind(this),
apply: this._apply.bind(this),
close: this._close.bind(this),
valueEncoding: 'json',
ackInterval: 5000
})
await this.base.ready()
// Persist bootstrap key for subsequent runs
this._saveKey(this.base.key)
// Network replication + writer exchange
if (this._network) {
this.swarm = new Hyperswarm()
this.swarm.on('connection', (socket) => {
console.log('[metadata] swarm connection from', socket.remotePublicKey?.toString('hex')?.slice(0, 12) || 'unknown')
this.store.replicate(socket)
this._exchangeWriters(socket)
})
const dkey = this.base.discoveryKey
console.log('[metadata] joining swarm, discoveryKey:', dkey?.toString('hex')?.slice(0, 16) || 'none')
this.swarm.join(dkey)
}
this._ready = true
}
// ── View lifecycle (called by Autobase) ────────────────────
_open (store) {
return new Hyperbee(store.get('view'), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
})
}
/**
* The smart contract. Deterministic state transitions.
* Add new `case` branches as features grow. The `default` case
* stores unknown namespaces verbatim for forward compatibility.
*/
async _apply (nodes, view, base) {
const batch = view.batch({ update: false })
for (const node of nodes) {
const op = node.value
const writer = node.from?.key?.toString('hex')?.slice(0, 16) || 'local'
switch (op.ns) {
// ── Room metadata ────────────────────────────
case 'room': {
const key = `room:${op.id}`
const prev = await view.get(key)
const merged = mergeFields(prev?.value, op.fields, op.ts)
await batch.put(key, { ...merged, _by: writer, _ts: op.ts })
// Secondary index: topics for search
if (op.fields.topics && Array.isArray(op.fields.topics)) {
// Clear old topic entries for this room
const existing = prev?.value?.topics
if (existing && Array.isArray(existing)) {
for (const topic of existing) {
await batch.del(`idx:topic:${topic}:${op.id}`)
}
}
// Write new topic entries
for (const topic of op.fields.topics) {
await batch.put(`idx:topic:${topic}:${op.id}`, {
roomId: op.id, topic, _ts: op.ts
})
}
}
break
}
// ── Bridge configurations ────────────────────
case 'bridge': {
await batch.put(`bridge:${op.roomId}:${op.platform}`, {
...op.config,
_by: writer,
_ts: op.ts
})
break
}
// ── Governance rules ─────────────────────────
case 'gov': {
const key = `gov:${op.roomId}:${op.rule}`
const prev = await view.get(key)
const merged = mergeFields(prev?.value, op.params, op.ts)
await batch.put(key, { ...merged, _by: writer, _ts: op.ts })
break
}
// ── Delete operation ─────────────────────────
case '_del': {
if (op.key) {
await batch.del(op.key)
}
break
}
// ── Writer management (via Autobase apply calls) ─
case '_addWriter': {
if (op.key) {
const keyBuf = typeof op.key === 'string' ? Buffer.from(op.key, 'hex') : op.key
await base.addWriter(keyBuf, { isIndexer: op.indexer !== false })
}
break
}
case '_removeWriter': {
if (op.key) {
const keyBuf = typeof op.key === 'string' ? Buffer.from(op.key, 'hex') : op.key
await base.removeWriter(keyBuf)
}
break
}
// ── Forward compatibility: unknown namespaces ─
default: {
if (op.ns && op.id) {
const key = `${op.ns}:${op.id}`
const prev = await view.get(key)
const merged = mergeFields(prev?.value, op.fields || op.value || {}, op.ts)
await batch.put(key, {
...merged,
_by: writer,
_ts: op.ts,
_ns: op.ns
})
}
break
}
}
}
await batch.flush()
}
async _close (view) {
if (view) await view.close()
}
// ── Writer exchange (Protomux) ───────────────────────────────
/**
* Exchange local writer keys with a connected peer via Protomux.
* Both sides send their local writer key and add the remote as an indexer.
* This enables multi-device Autobase sync for the same identity.
*/
_exchangeWriters (socket) {
const mux = Protomux.from(socket)
const channel = mux.createChannel({
protocol: 'metadata-base/writer-exchange',
id: this.base.key
})
const writerMsg = channel.addMessage({
encoding: c.buffer,
onmessage: async (remoteKey) => {
if (remoteKey.length !== 32) return
console.log('[metadata] received writer key:', remoteKey.toString('hex').slice(0, 12))
try {
// Add remote writer via the Autobase contract (processed in _apply)
await this.base.append({
ns: '_addWriter',
key: remoteKey.toString('hex'),
indexer: true,
ts: Date.now()
})
await this.base.update()
console.log('[metadata] added remote writer:', remoteKey.toString('hex').slice(0, 12))
} catch (err) {
console.log('[metadata] writer exchange note:', err.message)
}
}
})
channel.open()
console.log('[metadata] sending local writer key:', this.base.local.key.toString('hex').slice(0, 12))
writerMsg.send(this.base.local.key)
}
// ── Public API ─────────────────────────────────────────────
/**
* Get metadata by namespace and id.
* @param {string} ns - Namespace ('room', 'bridge', 'gov', etc.)
* @param {string} id - Entity ID (roomId, etc.)
* @returns {object|null} Metadata value or null
*/
async get (ns, id) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.update()
const key = `${ns}:${id}`
const entry = await this.base.view.get(key)
return entry ? entry.value : null
}
/**
* Set metadata fields. Merges with existing (field-level LWW in apply).
* @param {string} ns - Namespace
* @param {string} id - Entity ID
* @param {object} fields - Fields to set
*/
async set (ns, id, fields) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.append({
ns,
id,
fields,
ts: Date.now()
})
await this.base.update()
}
/**
* Set with extra properties (for bridge, gov, etc.).
* @param {object} op - Full operation object (must include ns, ts)
*/
async append (op) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.append({ ...op, ts: op.ts || Date.now() })
await this.base.update()
}
/**
* Delete a key from the view.
* @param {string} key - Full key (e.g. 'room:abc', 'bridge:abc:irc')
*/
async del (key) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.append({ ns: '_del', key, ts: Date.now() })
await this.base.update()
}
/**
* Get all entries for a namespace.
* @param {string} ns - Namespace prefix
* @returns {object} Map of id → value
*/
async getAll (ns) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.update()
const results = {}
const prefix = `${ns}:`
const stream = this.base.view.createReadStream({
gte: prefix,
lt: prefix + '~'
})
for await (const entry of stream) {
const id = entry.key.slice(prefix.length)
results[id] = entry.value
}
return results
}
/**
* Search by topic (uses secondary index).
* @param {string} topic - Topic to search for
* @returns {Array<{roomId, topic}>}
*/
async searchByTopic (topic) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.update()
const results = []
const prefix = `idx:topic:${topic}:`
const stream = this.base.view.createReadStream({
gte: prefix,
lt: prefix + '~'
})
for await (const entry of stream) {
results.push(entry.value)
}
return results
}
/**
* List all entries matching a key prefix.
* @param {string} prefix - Key prefix (e.g. 'bridge:roomId:')
* @returns {Array<{key, value}>}
*/
async list (prefix) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.update()
const results = []
const stream = this.base.view.createReadStream({
gte: prefix,
lt: prefix + '~'
})
for await (const entry of stream) {
results.push({ key: entry.key, value: entry.value })
}
return results
}
// ── Writer management ──────────────────────────────────────
/**
* Add a writer to this metadata Autobase.
* Appends an _addWriter operation processed by the apply contract.
* @param {Buffer} key - Writer's public key
* @param {object} [opts] - Options (e.g. { indexer: true })
*/
async addWriter (key, opts = {}) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.append({
ns: '_addWriter',
key: key.toString('hex'),
indexer: opts.indexer !== false,
ts: Date.now()
})
await this.base.update()
}
/**
* Remove a writer from this metadata Autobase.
* Appends a _removeWriter operation processed by the apply contract.
* @param {Buffer} key - Writer's public key
*/
async removeWriter (key) {
if (!this._ready) throw new Error('MetadataBase not ready')
await this.base.append({
ns: '_removeWriter',
key: key.toString('hex'),
ts: Date.now()
})
await this.base.update()
}
/**
* Get the local writer's public key.
* @returns {Buffer}
*/
get localKey () {
return this.base?.local?.key || null
}
/**
* Get the bootstrap key (the "contract address").
* Share this with other admins to join the metadata Autobase.
* @returns {Buffer}
*/
get key () {
return this.base?.key || null
}
/**
* Get the discovery key for Hyperswarm.
* @returns {Buffer}
*/
get discoveryKey () {
return this.base?.discoveryKey || null
}
/**
* Whether this peer can write to the metadata Autobase.
* @returns {boolean}
*/
get writable () {
return this.base?.writable || false
}
// ── Device pairing ────────────────────────────────────────
/**
* Join another device's MetadataBase by key (like a room invite).
* Closes the current Autobase, writes the new key, and reopens.
* Hyperswarm + Protomux writer exchange handle the rest.
*
* @param {Buffer} key - Remote MetadataBase bootstrap key (32 bytes)
*/
async joinKey (key) {
if (!key || key.length !== 32) throw new Error('Invalid key: must be 32-byte Buffer')
const currentKey = this.key
if (currentKey && currentKey.equals(key)) {
console.log('[metadata] joinKey: already on this key')
return // Already on this key
}
console.log('[metadata] joinKey: closing current, switching to', key.toString('hex').slice(0, 16))
// Close current instance
await this.close()
// Write new key and reopen
this._bootstrapKey = key
this._saveKey(key)
await this.ready()
console.log('[metadata] joinKey: reopened, key=%s writable=%s writers=%d',
this.base.key?.toString('hex')?.slice(0, 12),
this.base.writable,
this.base.activeWriters?.length || 0
)
}
// ── Lifecycle ──────────────────────────────────────────────
async close () {
if (this._closing) return
this._closing = true
if (this.swarm) {
await this.swarm.destroy()
this.swarm = null
}
if (this.base) {
await this.base.close()
this.base = null
}
if (this.store) {
await this.store.close()
this.store = null
}
this._ready = false
this._closing = false
}
// ── Key persistence ────────────────────────────────────────
_loadKey () {
try {
const hex = fs.readFileSync(this._keyPath, 'utf8').trim()
if (hex.length === 64) return Buffer.from(hex, 'hex')
} catch {
// No key file — will create new Autobase
}
return null
}
_saveKey (key) {
if (!key) return
try {
const dir = path.dirname(this._keyPath)
fs.mkdirSync(dir, { recursive: true })
fs.writeFileSync(this._keyPath, key.toString('hex'))
} catch (err) {
console.error('[metadata-base] Failed to save key:', err.message)
}
}
}
// ── Helpers ──────────────────────────────────────────────────
/**
* Field-level Last-Write-Wins merge.
* Each field carries its own timestamp so concurrent updates
* to different fields don't overwrite each other.
*
* @param {object|null} existing - Current value from view
* @param {object} incoming - New fields
* @param {number} ts - Timestamp of the incoming operation
* @returns {object} Merged result
*/
function mergeFields (existing, incoming, ts) {
if (!existing) {
const result = { ...incoming }
result._fieldTs = {}
for (const k of Object.keys(incoming)) {
if (!k.startsWith('_')) result._fieldTs[k] = ts
}
return result
}
const result = { ...existing }
const fieldTs = { ...(existing._fieldTs || {}) }
for (const [k, v] of Object.entries(incoming)) {
if (k.startsWith('_')) continue // Skip internal fields
const prevTs = fieldTs[k] || 0
if (ts >= prevTs) {
// Null/undefined = delete the field
if (v === null || v === undefined) {
delete result[k]
delete fieldTs[k]
} else {
result[k] = v
fieldTs[k] = ts
}
}
}
result._fieldTs = fieldTs
return result
}
export { mergeFields }