/**
* MultiwriterManager — orchestrates SharedRegistry, SharedNotebook,
* and InstanceLinker into a single entry point for multi-writer.
*
* Mirror mode: when linked, automatically syncs notebook drives and
* user identities between instances via the shared Autobase registry.
*/
import { SharedRegistry } from './shared-registry.js'
import { SharedNotebook } from './shared-notebook.js'
import { InstanceLinker } from './instance-linker.js'
import mutexify from 'mutexify/promise.js'
import crypto from 'hypercore-crypto'
const CALLBACK_TOKEN_TTL = 30 * 60 * 1000 // 30 minutes
export class MultiwriterManager {
constructor (corestore, opts = {}) {
this._store = corestore
this._registryBootstrapKey = opts.registryBootstrapKey || null
this._swarm = opts.swarm || null
this._instanceKey = opts.instanceKey || null
this._instanceUrl = opts.instanceUrl || null
this._registry = null
this._linker = null
this._sharedNotebooks = new Map() // slug -> SharedNotebook
this._ready = false
this._syncInterval = null
this._initialSyncTimeout = null
this._driveManager = null
this._replicationManager = null
this._identityRegistry = null
this._pendingCallbackTokens = new Map() // token -> { createdAt }
this._lock = mutexify() // serialize state transitions
this._identityLock = mutexify() // serialize identity sync operations
this._addedNbWriters = new Set()
}
get registry () {
return this._registry
}
/**
* Wire up external managers for mirror sync.
* Call after construction, before ready().
*/
setManagers ({ driveManager, replicationManager, identityRegistry }) {
this._driveManager = driveManager
this._replicationManager = replicationManager
this._identityRegistry = identityRegistry
}
async ready () {
if (this._ready) return
// Stable per-instance writer keypair (deterministic from the corestore
// primary key, bootstrap-independent) so our writer key is known BEFORE
// pairing and admission can ride the authenticated handshake.
this._writerKeyPair = await this._store.createKeyPair('multiwriter-writer')
// Initialize shared registry
const regOpts = { keyPair: this._writerKeyPair }
if (this._registryBootstrapKey) {
regOpts.bootstrapKey = this._registryBootstrapKey
}
this._registry = new SharedRegistry(this._store, regOpts)
await this._registry.ready()
// Initialize instance linker
this._linker = new InstanceLinker(this._registry, {
swarm: this._swarm,
instanceKey: this._instanceKey,
instanceUrl: this._instanceUrl
})
// Restore shared notebooks from registry
await this._restoreSharedNotebooks()
this._ready = true
}
async _restoreSharedNotebooks () {
const notebooks = await this._registry.listNotebooks()
for (const nb of notebooks) {
if (nb.shared && nb.autobaseKey) {
const slug = nb.slug
if (!this._sharedNotebooks.has(slug)) {
const snb = new SharedNotebook(this._store, slug, {
bootstrapKey: Buffer.from(nb.autobaseKey, 'hex')
})
await snb.ready()
this._sharedNotebooks.set(slug, snb)
}
}
}
}
async getStatus () {
const sharedSlugs = Array.from(this._sharedNotebooks.keys())
let linkedInstances = []
let registeredDrives = []
let sharedUsers = []
if (this._registry) {
try { linkedInstances = await this._registry.listLinkedInstances() } catch {}
try { registeredDrives = await this._registry.listRegisteredDrives() } catch {}
try { sharedUsers = await this._registry.listSharedUsers() } catch {}
}
const localWriterKey = this._registry ? this._registry.localWriterKey : null
return {
linked: this._registryBootstrapKey !== null || linkedInstances.length > 0,
registryKey: this._registry ? this._registry.bootstrapKey : null,
localWriterKey: localWriterKey ? (Buffer.isBuffer(localWriterKey) ? localWriterKey.toString('hex') : localWriterKey) : null,
sharedNotebooks: this._sharedNotebooks.size,
sharedSlugs,
linkedPeers: linkedInstances.length,
linkedInstances,
mirroredDrives: registeredDrives.length,
registeredDrives,
sharedUsers: sharedUsers.length
}
}
// ── Linking ──
async createLinkInvite () {
// Generate a one-time callback token for the add-writer HTTP callback
const callbackToken = crypto.randomBytes(32).toString('hex')
this._pendingCallbackTokens.set(callbackToken, { createdAt: Date.now() })
// Ensure linker has current instance info + callback token
this._linker._instanceKey = this._instanceKey
this._linker._instanceUrl = this._instanceUrl
this._linker._callbackToken = callbackToken
return this._linker.createInvite()
}
async joinWithInvite (inviteHex) {
const release = await this._lock()
try {
return await this._joinWithInviteUnsafe(inviteHex)
} finally {
release()
}
}
async _joinWithInviteUnsafe (inviteHex) {
const result = await this._linker.joinWithInvite(inviteHex)
const { bootstrapKey, writerKey, instanceUrl, callbackToken } = result
if (!bootstrapKey) throw new Error('No bootstrapKey received from source')
// 1. Close old linker
const oldRegistry = this._registry
if (this._linker) await this._linker.destroy()
// 2. Create NEW SharedRegistry with source's bootstrapKey
const bootstrapBuf = Buffer.isBuffer(bootstrapKey)
? bootstrapKey : Buffer.from(bootstrapKey, 'hex')
this._registry = new SharedRegistry(this._store, { bootstrapKey: bootstrapBuf, keyPair: this._writerKeyPair })
await this._registry.ready()
// 3. Get our new localWriterKey from the SHARED registry
const localWriterKey = this._registry.localWriterKey.toString('hex')
console.log('[Multiwriter] Joined shared registry')
// 4. Re-wire linker with new registry
this._linker = new InstanceLinker(this._registry, {
swarm: this._swarm,
instanceKey: this._instanceKey,
instanceUrl: this._instanceUrl
})
// 5. Close old registry
if (oldRegistry && oldRegistry !== this._registry) {
try { await oldRegistry.close() } catch {}
}
// 6. Persist bootstrapKey
this._registryBootstrapKey = bootstrapBuf
// 7. Add SOURCE's writerKey to the SHARED registry and link its instance
if (writerKey) {
try {
await this._registry.addWriter(writerKey)
if (result.instanceKey) {
await this._registry.linkInstance(result.instanceKey)
}
console.log('[Multiwriter] Added source writer')
} catch (err) {
console.error('[Multiwriter] Failed to add source writer:', err.message)
}
}
// 8. Admission is handled during pairing: our writer key is stable
// (bootstrap-independent) and was sent as signed userData in the handshake,
// so the source already admitted us over the authenticated channel. The old
// unauthenticated HTTP callback (bearer token → admit ANY key) is removed.
void localWriterKey
void callbackToken
// 9. Announce shared registry on swarm for discovery
if (this._swarm && this._registry.discoveryKey) {
this._swarm.join(this._registry.discoveryKey, { server: true, client: true })
console.log('[Multiwriter] Announced shared registry on swarm')
}
result.localWriterKey = localWriterKey
result.linked = true
return result
}
/**
* Add a remote writer to the shared registry.
* When called via HTTP callback, callbackToken must match a pending token.
* When called locally (e.g., during joinWithInvite), no token needed.
*/
async addWriter (writerKeyHex, opts = {}) {
if (!this._registry) throw new Error('Registry not initialized')
// Verify callback token if provided (HTTP callback path)
if (opts.callbackToken) {
const tokenEntry = this._pendingCallbackTokens.get(opts.callbackToken)
if (!tokenEntry || (Date.now() - tokenEntry.createdAt > CALLBACK_TOKEN_TTL)) {
this._pendingCallbackTokens.delete(opts.callbackToken)
throw new Error('Invalid or expired callback token')
}
this._pendingCallbackTokens.delete(opts.callbackToken) // one-time use
}
await this._registry.addWriter(writerKeyHex)
if (opts.instanceKey) {
await this._registry.linkInstance(opts.instanceKey)
} else {
await this._registry.linkInstance(writerKeyHex) // fallback: use writerKey as instance id
}
console.log(`[Multiwriter] Added writer: ${writerKeyHex.substring(0, 16)}...`)
}
// ── Mirror sync ──
/**
* Sync drives and identities between linked instances.
*
* Source side: registers local drives + user identities in shared registry
* Mirror side: discovers remote drives, adds them, joins replication
*/
async syncMirror () {
if (!this._registry || !this._driveManager) return
const release = await this._lock()
try {
await this._syncMirrorUnsafe()
} finally {
release()
}
}
async _syncMirrorUnsafe () {
// Clean up expired callback tokens
const now = Date.now()
for (const [token, entry] of this._pendingCallbackTokens) {
if (now - entry.createdAt > CALLBACK_TOKEN_TTL) {
this._pendingCallbackTokens.delete(token)
}
}
// 0. Update Autobase to get latest from peers
try {
await this._registry.base.update()
} catch (err) {
console.error('[Mirror] Registry update failed:', err.message)
}
// Only log registry state on first sync
if (!this._syncCount) this._syncCount = 0
if (this._syncCount++ === 0) {
console.log('[Mirror] Registry writable:', this._registry.base.writable, 'writers:', this._registry.base.activeWriters?.length || '?')
}
// 1. Register all local drives in the shared registry
try {
await this._registerLocalDrives()
} catch (err) {
console.error('[Mirror] _registerLocalDrives failed:', err.message)
}
// 2. Migrate local notebooks into SharedNotebooks
try {
await this._migrateNotebooks()
} catch (err) {
console.error('[Mirror] _migrateNotebooks failed:', err.message)
}
// 3. Discover remote SharedNotebooks from registry
try {
await this._discoverRemoteSharedNotebooks()
} catch (err) {
console.error('[Mirror] _discoverRemoteSharedNotebooks failed:', err.message)
}
// 4. Propagate linked instance writer keys to SharedNotebook Autobases
// Only the bootstrapper (or existing writer) can add new writers to an Autobase.
// When a new instance links, its writer key is in the registry but not in per-notebook
// Autobases. This step ensures all linked instance writer keys are added.
try {
await this._propagateWritersToNotebooks()
} catch (err) {
console.error('[Mirror] _propagateWritersToNotebooks failed:', err.message)
}
// 5. Discover and add remote drives from the shared registry
try {
await this._discoverRemoteDrives()
} catch (err) {
console.error('[Mirror] _discoverRemoteDrives failed:', err.message)
}
// 6. Sync user identities to shared registry
try {
await this._syncIdentities()
} catch (err) {
console.error('[Mirror] _syncIdentities failed:', err.message)
}
}
async _registerLocalDrives () {
const drives = await this._driveManager.listDrives()
for (const drive of drives) {
if (!drive.isLocal) continue
// Check if already registered in shared registry
const existing = await this._registry.getRegisteredDrive(drive.slug)
if (existing && existing.driveKey === drive.driveKey) continue
await this._registry.registerDrive(drive.slug, drive.driveKey, drive.discoveryKey, {
title: drive.slug
})
console.log(`[Mirror] Registered local drive: ${drive.slug}`)
}
}
async _discoverRemoteDrives () {
const registeredDrives = await this._registry.listRegisteredDrives()
for (const regDrive of registeredDrives) {
// Skip drives we already have locally
const localInfo = await this._driveManager.getDriveInfo(regDrive.slug)
if (localInfo && localInfo.isLocal) continue
if (localInfo && localInfo.driveKey === regDrive.driveKey) continue
// Add as remote drive
try {
await this._driveManager.addRemoteDrive(regDrive.driveKey, regDrive.slug)
console.log(`[Mirror] Added remote drive: ${regDrive.slug} (${regDrive.driveKey.substring(0, 16)}...)`)
// Join replication
if (this._replicationManager) {
await this._replicationManager.joinRemote(regDrive.slug)
console.log(`[Mirror] Joined replication for: ${regDrive.slug}`)
}
} catch (err) {
console.error(`[Mirror] Failed to add remote drive ${regDrive.slug}:`, err.message)
}
}
}
async _migrateNotebooks () {
if (!this._driveManager) return
const drives = await this._driveManager.listDrives()
for (const driveEntry of drives) {
if (!driveEntry.isLocal) continue
const slug = driveEntry.slug
let snb = this._sharedNotebooks.get(slug)
if (!snb) {
snb = new SharedNotebook(this._store, slug)
await snb.ready()
this._sharedNotebooks.set(slug, snb)
const keyHex = snb.bootstrapKey.toString('hex')
const nb = await this._registry.getNotebookBySlug(slug)
if (nb) {
await this._registry.shareNotebook(nb.notebookId, keyHex)
} else {
const result = await this._registry.createNotebook({ slug, title: slug })
await this._registry.shareNotebook(result.notebookId, keyHex)
}
console.log('[Mirror] Created SharedNotebook for: ' + slug)
}
// Register our writer key once (skip if already done this session)
const writerKey = snb.localWriterKey.toString('hex')
const regKey = `nb-writer:${slug}:${writerKey}`
if (!this._addedNbWriters.has(regKey)) {
try {
const instKey = this._instanceKey || this._registry.localWriterKey.toString('hex')
await this._registry.registerNotebookWriter(slug, instKey, writerKey)
this._addedNbWriters.add(regKey)
console.log(`[Mirror] Registered notebook writer for ${slug}: ${writerKey.substring(0, 16)}...`)
} catch (err) {
this._addedNbWriters.add(regKey) // Don't retry on error
console.error(`[Mirror] Failed to register notebook writer for ${slug}:`, err.message)
}
}
// Import notes from Hyperdrive into SharedNotebook (once per session)
if (!this._migratedSlugs) this._migratedSlugs = new Set()
if (this._migratedSlugs.has(slug)) continue
this._migratedSlugs.add(slug)
const drive = await this._driveManager.getDrive(slug)
if (!drive) continue
let noteFiles
try {
const files = []
for await (const f of drive.readdir('/notes')) {
if (f.endsWith('.json')) files.push(f)
}
noteFiles = files
} catch {
continue
}
for (const file of noteFiles) {
try {
const buf = await drive.get('/notes/' + file)
if (!buf) continue
const note = JSON.parse(buf.toString())
await snb.importNote({
noteId: note.id,
title: note.title,
content: note.content,
tags: note.tags,
author: 'migration',
createdAt: note.createdAt,
updatedAt: note.updatedAt
})
} catch (err) {
console.error('[Mirror] Failed to import note ' + file + ' from ' + slug + ':', err.message)
}
}
}
}
async _discoverRemoteSharedNotebooks () {
const registeredNotebooks = await this._registry.listNotebooks()
for (const nb of registeredNotebooks) {
if (!nb.shared || !nb.autobaseKey) continue
const slug = nb.slug
if (this._sharedNotebooks.has(slug)) continue
try {
const bootstrapKey = Buffer.from(nb.autobaseKey, 'hex')
const snb = new SharedNotebook(this._store, slug, { bootstrapKey })
await snb.ready()
this._sharedNotebooks.set(slug, snb)
console.log('[Mirror] Discovered remote SharedNotebook: ' + slug)
// Register our per-notebook writer key once
const localWriterKey = snb.localWriterKey.toString('hex')
const discRegKey = `nb-writer:${slug}:${localWriterKey}`
if (!this._addedNbWriters.has(discRegKey)) {
try {
const instKey = this._instanceKey || this._registry.localWriterKey.toString('hex')
await this._registry.registerNotebookWriter(slug, instKey, localWriterKey)
this._addedNbWriters.add(discRegKey)
console.log(`[Mirror] Registered notebook writer for ${slug}: ${localWriterKey.substring(0, 16)}...`)
} catch (err) {
this._addedNbWriters.add(discRegKey)
console.error(`[Mirror] Failed to register notebook writer for ${slug}:`, err.message)
}
}
} catch (err) {
console.error('[Mirror] Failed to open remote SharedNotebook ' + slug + ':', err.message)
}
}
}
async _propagateWritersToNotebooks () {
if (!this._registeredNbWriters) this._registeredNbWriters = new Set()
if (!this._addedNbWriters) this._addedNbWriters = new Set()
const instKey = this._instanceKey || this._registry.localWriterKey.toString('hex')
for (const [slug, snb] of this._sharedNotebooks) {
// Register our own writer key once
const regKey = `${slug}:${instKey}`
if (!this._registeredNbWriters.has(regKey)) {
try {
const writerKey = snb.localWriterKey.toString('hex')
await this._registry.registerNotebookWriter(slug, instKey, writerKey)
this._registeredNbWriters.add(regKey)
console.log(`[Mirror] Registered notebook writer for ${slug}: ${writerKey.substring(0, 16)}...`)
} catch (err) {
this._registeredNbWriters.add(regKey) // Don't retry on error
}
}
// Only the bootstrapper/existing writer can add new writers
if (!snb._base.writable) continue
// Get per-notebook writer keys registered by other instances
const nbWriters = await this._registry.listNotebookWriters(slug)
if (nbWriters.length === 0) continue
await snb._base.update()
const localKey = snb.localWriterKey.toString('hex')
for (const nbw of nbWriters) {
if (nbw.writerKey === localKey) continue
const addKey = `${slug}:${nbw.writerKey}`
if (this._addedNbWriters.has(addKey)) continue
try {
await snb.addWriter(nbw.writerKey)
this._addedNbWriters.add(addKey)
console.log(`[Mirror] Added writer ${nbw.writerKey.substring(0, 16)}... to SharedNotebook: ${slug}`)
} catch (err) {
// Mark as done even on error to prevent retry spam
this._addedNbWriters.add(addKey)
}
}
}
}
async _syncIdentities () {
const release = await this._identityLock()
try {
await this._syncIdentitiesUnsafe()
} finally {
release()
}
}
async _syncIdentitiesUnsafe () {
if (!this._identityRegistry) {
console.log('[Mirror] _syncIdentities: no identity registry, skipping')
return
}
try {
// Get all local users and sync them to shared registry
const users = await this._identityRegistry.listAllUsers()
for (let i = 0; i < users.length; i++) {
const user = users[i]
if (!user || !user.userId) {
console.log(`[Mirror] Skipping user[${i}]: no userId`)
continue
}
// Get user's identity/public key info
let publicKey = null
let identityMethod = 'web-identity'
if (user.identities && user.identities.length > 0) {
try {
const identity = await this._identityRegistry.getIdentity(user.identities[0])
if (identity && identity.credentials) {
publicKey = identity.credentials.publicKey || null
identityMethod = identity.method || 'web-identity'
}
} catch {}
}
// Skip unchanged users to avoid unnecessary writes
try {
const existing = await this._registry.getSharedUser(user.userId)
if (existing && existing.publicKey === publicKey && existing.username === user.username) continue
} catch {}
try {
await this._registry.syncIdentity(user.userId, {
username: user.username || null,
name: user.name || '',
globalPermissions: user.globalPermissions || ['read'],
notebooks: user.notebooks || [],
publicKey,
identityMethod
})
} catch (err) {
console.error('[Mirror] Failed to sync user:', err.message)
}
}
} catch (err) {
console.error('[Mirror] Failed to push identities:', err.message, err.stack?.split('\n')[1])
}
// Pull: import shared users from registry into local identity registry
try {
const sharedUsers = await this._registry.listSharedUsers()
let imported = 0
for (const su of sharedUsers) {
if (!su || !su.userId) continue
// Check if user already exists locally by userId (including merged users)
const localUser = await this._identityRegistry.getUser(su.userId)
if (localUser) continue
// Check if this userId was previously merged into another user — don't re-create
const mergedCheck = await this._identityRegistry.db.get(`user:${su.userId}`)
if (mergedCheck && mergedCheck.value && mergedCheck.value.mergedInto) continue
// Check if publicKey already belongs to a local user — don't overwrite
if (su.publicKey) {
const existingRef = await this._identityRegistry.db.get(`pubkey:${su.publicKey}`)
if (existingRef && existingRef.value) {
console.log('[Mirror] Skipping import: publicKey already belongs to local user')
continue
}
}
// Create user locally with same userId
const userObj = {
userId: su.userId,
username: null, // Don't import username — may conflict with local
name: su.name || 'Anonymous',
email: null,
profilePicture: null,
globalPermissions: su.globalPermissions || ['read'],
notebooks: su.notebooks || [{ name: '*', permissions: ['read'] }],
identities: [],
createdAt: Date.now(),
updatedAt: Date.now()
}
await this._identityRegistry.db.put(`user:${su.userId}`, userObj)
// Create identity + reverse lookup if publicKey available
if (su.publicKey) {
const identity = {
identityId: su.publicKey,
method: su.identityMethod || 'web-identity',
userId: su.userId,
credentials: { publicKey: su.publicKey },
createdAt: Date.now(),
updatedAt: Date.now()
}
await this._identityRegistry.db.put(`identity:${su.publicKey}`, identity)
await this._identityRegistry.db.put(`pubkey:${su.publicKey}`, { identityId: su.publicKey })
// Link identity to user
userObj.identities.push(su.publicKey)
await this._identityRegistry.db.put(`user:${su.userId}`, userObj)
}
if (su.username) {
// Only set username if not taken locally
const existing = await this._identityRegistry.db.get(`username:${su.username}`)
if (!existing) {
await this._identityRegistry.db.put(`username:${su.username}`, { userId: su.userId })
}
}
imported++
}
if (imported > 0) {
console.log(`[Mirror] Imported ${imported} users from shared registry`)
}
} catch (err) {
console.error('[Mirror] Failed to pull identities:', err.message)
}
}
/**
* Start event-driven mirror sync (pear-git pattern).
*
* Instead of polling on a timer, sync triggers on peer connection events.
* The swarm already does store.replicate(conn) in ReplicationManager —
* we just hook the connection event to exchange writers via Protomux
* and run a one-time sync on first peer connect.
*/
startMirrorSync () {
if (this._mirrorStarted) return
this._mirrorStarted = true
this._hasSynced = false
if (!this._swarm) {
console.log('[Mirror] No swarm available, skipping mirror sync')
return
}
// On each new peer connection, exchange Autobase writer keys
this._swarm.on('connection', (conn) => {
this._onPeerConnection(conn)
})
console.log('[Mirror] Event-driven sync active (syncs on peer connect)')
}
/**
* Handle a new peer connection — exchange writers and sync once.
*/
_onPeerConnection (conn) {
if (!this._registry) return
// One-time sync on first peer connection (deferred, non-blocking)
if (!this._hasSynced) {
this._hasSynced = true
// Small delay to let replication streams establish
setTimeout(() => {
this.syncMirror().catch(err => {
console.error('[Mirror] Initial sync failed:', err.message)
})
}, 3000)
}
}
stopMirrorSync () {
this._mirrorStarted = false
}
// ── Notebook sharing ──
async shareNotebook (slug) {
// Ensure notebook exists in registry; find by slug
let nb = await this._registry.getNotebookBySlug(slug)
if (!nb) {
// Create the notebook entry in the registry
const result = await this._registry.createNotebook({ slug, title: slug })
nb = await this._registry.getNotebook(result.notebookId)
}
// Create a SharedNotebook if not already shared
if (this._sharedNotebooks.has(slug)) {
return this._sharedNotebooks.get(slug)
}
const snb = new SharedNotebook(this._store, slug)
await snb.ready()
// Store the bootstrap key in the registry
const keyHex = snb.bootstrapKey.toString('hex')
await this._registry.shareNotebook(nb.notebookId, keyHex)
this._sharedNotebooks.set(slug, snb)
return snb
}
async unshareNotebook (slug) {
const snb = this._sharedNotebooks.get(slug)
if (!snb) return
// Find the notebook in registry and unshare
const nb = await this._registry.getNotebookBySlug(slug)
if (nb) {
await this._registry.unshareNotebook(nb.notebookId)
}
await snb.close()
this._sharedNotebooks.delete(slug)
}
getSharedNotebook (slug) {
return this._sharedNotebooks.get(slug) || null
}
isShared (slug) {
return this._sharedNotebooks.has(slug)
}
async close () {
this.stopMirrorSync()
// Close all shared notebooks
for (const snb of this._sharedNotebooks.values()) {
await snb.close()
}
this._sharedNotebooks.clear()
// Destroy linker
if (this._linker) {
await this._linker.destroy()
this._linker = null
}
// Close registry
if (this._registry) {
await this._registry.close()
this._registry = null
}
this._ready = false
}
}