/**
* state-base.js — Per-room game-state Autobase.
*
* Separate from MetadataBase (which is per-identity and stores
* cross-room registry data). StateBase is per-room, scoped by the
* keet roomKey, and only stores game-state writes for that room.
*
* Bootstrap-key distribution
* --------------------------
* Autobase needs a real keypair as its genesis writer. We can't
* derive that deterministically from a shared roomKey. Instead we
* use the same pattern keet itself uses for cross-instance control
* signaling: a dedicated Hyperswarm topic derived from roomKey,
* with a Protomux channel for the actual payload exchange. See
* keet-core/index.js:1238 (protomuxWakeup.addStream) for the
* canonical idiom.
*
* announceTopic = hash("hyper-tps/state-announce/v1" || roomKey)
*
* Every keet-room member's daemon joins that topic. On connect we
* open a Protomux channel `hyper-tps/state-announce` and broadcast
* whatever stateBaseKey we know (founder broadcasts theirs; joiners
* with no key send nothing, wait to receive).
*
* Founder election
* ----------------
* First peer to call set() with no existing key becomes founder
* (creates the state-base locally, becomes genesis indexer). If two
* peers race, both create independent state-bases; on first connect
* the lex-smaller key wins and the loser switches over. Loser's
* pre-merge writes are lost — acceptable since this is a startup
* race only, not a steady-state concern.
*
* Once a peer has the bootstrap key, the state-base's own swarm
* (on base.discoveryKey, separate from the announce topic) handles
* replication, and an `_exchangeWriters` Protomux channel mirrors
* the metadata-base.js writer-exchange logic to add joiners to the
* writer set.
*/
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hypercore from 'hypercore'
import Hyperbee from 'hyperbee'
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import c from 'compact-encoding'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import path from 'bare-path'
const ANNOUNCE_NAMESPACE = Buffer.from('hyper-tps/state-announce/v1', 'utf-8')
// Deterministic gamestate-base identity, derived from the keet ROOM's shared
// secret (encryptionKey/seed from getRoomKeys). Every room member derives the
// SAME signing keypair and SAME base.key — keet's own pattern (room key =
// getWriterKey of a seed-derived signing keypair, invitations/index.js). This
// removes key conveyance (no announce gossip) and lets existing rooms converge
// by all computing the same key. Bump STATE_VERSION when the derivation or the
// view schema changes — the migration is gated on it.
const STATE_VERSION = 2
const GS_NAMESPACE = Buffer.from('hyper-tps/state/v' + STATE_VERSION, 'utf-8')
// Deterministic ed25519 keypair from the room seed. Only the elected creator
// ever opens the base WITH this keypair (to write genesis); others open by
// base.key with their own corestore-derived local writer.
function deriveGsKeyPair (roomSeed) {
return crypto.keyPair(crypto.hash([GS_NAMESPACE, toBuf(roomSeed)]))
}
// base.key whose genesis writer is the gs keypair (autobase Pattern A precompute).
function deriveBaseKey (gsPublicKey) {
return Hypercore.key({ version: 1, signers: [{ publicKey: gsPublicKey }] })
}
// Keet roles: admin=1, mod=2, peer=3, read-only=4. Admins/mods are eligible
// to be the genesis creator and become autobase indexers (room-state spec B1/B3).
function normKeyHex (k) {
if (!k) return null
if (Buffer.isBuffer(k)) return k.toString('hex')
const s = String(k)
return /^[0-9a-fA-F]{64}$/.test(s) ? s.toLowerCase() : s
}
function toBuf (k) {
if (Buffer.isBuffer(k)) return k
const s = String(k)
return /^[0-9a-fA-F]{64}$/.test(s) ? Buffer.from(s, 'hex') : Buffer.from(s)
}
export class StateBase {
constructor (storagePath, roomKey, opts = {}) {
this._roomKey = Buffer.isBuffer(roomKey)
? roomKey
: Buffer.from(String(roomKey), 'hex')
if (this._roomKey.length !== 32) {
throw new Error('roomKey must be 32 bytes (or 64-char hex)')
}
this._roomKeyHex = this._roomKey.toString('hex')
this._storagePath = path.join(storagePath, 'state-bases', this._roomKeyHex)
this._keyPath = path.join(this._storagePath, 'bootstrap-key')
this._network = opts.network !== false
this._swarmOpts = opts.swarmOpts || null // for tests: pass { bootstrap } to override public DHT
this._bootstrapKey = this._loadKey()
// Deterministic-creator bootstrap (room-state spec B1). Only the elected
// creator mints genesis; everyone else adopts the conveyed base.key. No
// random per-peer founding, no lossy lex-election → no split-brain.
// selfKey = this daemon's keet identity/device pubkey (hex|Buffer)
// getMembers = async () => [{ key|devicePubkey hex, role 1|2|3|4 }]
// When both are absent (lone daemon / unit tests with no membership
// context) we fall back to self-creation so a single writer still works.
// Signing identity for membership-gated writer admission (spec B2). The
// daemon signs its own writer-key announce with selfKeyPair; receiving
// writers verify it against keet membership before addWriter. selfKey
// (pubkey, used for creator election B1) is taken from selfKeyPair when
// provided, else from opts.selfKey.
this._selfKeyPair = opts.selfKeyPair || null
this._selfKey = this._selfKeyPair
? toBuf(this._selfKeyPair.publicKey)
: (opts.selfKey ? toBuf(opts.selfKey) : null)
// Normalized string identity for membership comparison. Computed from the
// ORIGINAL input so it handles both 32-byte hex pubkeys (tests) and z32
// memberId strings (real keet getMembers) consistently.
const selfKeyInput = this._selfKeyPair ? this._selfKeyPair.publicKey : opts.selfKey
this._selfKeyNorm = selfKeyInput ? normKeyHex(selfKeyInput) : null
this._getMembers = typeof opts.getMembers === 'function' ? opts.getMembers : null
// Injectable crypto for membership-gated admission (P3). verifyAttest lets
// the daemon verify a joiner's keet attestation (IdentityKey.verify, pure);
// attest lets it sign its own announce with its keet device credentials.
// When absent (tests), _admitWriters falls back to the raw-ed25519 path.
this._attest = typeof opts.attest === 'function' ? opts.attest : null
this._verifyAttest = typeof opts.verifyAttest === 'function' ? opts.verifyAttest : null
// Deterministic base identity (room-state v2). When the keet room's shared
// secret (roomSeed = encryptionKey from getRoomKeys) is wired, base.key is
// DERIVED (no random founder, no announce conveyance) — every member
// computes the same key. Without it we fall back to the v1 election +
// announce path (lone daemon / unit tests / no-seed contexts).
this._roomSeed = opts.roomSeed ? toBuf(opts.roomSeed) : null
this._gsKeyPair = this._roomSeed ? deriveGsKeyPair(this._roomSeed) : null
this._detBaseKey = this._gsKeyPair ? deriveBaseKey(this._gsKeyPair.publicKey) : null
// Members-only room encryptionKey → encrypt the base at rest, so even a
// non-member who derived/found base.key cannot READ it (keet's posture:
// it opens the room base `encrypted:true` with the room encryptionKey).
this._encryptionKey = opts.encryptionKey ? toBuf(opts.encryptionKey) : null
this.store = null
this.base = null
this.swarm = null // state-base own replication swarm
this.announceSwarm = null // bootstrap-key announce swarm
this._announceTopic = null
this._announceChannels = new Map() // socket → channel msg handle
this._ready = false
this._closing = false
this._watchers = new Set()
}
static announceTopic (roomKey) {
const rk = Buffer.isBuffer(roomKey) ? roomKey : Buffer.from(String(roomKey), 'hex')
return crypto.hash(Buffer.concat([ANNOUNCE_NAMESPACE, rk]))
}
// ── Lifecycle ──────────────────────────────────────────────
async ready () {
if (this._ready) return
if (this._closing) throw new Error('StateBase is closing')
fs.mkdirSync(this._storagePath, { recursive: true })
this.store = new Corestore(this._storagePath)
await this.store.ready()
try {
await this._open()
} catch (err) {
// Corruption recovery: a room's StateBase corestore can become
// unrecoverable (e.g. a damaged autobase system core →
// `DECODING_ERROR: Unknown wire type` thrown from SystemView._open during
// base.ready(); the #82-class optimistic-append oplog damage). When that
// happens the base can NEVER open and every state.* fails forever. We
// can't repair an undecodable oplog, so RESTART the room's StateBase:
// quarantine the corrupt corestore (renamed aside, NEVER deleted —
// never-destroy-state) and reopen a fresh one. A member then re-replicates
// the room's state from peers; a fresh creator starts clean. Guarded to a
// single attempt so a genuinely broken fresh open can't loop.
if (!this._isCorruptionError(err)) throw err
console.error('[state-base] room=%s CORRUPT base (%s) — quarantining + restarting StateBase',
this._roomKeyHex.slice(0, 12), err.message)
await this._quarantineAndReset()
}
this._ready = true
}
// The open sequence, factored out so ready() can wrap it in corruption
// recovery and _quarantineAndReset() can replay it on a fresh store.
async _open () {
if (this._detBaseKey) {
// Deterministic v2 path: every member computes the same base.key, so no
// announce/key-conveyance is needed (rendezvous is the derived
// discoveryKey). Migrate any pre-existing v1 base first (never lossy).
await this._migrateIfNeeded()
await this._openDeterministic()
} else {
// v1 path (no room seed wired — lone daemon / unit tests): announce-based
// key conveyance + creator election.
if (this._network) this._startAnnounceSwarm()
if (this._bootstrapKey) {
await this._openBase(this._bootstrapKey)
} else if (this._getMembers && this._selfKey) {
await this._maybeBootstrapAsCreator()
}
// Without membership context we do NOT mint at ready — preserve the lazy
// "first writer founds on set()" behaviour so two peers don't both mint.
}
}
// Storage-corruption signals that mean "this corestore can't be opened/read"
// — as opposed to transient/network errors, which must NOT trigger a reset.
_isCorruptionError (err) {
const m = (err && err.message ? err.message : String(err)).toUpperCase()
return m.includes('DECODING_ERROR') ||
m.includes('UNKNOWN WIRE TYPE') ||
m.includes('CORRUPT') ||
m.includes('INVALID DEVICE FILE') ||
m.includes('CHECKSUM') ||
m.includes('INVALID OPLOG') ||
m.includes('INVALID HEADER')
}
// Quarantine the corrupt corestore (NEVER delete) and reopen a fresh one.
// Also the explicit-restart path behind the state.reset RPC.
async _quarantineAndReset () {
// Release everything holding the corrupt corestore (its fd-lock) first.
if (this.swarm) { try { await this.swarm.destroy() } catch {} ; this.swarm = null }
if (this.announceSwarm) { try { await this.announceSwarm.destroy() } catch {} ; this.announceSwarm = 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 }
// Move the corrupt dir aside — never destroy persistent state. The operator
// can inspect/recover it offline; on disk it's <path>.corrupt-<ts>.
const quarantine = this._storagePath + '.corrupt-' + Date.now()
try {
fs.renameSync(this._storagePath, quarantine)
console.error('[state-base] room=%s quarantined corrupt store -> %s',
this._roomKeyHex.slice(0, 12), quarantine)
} catch (e) {
console.error('[state-base] room=%s quarantine rename failed: %s', this._roomKeyHex.slice(0, 12), e.message)
throw e
}
// Forget the stale on-disk key + any pending migration — open fresh as the
// deterministic v2 base (or v1-elected), with the migration marked done so
// we don't re-attempt the (now-quarantined) v1 snapshot.
this._bootstrapKey = null
this._pendingMigration = null
fs.mkdirSync(this._storagePath, { recursive: true })
this.store = new Corestore(this._storagePath)
await this.store.ready()
try { fs.writeFileSync(path.join(this._storagePath, 'migrated-v' + STATE_VERSION), String(Date.now())) } catch {}
// Reopen — must succeed on a fresh store; if not, surface the error (no loop).
await this._open()
console.log('[state-base] room=%s StateBase restarted clean (writable=%s)',
this._roomKeyHex.slice(0, 12), this.writable)
}
/**
* Open the deterministic base (room-state v2). The elected creator opens WITH
* the gs keypair → it is the genesis writer of _detBaseKey; everyone else
* opens by base.key with their own corestore-derived local writer and
* replicates the creator's genesis. With no membership context (lone daemon)
* we act as creator so a single writer still works.
*/
async _openDeterministic () {
const amCreator = (!this._getMembers || !this._selfKey) ? true : await this._amCreator()
if (amCreator) {
console.log('[state-base] room=%s deterministic creator — genesis writer (base=%s)',
this._roomKeyHex.slice(0, 12), this._detBaseKey.toString('hex').slice(0, 12))
await this._openBase(this._detBaseKey, this._gsKeyPair)
} else {
console.log('[state-base] room=%s deterministic member — open derived base %s',
this._roomKeyHex.slice(0, 12), this._detBaseKey.toString('hex').slice(0, 12))
await this._openBase(this._detBaseKey)
}
if (this._pendingMigration) this._scheduleMigrationReplay()
}
// ── Version-gated migration: v1 (random-founder) → v2 (deterministic) ──
//
// Modelled on keet's migrations/3 (one-time, gated, transform-and-preserve;
// never lossy). On boot, if this storage holds a v1 base whose key ≠ the
// derived v2 key and no migration marker exists, snapshot the v1 view (our
// own writes — owner-partitioned). After the v2 base opens and becomes
// writable, replay each key onto v2 (LWW by ts in _apply) and stamp the
// marker. The v1 corestore data is NEVER deleted.
async _migrateIfNeeded () {
const markerPath = path.join(this._storagePath, 'migrated-v' + STATE_VERSION)
try { if (fs.existsSync(markerPath)) return } catch {}
const v1Key = this._bootstrapKey
if (!v1Key || (this._detBaseKey && v1Key.equals(this._detBaseKey))) return
console.log('[state-base] room=%s migration v1(%s)→v%d: snapshotting old base',
this._roomKeyHex.slice(0, 12), v1Key.toString('hex').slice(0, 12), STATE_VERSION)
let v1 = null
let snapStore = null
try {
// Open the v1 base on a corestore SESSION so closing it after the
// snapshot does NOT close the shared store the v2 base will use.
snapStore = this.store.session()
v1 = new Autobase(snapStore, v1Key, {
open: (store) => new Hyperbee(store.get('view'), { keyEncoding: 'utf-8', valueEncoding: 'json' }),
apply: this._apply.bind(this),
close: async (view) => { if (view) await view.close() },
valueEncoding: 'json', optimistic: true, ackInterval: 1000, ackThreshold: 4
})
await v1.ready()
await v1.update()
const snap = {}
const stream = v1.view.createReadStream()
for await (const entry of stream) snap[entry.key] = entry.value // { value, ts, _by }
this._pendingMigration = snap
console.log('[state-base] room=%s migration snapshot: %d keys', this._roomKeyHex.slice(0, 12), Object.keys(snap).length)
} catch (err) {
console.error('[state-base] migration snapshot failed:', err.message)
this._pendingMigration = null
} finally {
if (v1) { try { await v1.close() } catch {} }
if (snapStore) { try { await snapStore.close() } catch {} }
}
}
_scheduleMigrationReplay () {
const markerPath = path.join(this._storagePath, 'migrated-v' + STATE_VERSION)
const replay = async () => {
if (!this.base || !this.base.writable || !this._pendingMigration) return false
const snap = this._pendingMigration
this._pendingMigration = null
let n = 0
for (const key of Object.keys(snap)) {
const w = snap[key]
if (!w || typeof w !== 'object') continue
// Preserve the original value + ts so LWW in _apply keeps the newest
// across both histories — never clobbers a newer v2 write.
try { await this.base.append({ op: 'set', key, value: w.value, ts: w.ts || 0 }); n++ } catch {}
}
await this.base.update().catch(() => {})
try { fs.writeFileSync(markerPath, String(Date.now())) } catch {}
console.log('[state-base] room=%s migration replayed %d keys onto v%d',
this._roomKeyHex.slice(0, 12), n, STATE_VERSION)
return true
}
if (this.base && this.base.writable) {
replay().catch((e) => console.error('[state-base] migration replay threw:', e.message))
} else if (this.base) {
const onWritable = () => {
replay().then((done) => { if (done && this.base) { try { this.base.off('writable', onWritable) } catch {} } })
.catch((e) => console.error('[state-base] migration replay threw:', e.message))
}
this.base.on('writable', onWritable)
}
}
// ── Deterministic creator election + single bootstrap (spec B1) ──
/**
* Elect the genesis creator from keet membership and mint ONLY if this
* daemon is that creator. Idempotent; safe to call repeatedly (e.g. when
* membership becomes available later). Never mints a second base.
*/
async _maybeBootstrapAsCreator () {
if (this.base || this._bootstrapKey) return
const amCreator = await this._amCreator()
if (amCreator) {
console.log('[state-base] room=%s elected creator — minting genesis',
this._roomKeyHex.slice(0, 12))
await this._openBase(null) // null bootstrap → autobase mints a fresh key
this._broadcastAnnounce()
} else {
console.log('[state-base] room=%s not creator — awaiting announced base.key',
this._roomKeyHex.slice(0, 12))
}
}
/**
* Deterministic across all peers: the creator is the admin/mod (role 1|2)
* with the lex-smallest identity key. Every daemon computes the same answer
* from the same membership, so exactly one peer ever mints genesis.
* Falls back to true (self-create) when no membership context is wired —
* preserves lone-daemon / single-writer behaviour.
*/
async _amCreator () {
if (!this._getMembers || !this._selfKey) return false // no deterministic election context
let members
try { members = await this._getMembers() } catch { return false }
if (!Array.isArray(members) || members.length === 0) return false
const eligible = members
.map(m => ({ key: normKeyHex(m.key || m.devicePubkey || m.identity || m.memberId), role: Number(m.role) }))
.filter(m => m.key && (m.role === 1 || m.role === 2))
.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
if (eligible.length === 0) return false
return eligible[0].key === this._selfKey.toString('hex')
}
async _openBase (bootstrapKey, keyPair = null) {
const baseOpts = {
open: (store) => new Hyperbee(store.get('view'), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
}),
apply: this._apply.bind(this),
close: async (view) => { if (view) await view.close() },
valueEncoding: 'json',
optimistic: true,
ackInterval: 1000,
ackThreshold: 4
}
// Deterministic creator pins the gs keypair so its local writer IS the
// genesis writer of base.key (autobase Pattern A). Never share this keypair
// across peers — only the single elected creator opens with it.
if (keyPair) baseOpts.keyPair = keyPair
// Encrypt the base at rest with the room's members-only encryptionKey.
if (this._encryptionKey) {
baseOpts.encryptionKey = this._encryptionKey
baseOpts.encrypted = true
}
this.base = new Autobase(this.store, bootstrapKey, baseOpts)
await this.base.ready()
this._saveKey(this.base.key)
this._bootstrapKey = this.base.key
if (this._network) {
if (!this.swarm) {
this.swarm = new Hyperswarm(this._swarmOpts || undefined)
this.swarm.on('connection', (socket) => {
try {
this.store.replicate(socket)
this._admitWriters(socket)
} catch (err) {
console.error('[state-base] swarm connection threw:', err.message)
}
})
}
this.swarm.join(this.base.discoveryKey)
}
console.log('[state-base] room=%s base.key=%s writable=%s',
this._roomKeyHex.slice(0, 12),
this.base.key.toString('hex').slice(0, 12),
this.base.writable
)
}
// ── Announce swarm (bootstrap-key gossip) ─────────────────
_startAnnounceSwarm () {
this._announceTopic = StateBase.announceTopic(this._roomKey)
this.announceSwarm = new Hyperswarm(this._swarmOpts || undefined)
this.announceSwarm.on('connection', (socket) => {
this._handleAnnounceConnection(socket)
})
this.announceSwarm.join(this._announceTopic)
console.log('[state-base] room=%s announce-topic=%s',
this._roomKeyHex.slice(0, 12),
this._announceTopic.toString('hex').slice(0, 16)
)
}
_handleAnnounceConnection (socket) {
try {
const mux = Protomux.from(socket)
const channel = mux.createChannel({
protocol: 'hyper-tps/state-announce',
id: this._roomKey,
onclose: () => { this._announceChannels.delete(socket) }
})
if (!channel) return // channel collision, already open
const msg = channel.addMessage({
encoding: c.buffer,
onmessage: (remoteKey) => {
// Async work scheduled via .catch so a throw can't escape as
// an unhandled rejection (Bare crashes the process on those).
if (!remoteKey || remoteKey.length !== 32) return
this._onAnnounceReceived(remoteKey).catch((err) => {
console.error('[state-base] _onAnnounceReceived threw:', err.message)
})
}
})
channel.open()
this._announceChannels.set(socket, msg)
if (this._bootstrapKey) {
try { msg.send(this._bootstrapKey) } catch (err) {
console.error('[state-base] announce send threw:', err.message)
}
}
} catch (err) {
console.error('[state-base] _handleAnnounceConnection threw:', err.message)
}
}
async _onAnnounceReceived (remoteKey) {
// Deterministic-creator design (spec B1): there is exactly ONE base.key per
// room — the elected creator's. If we already have a base (minted or
// previously adopted), ignore every other announce. The lossy lex-smaller
// _switchBootstrap election is GONE — it discarded the loser's writes and
// was the source of the bd2b0d/4d8550 split-brain.
if (this._bootstrapKey) return
console.log('[state-base] room=%s adopting creator base.key %s',
this._roomKeyHex.slice(0, 12),
remoteKey.toString('hex').slice(0, 12)
)
await this._openBase(remoteKey)
}
_broadcastAnnounce () {
if (!this._bootstrapKey) return
for (const msg of this._announceChannels.values()) {
try { msg.send(this._bootstrapKey) } catch {}
}
}
/**
* Become founder: create a state-base locally if none exists yet.
* Idempotent — no-op if already bootstrapped.
*/
async ensureFounder () {
// Backwards-compatible shim. With keet membership it is creator-gated
// (mints only if this daemon is the elected creator); without membership
// it lazily self-founds (lone-daemon / legacy behaviour).
if (this.base) return this.base.key
if (this._getMembers && this._selfKey) {
await this._maybeBootstrapAsCreator()
} else {
await this._openBase(null)
this._broadcastAnnounce()
}
return this.base ? this.base.key : null
}
// ── Apply contract ────────────────────────────────────────
async _apply (nodes, view, base) {
const batch = view.batch({ update: false })
for (const node of nodes) {
const op = node.value
// Autobase can surface nodes with no user value — system/membership
// checkpoints (from addWriter) and flush nodes that appear under
// concurrent appends. Reading op.op on those throws TypeError, and
// since keet-daemon 0.526 removed the uncaughtException net the throw
// escapes autobase._drain → SIGABRT (crashed the dedicated server +
// graphical client; also the #30 "Autobase is closing"). Skip them.
if (op == null || typeof op !== 'object') continue
const writer = node.from?.key?.toString('hex')?.slice(0, 16) || 'local'
// D1b: optimistic admission (liveness fast-path). A not-yet-admitted member
// writes optimistically with an `_admit` proof; verify it DETERMINISTICALLY
// here (pure crypto + the projected `__members__` local state — NO external
// lookup, Rule #3) and ack the writer so the block commits. An unverifiable
// or non-member block is NOT acked → autobase does not apply it (README
// optimistic contract), so we skip it. `_admitWriters` stays the role-aware
// authority (peer→writer vs admin/mod→indexer); this only makes writes land.
if (op._admit && node.from?.key) {
const ok = await this._verifyAdmissionInApply(node.from.key, op._admit, batch)
if (!ok) continue
try { await base.ackWriter(node.from.key) } catch {}
}
switch (op.op) {
case 'set': {
if (!op.key) break
// Last-write-wins by ts: an older write (notably a migration replay
// of v1 data) must not clobber a newer value already in the view.
const existing = await batch.get(op.key)
if (existing && existing.value && typeof existing.value.ts === 'number' &&
existing.value.ts > (typeof op.ts === 'number' ? op.ts : 0)) break
await batch.put(op.key, { value: op.value, ts: op.ts, _by: writer })
break
}
case 'del': {
if (!op.key) break
await batch.del(op.key)
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 '_members': {
// Projected keet-room membership (cross-base sync, spec §3.0). The keet
// room base is a closed contract and lives in a SEPARATE autobase, so
// membership cannot be read in apply (external = non-deterministic,
// Rule #3). Instead the daemon's out-of-apply watcher copies the
// membership snapshot in via this op; we store the whole map under one
// key so apply (admission gating, D1b) can read it as deterministic
// LOCAL state. Last-writer-wins on the snapshot key.
if (op.members && typeof op.members === 'object') {
await batch.put('__members__', op.members)
}
break
}
}
}
await batch.flush()
for (const watcher of this._watchers) {
try { watcher() } catch {}
}
}
// ── Writer exchange on state-base swarm ───────────────────
// ── Membership-gated writer admission (spec B2) ──
//
// Replaces the old unauthenticated _exchangeWriters (which added ANY peer on
// the topic as an indexer). A joiner announces its autobase local writer key
// together with its keet identity and a signature over (localKey ‖ base.key).
// The receiving writer/indexer verifies the signature, confirms the identity
// is an active keet room member, and only then appends an _addWriter op —
// isIndexer set from the keet role (admin/mod → indexer, peer → optimistic
// writer). Mirrors keet's addDevice: verify attestation, THEN addWriter
// (keet-cli/.../contract/members/index.js). The membership check is a side
// effect at the admit point (NOT in apply — autobase Rule #3).
_admitWriters (socket) {
try {
const mux = Protomux.from(socket)
const channel = mux.createChannel({
protocol: 'hyper-tps/state-writer-exchange/v2',
id: this.base.key
})
if (!channel) return
const msg = channel.addMessage({
encoding: c.buffer,
onmessage: (buf) => {
this._onAdmitAnnounce(buf).catch((err) => {
console.error('[state-base] _onAdmitAnnounce threw:', err.message)
})
}
})
channel.open()
this._buildAdmitAnnounce().then((mine) => {
if (mine) {
try { msg.send(mine) } catch (err) {
console.error('[state-base] admit announce send threw:', err.message)
}
}
}).catch((err) => console.error('[state-base] _buildAdmitAnnounce threw:', err.message))
} catch (err) {
console.error('[state-base] _admitWriters threw:', err.message)
}
}
// Announce = localKey(32) ‖ identity(32) ‖ proof(rest). The proof attests
// (over localKey‖base.key, so it can't be replayed onto another base) that
// `identity` vouches for localKey. Two modes, by injected crypto:
// • keet attestation (P3): proof = attest(localKey‖base.key) — variable
// length keet-identity-key ProofEncoding (deviceProof chain identity→device).
// • raw ed25519 (tests/fallback): proof = sign(localKey‖base.key) — 64 bytes.
async _buildAdmitAnnounce () {
if (!this.base || !this.base.local || !this.base.local.key) return null
const localKey = this.base.local.key
const signData = Buffer.concat([localKey, this.base.key])
if (this._attest && this._selfKey) {
const proof = await this._attest(signData)
if (!proof) return null
return Buffer.concat([localKey, toBuf(this._selfKey), Buffer.isBuffer(proof) ? proof : Buffer.from(proof)])
}
if (this._selfKeyPair && this._selfKeyPair.secretKey) {
const sig = crypto.sign(signData, toBuf(this._selfKeyPair.secretKey))
return Buffer.concat([localKey, toBuf(this._selfKeyPair.publicKey), sig])
}
return null
}
async _onAdmitAnnounce (buf) {
if (!Buffer.isBuffer(buf) || buf.length < 96) return // 32+32+(>=32 proof/64 sig)
if (!this.base || !this.base.writable) return // only writers can admit
const localKey = buf.subarray(0, 32)
const identity = buf.subarray(32, 64)
const proof = buf.subarray(64)
const signData = Buffer.concat([localKey, this.base.key])
// 1. Proof — attestation (P3) or raw sig (fallback) over localKey‖base.key.
let ok = false
if (this._verifyAttest) {
ok = await this._verifyAttest(proof, signData, identity)
} else {
if (proof.length !== 64) return
ok = crypto.verify(signData, proof, identity)
}
if (!ok) {
console.log('[state-base] admit: bad proof — refusing writer')
return
}
// 2. Membership — identity must be an active keet room member.
const role = await this._memberRole(identity)
if (role == null) {
console.log('[state-base] admit: %s not a room member — refusing writer',
identity.toString('hex').slice(0, 12))
return
}
// 3. Admit. admin(1)/mod(2) → indexer; peer(3) → optimistic writer.
const isIndexer = role === 1 || role === 2
console.log('[state-base] admit: %s role=%d → addWriter(indexer=%s)',
identity.toString('hex').slice(0, 12), role, isIndexer)
try {
await this.base.append({
op: '_addWriter',
key: localKey.toString('hex'),
indexer: isIndexer,
ts: Date.now()
})
} catch {
// Lost writability mid-flight — the next admit round retries.
}
}
async _memberRole (identityKey) {
if (!this._getMembers) return null
let members
try { members = await this._getMembers() } catch { return null }
if (!Array.isArray(members)) return null
const hex = normKeyHex(identityKey)
for (const m of members) {
if (normKeyHex(m.key || m.devicePubkey || m.identity || m.memberId) === hex) {
return Number(m.role) || 3
}
}
return null
}
// ── D1b: optimistic admission proof (op-embedded) ──────────
// Op-embedded form of _buildAdmitAnnounce. Returns { identity, proof } hex
// parts to carry in an optimistic op; proof attests (over localKey‖base.key —
// so it can't be replayed onto another base) that `identity` vouches for this
// writer's local key. attestation (keet) or raw ed25519 (fallback).
async _buildAdmissionProof () {
if (!this.base || !this.base.local || !this.base.local.key || !this.base.key) return null
const localKey = this.base.local.key
const signData = Buffer.concat([localKey, this.base.key])
if (this._attest && this._selfKey) {
const proof = await this._attest(signData)
if (!proof) return null
const pbuf = Buffer.isBuffer(proof) ? proof : Buffer.from(proof)
return { identity: toBuf(this._selfKey).toString('hex'), proof: pbuf.toString('hex') }
}
if (this._selfKeyPair && this._selfKeyPair.secretKey) {
const sig = crypto.sign(signData, toBuf(this._selfKeyPair.secretKey))
return { identity: toBuf(this._selfKeyPair.publicKey).toString('hex'), proof: sig.toString('hex') }
}
return null
}
// DETERMINISTIC in-apply verification for an optimistic `_admit` op (D1b):
// pure crypto verify of the proof over localKey‖base.key + membership against
// the projected `__members__` LOCAL state (no external lookup — Rule #3).
// Returns whether to ack the writer. Mirrors _onAdmitAnnounce but in-apply and
// gated on projected membership rather than a live getMembers() call.
async _verifyAdmissionInApply (localKey, admit, batch) {
try {
if (!admit || !admit.identity || !admit.proof || !this.base || !this.base.key) return false
// Decode variable-length hex explicitly — toBuf() only hex-decodes 32-byte
// (64-char) strings, but proof is a 64-byte sig (128 hex) or a longer keet
// attestation, so toBuf would mis-read it as UTF-8.
const identity = Buffer.from(admit.identity, 'hex')
const proof = Buffer.from(admit.proof, 'hex')
const signData = Buffer.concat([localKey, this.base.key])
let ok = false
if (this._verifyAttest) ok = await this._verifyAttest(proof, signData, identity)
else { if (proof.length !== 64) return false; ok = crypto.verify(signData, proof, identity) }
if (!ok) return false
const entry = await batch.get('__members__')
if (!entry || !entry.value) return false
return entry.value[identity.toString('hex')] != null || entry.value[normKeyHex(admit.identity)] != null
} catch { return false }
}
// ── Public API ────────────────────────────────────────────
async set (key, value) {
if (!this._ready) throw new Error('StateBase not ready')
if (!this.base) {
// Creator-gated (with membership) or lazy self-found (without) — see
// ensureFounder. A non-creator that hasn't adopted the creator's key yet
// cannot write; surfaced as an error the client handles (write_rejected),
// never a silent random found.
await this.ensureFounder()
if (!this.base) throw new Error('StateBase not bootstrapped yet (no creator base.key available)')
}
if (this.base.writable) {
// Admitted writer: PLAIN append (the proven metadata-base.js pattern).
// Unconditional `optimistic:true` corrupted the oplog under concurrent
// writes (#80 SIGABRT), so optimistic is used ONLY for the not-yet-admitted
// case below — a single op until acked, never for steady-state writes.
await this.base.append({ op: 'set', key, value, ts: Date.now() })
} else {
// D1b liveness: not yet admitted (e.g. the creator was offline when we
// joined). Append OPTIMISTICALLY carrying our admission proof so the write
// LANDS instead of throwing "Not writable"; apply verifies the proof +
// projected membership and acks us, committing the block. `_admitWriters`
// stays the role-aware authority that promotes us to a real writer.
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('StateBase: not writable and no identity to build an admission proof')
await this.base.append({ op: 'set', key, value, ts: Date.now(), _admit: admit }, { optimistic: true })
}
await this.base.update()
}
async get (key) {
if (!this._ready) throw new Error('StateBase not ready')
if (!this.base) return null
await this.base.update()
const entry = await this.base.view.get(key)
return entry ? entry.value : null
}
async del (key) {
if (!this._ready) throw new Error('StateBase not ready')
if (!this.base) return
await this.base.append({ op: 'del', key, ts: Date.now() })
await this.base.update()
}
async list (prefix = '') {
if (!this._ready) throw new Error('StateBase not ready')
if (!this.base) return {}
await this.base.update()
const out = {}
const stream = this.base.view.createReadStream({
gte: prefix,
lt: prefix + '~'
})
for await (const entry of stream) {
out[entry.key] = entry.value
}
return out
}
/**
* Project keet-room membership into this StateBase (cross-base sync, spec §3.0).
* Called by the daemon's out-of-apply watcher on the keet room base's 'update'
* event with the current members, so membership becomes deterministic LOCAL
* state that apply can gate admission on (D1b). No-op if this daemon isn't a
* writer yet (another writer projects; eventual). `members` accepts a
* [{ key|memberId|id, role }] array or a { id: role } map. Returns whether it
* appended. Verified: test/probe-state-membership-projection.js.
*/
async projectMembers (members) {
if (!this._ready || !this.base || !this.base.writable) return false
const map = {}
if (Array.isArray(members)) {
for (const m of members) {
const id = normKeyHex(m.key || m.devicePubkey || m.identity || m.memberId || m.id)
if (id) map[id] = Number(m.role) || 3
}
} else if (members && typeof members === 'object') {
for (const id of Object.keys(members)) map[normKeyHex(id) || id] = Number(members[id]) || 3
}
await this.base.append({ op: '_members', members: map })
await this.base.update()
return true
}
/** Read a projected member's role (admin=1/mod=2/peer=3) from LOCAL state, or null. */
async getProjectedRole (id) {
if (!this._ready || !this.base) return null
await this.base.update()
const entry = await this.base.view.get('__members__')
if (!entry || !entry.value) return null
const hex = normKeyHex(id)
const role = entry.value[hex] != null ? entry.value[hex] : entry.value[id]
return role != null ? Number(role) : null
}
watch (fn) {
this._watchers.add(fn)
return () => this._watchers.delete(fn)
}
get stateBaseKey () { return this.base?.key || null }
get writable () { return this.base?.writable || false }
get hasBootstrap () { return !!this._bootstrapKey }
// ── Close ─────────────────────────────────────────────────
async close () {
if (this._closing) return
this._closing = true
if (this.announceSwarm) { await this.announceSwarm.destroy(); this.announceSwarm = null }
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._announceChannels.clear()
this._watchers.clear()
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 {}
return null
}
_saveKey (key) {
if (!key) return
try {
fs.mkdirSync(path.dirname(this._keyPath), { recursive: true })
fs.writeFileSync(this._keyPath, key.toString('hex'))
} catch (err) {
console.error('[state-base] save key failed:', err.message)
}
}
}