// probe-cross-base-membership-sync.js
//
// Question: can we keep PART of one autobase (membership/roles in a keet-room-
// like SOURCE base) in sync into another autobase (a StateBase-like TARGET base)?
//
// Mechanism under test (the pgv3 pattern, grounded): an OUT-OF-APPLY watcher on
// SOURCE.on('update') reads source's membership slice and PROJECTS it into TARGET
// as ops. The projected state then lives in TARGET's own log, so TARGET's apply
// can read membership DETERMINISTICALLY (Rule #3-safe) — which is what D1 needs.
//
// This is application-level projection, NOT autobase-native sync. Full-snapshot
// projection (idempotent: each proj-sync replaces the member:* set) handles
// add / role-change / remove uniformly.
//
// Pure node (autobase 7.28). On-disk corestores so the result is real.
import os from 'os'
import path from 'path'
import fs from 'fs'
import crypto from 'hypercore-crypto'
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hyperbee from 'hyperbee'
const ROOT = fs.mkdtempSync(path.join(os.tmpdir(), 'xbase-sync-'))
let failures = 0
const check = (label, cond) => { console.log((cond ? 'PASS ' : 'FAIL ') + label); if (!cond) failures++ }
const sleep = (ms) => new Promise(r => setTimeout(r, ms))
function open (dir, keyPair, apply) {
const store = new Corestore(path.join(ROOT, dir))
const base = new Autobase(store, null, {
open: (s) => new Hyperbee(s.get('view'), { keyEncoding: 'utf-8', valueEncoding: 'json' }),
apply,
close: async (v) => { if (v) await v.close() },
valueEncoding: 'json', keyPair,
optimistic: true, ackInterval: 1000, ackThreshold: 4
})
return { store, base }
}
// SOURCE apply (keet-room-like): membership ops.
async function sourceApply (nodes, view) {
for (const node of nodes) {
const op = node.value
if (!op || typeof op !== 'object') continue
if (op.type === 'set-member') await view.put('member:' + op.id, { role: op.role })
else if (op.type === 'del-member') await view.del('member:' + op.id)
}
}
// TARGET apply (StateBase-like): a proj-sync op REPLACES the member:* set from a
// snapshot — deterministic (operates only on view + op data; no external lookup).
async function targetApply (nodes, view) {
for (const node of nodes) {
const op = node.value
if (!op || typeof op !== 'object') continue
if (op.type === 'proj-sync') {
for await (const e of view.createReadStream({ gte: 'member:', lt: 'member:~' })) await view.del(e.key)
for (const id of Object.keys(op.members)) await view.put('member:' + id, op.members[id])
}
}
}
async function readMembers (base) {
await base.update()
const out = {}
for await (const e of base.view.createReadStream({ gte: 'member:', lt: 'member:~' })) out[e.key.slice('member:'.length)] = e.value
return out
}
async function main () {
const source = open('source', crypto.keyPair(), sourceApply)
const target = open('target', crypto.keyPair(), targetApply)
await source.base.ready()
await target.base.ready()
// The projection watcher: on every SOURCE update, snapshot membership and
// write one proj-sync op into TARGET. Event-driven — no timers. Single-flight
// WITH re-run so the latest source state is always eventually projected even
// if updates arrive while a projection is in flight.
let projecting = false
let again = false
async function project () {
if (projecting) { again = true; return }
projecting = true
try {
do {
again = false
const members = await readMembers(source.base)
await target.base.append({ type: 'proj-sync', members })
await target.base.update()
} while (again)
} finally { projecting = false }
}
source.base.on('update', () => { project().catch(e => console.error('project threw', e.message)) })
// Eventually-consistent assertions: poll target until it converges.
async function waitFor (pred, ms = 4000) {
for (let i = 0; i < ms / 50; i++) { if (await pred()) return true; await sleep(50) }
return false
}
// 1. add two members with roles
await source.base.append({ type: 'set-member', id: 'alice', role: 1 }) // admin
await source.base.append({ type: 'set-member', id: 'bob', role: 3 }) // peer
await source.base.update()
check('1. target reflects source membership (alice=admin, bob=peer)',
await waitFor(async () => { const m = await readMembers(target.base); return m.alice?.role === 1 && m.bob?.role === 3 }))
// 2. role change propagates
await source.base.append({ type: 'set-member', id: 'bob', role: 2 }) // promote to mod
await source.base.update()
check('2. role change propagates (bob=mod)',
await waitFor(async () => (await readMembers(target.base)).bob?.role === 2))
// 3. removal propagates (full-snapshot replace drops it)
await source.base.append({ type: 'del-member', id: 'bob' })
await source.base.update()
check('3. removal propagates (bob gone, alice stays)',
await waitFor(async () => { const m = await readMembers(target.base); return m.bob === undefined && m.alice?.role === 1 }))
// 4. the projected membership is now DETERMINISTIC LOCAL state in target —
// target.apply could gate admission on it with no external lookup.
const tm = await readMembers(target.base)
check('4. target holds membership as its own replicated state (deterministic apply input)',
Object.keys(tm).length === 1 && tm.alice?.role === 1)
// 5. the auto-watcher keeps syncing new members live
await source.base.append({ type: 'set-member', id: 'carol', role: 3 })
await source.base.update()
check('5. event-driven watcher (no timer) syncs new member (carol)',
await waitFor(async () => (await readMembers(target.base)).carol?.role === 3))
await source.base.close(); await source.store.close()
await target.base.close(); await target.store.close()
try { fs.rmSync(ROOT, { recursive: true, force: true }) } catch {}
console.log('\n' + (failures === 0 ? 'ALL PASS' : failures + ' FAILURE(S)'))
process.exit(failures === 0 ? 0 : 1)
}
main().catch((e) => { console.error('THREW:', e); process.exit(2) })