/**
* metadata-base.js — Unified Autobase for all pear-git shared metadata
*
* Owns its own Corestore and Hyperswarm (keet-daemon pattern).
* joinKey() does a full close/reopen cycle — destroys swarm, store,
* recreates fresh with the new key, automatic replication.
*
* Namespaces:
* identity → user:{id}, identity:{id}, pubkey:{pk}, sshkey:{fp}, noisekey:{pk}, username:{name}, revoked:{id}
* repo → repo:{name} (catalog)
* ref → ref:{repo}:{path} (git refs)
* group → group:{name}, peer:{peerId}, grant:{id}, invite:{id}, request:{id}, *-grants, *-requests, group-meta:{key}
* deploy → server:{key}, deploy:{repo}, deploy-log:{repo}:{ts}
* issue → issue:{repo}:{id}, comment:{repo}:{issueId}:{commentId}, label:{repo}:{name},
* repo-writer:{repo}:{userId}, repo-ref:{repo}:{ref}, repo-meta:{repo}:{key}, issue-counter:{repo}
*/
import path from 'bare-path'
import fs from 'bare-fs'
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 {
applyIdentityOp,
applyGroupOp,
applyDeployOp,
applyRepoOp
} from './apply/index.js'
export class MetadataBase {
/**
* @param {object} opts
* @param {string} opts.storagePath - Base storage directory (MetadataBase creates a subdirectory)
* @param {Buffer|string|null} opts.bootstrapKey - Existing Autobase key to join (null = create new)
* @param {object} opts.swarmOpts - Hyperswarm options (e.g. { bootstrap } for testing)
*/
constructor (opts = {}) {
this._storagePath = path.join(opts.storagePath, 'metadata')
this._keyPath = path.join(opts.storagePath, 'metadata-key')
this._bootstrapKey = opts.bootstrapKey
? (typeof opts.bootstrapKey === 'string' ? Buffer.from(opts.bootstrapKey, 'hex') : opts.bootstrapKey)
: null
this._swarmOpts = opts.swarmOpts || {}
this._listeners = new Map()
// Created in ready()
this.store = null
this._base = null
this.swarm = null
this._ready = false
}
async ready () {
if (this._ready) return
// Own Corestore — separate from server's store
this.store = new Corestore(this._storagePath)
await this.store.ready()
// Load persisted key or use provided one
const savedKey = this._loadKey()
const bootstrapKey = this._bootstrapKey || savedKey
// Create Autobase
this._base = new Autobase(this.store, bootstrapKey, {
open: (store) => new Hyperbee(store.get('view'), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
}),
apply: this._apply.bind(this),
valueEncoding: 'json',
ackInterval: 5000
})
await this._base.ready()
// Persist key
this._saveKey(this._base.key)
// Own Hyperswarm — joins discovery key, handles connections
this.swarm = new Hyperswarm(this._swarmOpts)
this.swarm.on('connection', (socket) => {
const peerKey = socket.remotePublicKey?.toString('hex')?.slice(0, 8) || 'unknown'
console.log(`[Metadata] Swarm connection from ${peerKey}`)
this.store.replicate(socket)
this._exchangeWriters(socket)
})
this.swarm.join(this._base.discoveryKey)
console.log(`[Metadata] Swarm joined discovery key: ${this._base.discoveryKey.toString('hex').slice(0, 16)}`)
this._ready = true
}
// ── Key persistence ─────────────────────────────────────────
_loadKey () {
try {
const hex = fs.readFileSync(this._keyPath, 'utf-8').trim()
if (hex.length === 64) return Buffer.from(hex, 'hex')
} catch {}
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] Failed to save key:', err.message)
}
}
// ── Properties ──────────────────────────────────────────────
get key () { return this._base?.key || null }
get bootstrapKeyHex () { return this._base?.key ? this._base.key.toString('hex') : null }
get discoveryKey () { return this._base?.discoveryKey || null }
get localKey () { return this._base?.local ? this._base.local.key : null }
get writable () { return this._base?.writable || false }
get view () { return this._base?.view || null }
// ── Unified apply function (the smart contract) ─────────────
async _apply (nodes, view, base) {
const batch = view.batch({ update: false })
for (const node of nodes) {
const op = node.value
if (!op || !op.ns) continue
const writer = node.from?.key?.toString('hex')?.slice(0, 16) || 'local'
switch (op.ns) {
case 'identity':
await applyIdentityOp(op, batch, base)
break
case 'group':
await applyGroupOp(op, batch, base)
break
case 'deploy':
await applyDeployOp(op, batch, base)
break
case 'issue':
await applyRepoOp(op, batch, base)
break
case 'repo': {
const key = `repo:${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 })
break
}
case 'ref': {
const key = `ref:${op.repoName}:${op.ref}`
await batch.put(key, {
hash: op.hash,
_by: writer,
_ts: op.ts
})
break
}
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
}
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.ts)
await batch.put(key, { ...merged, _by: writer, _ts: op.ts, _ns: op.ns })
}
break
}
}
}
await batch.flush()
// Notify listeners
for (const node of nodes) {
const op = node.value
if (!op || !op.ns) continue
const fns = this._listeners.get(op.ns) || []
for (const fn of fns) {
try { fn(op) } catch {}
}
}
}
// ── Public API: generic ─────────────────────────────────────
async append (op) {
if (!this._base) throw new Error('MetadataBase not ready')
await this._base.append(op)
}
async set (ns, id, fields) {
await this._base.append({ ns, id, fields, ts: Date.now() })
}
async get (ns, id) {
await this._base.update()
const entry = await this._base.view.get(`${ns}:${id}`)
return entry ? entry.value : null
}
async getByKey (key) {
if (!this._base) return null
await this._base.update()
const entry = await this._base.view.get(key)
return entry ? entry.value : null
}
async list (ns) {
if (!this._base) return []
await this._base.update()
const results = []
for await (const entry of this._base.view.createReadStream({
gt: `${ns}:`,
lt: `${ns}:\xff`
})) {
results.push({ id: entry.key.slice(ns.length + 1), ...entry.value })
}
return results
}
async listByRange (gt, lt) {
await this._base.update()
const results = []
for await (const entry of this._base.view.createReadStream({ gt, lt })) {
results.push({ key: entry.key, ...entry.value })
}
return results
}
async getByPrefix (prefix) {
await this._base.update()
const results = []
for await (const entry of this._base.view.createReadStream({
gte: prefix,
lt: prefix + '\xff'
})) {
results.push({ key: entry.key, value: entry.value })
}
return results
}
async del (key) {
await this._base.append({ ns: '_del', key, ts: Date.now() })
}
// ── Ref operations (convenience) ────────────────────────────
async updateRef (repoName, ref, hash) {
await this._base.append({ ns: 'ref', repoName, ref, hash, ts: Date.now() })
}
async listRefs (repoName) {
await this._base.update()
const refs = {}
for await (const entry of this._base.view.createReadStream({
gt: `ref:${repoName}:`,
lt: `ref:${repoName}:\xff`
})) {
const refPath = entry.key.slice(`ref:${repoName}:`.length)
refs[refPath] = entry.value.hash
}
return refs
}
// ── Writer management ───────────────────────────────────────
async addWriter (key) {
if (!this._base) return
const keyHex = Buffer.isBuffer(key) ? key.toString('hex') : key
await this._base.append({ ns: '_addWriter', key: keyHex, indexer: true, ts: Date.now() })
}
async removeWriter (key) {
const keyHex = Buffer.isBuffer(key) ? key.toString('hex') : key
await this._base.append({ ns: '_removeWriter', key: keyHex, ts: Date.now() })
}
// ── Event listeners ─────────────────────────────────────────
on (ns, fn) {
if (!this._listeners.has(ns)) this._listeners.set(ns, [])
this._listeners.get(ns).push(fn)
}
off (ns, fn) {
const fns = this._listeners.get(ns)
if (!fns) return
const idx = fns.indexOf(fn)
if (idx !== -1) fns.splice(idx, 1)
}
// ── Writer exchange (Protomux) ──────────────────────────────
_exchangeWriters (socket) {
if (!this._base) return
const mux = Protomux.from(socket)
if (!mux || !this._base.key) return
const channel = mux.createChannel({
protocol: 'pear-git/metadata-exchange',
id: this._base.key
})
if (!channel) return
const writerMsg = channel.addMessage({
encoding: c.buffer,
onmessage: async (remoteKey) => {
if (!remoteKey || remoteKey.length !== 32) return
try {
await this.addWriter(remoteKey)
await this._base.update()
console.log(`[Metadata] Added writer: ${remoteKey.toString('hex').slice(0, 12)}`)
} catch (err) {
if (!err.message.includes('Already')) {
console.log(`[Metadata] Writer exchange note: ${err.message}`)
}
}
}
})
channel.open()
if (this._base.local?.key) {
writerMsg.send(this._base.local.key)
}
}
// ── Join existing MetadataBase (federation) ──────────────────
/**
* Switch to a remote peer's MetadataBase.
* Full close/reopen cycle: destroys swarm + store, recreates with new key.
* New swarm joins new discovery key, peers connect, data replicates automatically.
*/
async joinKey (key) {
const keyBuf = typeof key === 'string' ? Buffer.from(key, 'hex') : key
if (!keyBuf || keyBuf.length !== 32) throw new Error('joinKey requires a 32-byte key')
if (this._base?.key && this._base.key.equals(keyBuf)) return
console.log(`[Metadata] joinKey: switching to ${keyBuf.toString('hex').slice(0, 16)}`)
// Full teardown
await this.close()
// Save new key and reopen
this._bootstrapKey = keyBuf
this._saveKey(keyBuf)
await this.ready()
console.log(`[Metadata] joinKey: now on ${this._base.key.toString('hex').slice(0, 16)}, writable=${this._base.writable}`)
}
// ── Lifecycle ───────────────────────────────────────────────
async update () {
if (!this._base) return
await this._base.update()
}
async close () {
this._ready = false
if (this.swarm) {
try { await this.swarm.destroy() } catch {}
this.swarm = null
}
if (this._base) {
try { await this._base.close() } catch {}
this._base = null
}
if (this.store) {
try { await this.store.close() } catch {}
this.store = null
}
}
}
// ── Field-level LWW merge ─────────────────────────────────────
function mergeFields (prev, fields, ts) {
if (!prev) return { ...fields }
const result = { ...prev }
for (const [key, val] of Object.entries(fields)) {
if (key.startsWith('_')) continue
result[key] = val
}
return result
}