/**
* Peer Routes — Unified Server Peering
*
* Replaces the 5 independent linking mechanisms with a single
* unified peering system supporting two modes:
*
* mirror — Full replication (all repos, all Autobases shared)
* peer — Selective federation (browse remote, sync selected repos)
*
* Handles:
* - GET /peers Peers dashboard (HTML)
* - POST /api/peer/invite Create peering invite
* - POST /api/peer/join Join via invite hex
* - GET /api/peer/list List peered servers
* - GET /api/peer/status Detailed peering status
* - DELETE /api/peer/:serverKey Remove a peer
* - POST /api/peer/repair Detect + fix holes in partially merged state
* - POST /api/peer/:serverKey/grant Grant repo/group access (peer mode)
* - POST /api/peer/:serverKey/upgrade Upgrade peer → mirror
* - POST /api/peer/sync Force immediate sync
*/
import { json, html, badRequest, notFound, readJson, escapeHtml } from '../response.js'
import { requireAdmin } from '../middleware/admin.js'
import { RepoLinker } from '../repo-link/repo-linker.js'
import { renderPeerDashboard } from '../peer-ui.js'
import os from 'bare-os'
import path from 'bare-path'
/**
* Build the best HTTP URL for this server (LAN-accessible).
*/
function buildHttpUrl (server) {
const { port, pathPrefix, bindAddresses } = server.config
const prefix = pathPrefix || ''
let host = 'localhost'
if (bindAddresses && bindAddresses.length > 0) {
const lanAddr = bindAddresses.find(a => a !== 'localhost' && a !== '127.0.0.1' && a !== '::1')
host = lanAddr || bindAddresses[0]
if (host === '0.0.0.0' || host === '::') {
try { host = os.hostname() } catch { host = 'localhost' }
}
}
return `http://${host}:${port || 5005}${prefix}`
}
/**
* Register peer routes
*/
export function registerPeerRoutes (router) {
// Dashboard (HTML)
router.get('/peers', handlePeerDashboard) // Auth checked client-side (browser can't send custom headers on navigation)
// API endpoints
router.post('/api/peer/invite', requireAdmin, handlePeerInvite)
router.post('/api/peer/join', requireAdmin, handlePeerJoin)
router.get('/api/peer/list', handlePeerList)
router.get('/api/peer/status', handlePeerStatus)
router.post('/api/peer/repair', requireAdmin, handlePeerRepair)
router.post('/api/peer/sync', requireAdmin, handlePeerSync)
router.post('/api/peer/sync/:repoName', requireAdmin, handlePeerSyncRepo)
router.get('/api/peer/sync-log', handlePeerSyncLog)
router.delete('/api/peer/:serverKey', requireAdmin, handlePeerRemove)
router.post('/api/peer/:serverKey/grant', requireAdmin, handlePeerGrant)
router.post('/api/peer/:serverKey/upgrade', requireAdmin, handlePeerUpgrade)
}
/**
* GET /peers — Peer dashboard HTML page
*/
async function handlePeerDashboard (req, res, server) {
const { serverIdentity } = server.p2p
const { pathPrefix } = server.config
try {
const peers = await listPeers(server)
const selfKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
const hostname = (() => { try { return os.hostname() } catch { return 'unknown' } })()
html(res, renderPeerDashboard({ peers, selfKey, hostname }, pathPrefix))
} catch (err) {
console.error('[Peer] Dashboard error:', err)
html(res, '<h1>Error loading peer dashboard</h1><p>' + escapeHtml(err.message) + '</p>', 500)
}
}
/**
* POST /api/peer/invite
* Create a peering invite. Body: { mode: 'mirror' | 'peer' }
*/
async function handlePeerInvite (req, res, server) {
const { serverIdentity } = server.p2p
const { namespaceRegistry } = server.namespaceData
const { deployRegistry } = server.deployData
const { repos } = server.repoData
const { namespaceManager } = server.namespaceData
try {
const body = await readJson(req)
const mode = body.mode || 'peer'
if (mode !== 'mirror' && mode !== 'peer') {
return badRequest(res, 'mode must be "mirror" or "peer"')
}
const serverKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
if (!serverKey) {
return badRequest(res, 'Server identity not initialized — P2P not enabled')
}
const httpUrl = buildHttpUrl(server)
const hostname = (() => { try { return os.hostname() } catch { return 'unknown' } })()
// Build invite payload with all bootstrap keys
const payload = {
type: 'peer-server',
mode,
serverKey,
hostname,
httpUrl
}
// Include bootstrap keys for shared Autobases
if (server.identityRegistry) {
payload.identityRegistryKey = server.identityRegistry.bootstrapKeyHex || null
}
if (namespaceRegistry) {
payload.namespaceRegistryKey = namespaceRegistry.bootstrapKeyHex || null
}
if (deployRegistry) {
payload.deployRegistryKey = deployRegistry.bootstrapKeyHex || null
}
// For peer mode, include available repos/groups
if (mode === 'peer') {
payload.repoList = (repos || []).map(r => r.name)
if (namespaceManager) {
try {
const groups = await namespaceManager.listGroups()
payload.groupList = groups.map(g => g.name)
} catch { payload.groupList = [] }
}
}
const linker = new RepoLinker({ bootstrap: server._bootstrap })
const result = await linker.createInvite(payload)
// Handle peer join in background
result.waitForPeer().then(async (joinerInfo) => {
if (joinerInfo && joinerInfo.serverKey) {
await completePeerJoin(server, joinerInfo, mode)
}
}).catch((err) => {
console.error('[Peer] Invite wait error:', err.message)
}).finally(() => {
linker.close().catch(() => {})
})
json(res, {
invite: result.inviteHex,
mode,
serverKey
})
} catch (err) {
console.error('[Peer] Invite error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* POST /api/peer/join
* Join a peer via invite hex. Body: { invite: '<hex>' }
*/
async function handlePeerJoin (req, res, server) {
const { serverIdentity } = server.p2p
const { namespaceRegistry } = server.namespaceData
const { deployRegistry } = server.deployData
let body
try {
body = await readJson(req)
} catch (err) {
return badRequest(res, 'Invalid JSON: ' + err.message)
}
const { invite } = body
if (!invite) return badRequest(res, 'invite is required')
const serverKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
if (!serverKey) {
return badRequest(res, 'Server identity not initialized — P2P not enabled')
}
const linker = new RepoLinker({ bootstrap: server._bootstrap })
try {
const httpUrl = buildHttpUrl(server)
const hostname = (() => { try { return os.hostname() } catch { return 'unknown' } })()
// Build our response payload (sent back to inviter)
const ourPayload = {
type: 'peer-server',
serverKey,
hostname,
httpUrl
}
if (server.identityRegistry) {
ourPayload.identityRegistryKey = server.identityRegistry.bootstrapKeyHex || null
}
if (namespaceRegistry) {
ourPayload.namespaceRegistryKey = namespaceRegistry.bootstrapKeyHex || null
}
if (deployRegistry) {
ourPayload.deployRegistryKey = deployRegistry.bootstrapKeyHex || null
}
const payload = await linker.joinWithInvite(invite, ourPayload)
if (!payload.serverKey) {
return badRequest(res, 'Invalid invite — no server key received')
}
if (payload.type !== 'peer-server') {
return badRequest(res, 'Invalid invite — not a server peering invite (type: ' + payload.type + ')')
}
const mode = payload.mode || 'peer'
// Complete the join on our side
await completePeerJoin(server, payload, mode)
json(res, {
success: true,
mode,
serverKey: payload.serverKey,
hostname: payload.hostname,
identityRegistryKey: payload.identityRegistryKey || null,
namespaceRegistryKey: payload.namespaceRegistryKey || null,
deployRegistryKey: payload.deployRegistryKey || null
})
} catch (err) {
console.error('[Peer] Join error:', err)
json(res, { error: err.message }, 500)
} finally {
await linker.close()
}
}
/**
* Complete a peer join — store federation entry, join Autobases based on mode.
* Used by both invite-creator (after waitForPeer) and joiner.
*/
async function completePeerJoin (server, remoteInfo, mode) {
const { serverIdentity } = server.p2p
const { registry } = server.coreData
const selfKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
// Store federation record with mode
await registry.put(`federation:${remoteInfo.serverKey}`, {
serverKey: remoteInfo.serverKey,
displayName: remoteInfo.hostname || remoteInfo.displayName || null,
httpUrl: remoteInfo.httpUrl || null,
mode,
linkedAt: Date.now(),
lastSeenAt: Date.now(),
status: 'active'
})
console.log(`[Peer] Server linked (${mode}): ${remoteInfo.serverKey.slice(0, 16)}... (${remoteInfo.hostname || 'unknown'})`)
if (mode === 'mirror') {
// Mirror mode: join ALL shared Autobases
// 1. Identity Registry
if (remoteInfo.identityRegistryKey) {
await joinOrSkipRegistry(server, 'identity', remoteInfo.identityRegistryKey, remoteInfo)
}
// 2. Namespace Registry
if (remoteInfo.namespaceRegistryKey) {
await joinOrSkipRegistry(server, 'namespace', remoteInfo.namespaceRegistryKey, remoteInfo)
}
// 3. Deploy Registry
if (remoteInfo.deployRegistryKey) {
await joinOrSkipRegistry(server, 'deploy', remoteInfo.deployRegistryKey, remoteInfo)
}
} else {
// Peer mode: only join namespace registry for browsing
if (remoteInfo.namespaceRegistryKey) {
await joinOrSkipRegistry(server, 'namespace', remoteInfo.namespaceRegistryKey, remoteInfo)
}
}
// Add as namespace registry writer if applicable
if (server.namespaceRegistry && remoteInfo.serverKey) {
try {
const peerId = remoteInfo.serverKey.slice(0, 16)
// DANGER: serverKey bug — passes Ed25519 identity key, not Hypercore key.
await server.namespaceRegistry.addWriter(peerId, remoteInfo.serverKey, 'writer')
} catch (err) {
console.error('[Peer] Failed to add registry writer:', err.message)
}
}
}
/**
* Join or skip a registry based on whether we already have one with the same key.
*/
async function joinOrSkipRegistry (server, type, remoteKey, remoteInfo) {
const { registry, store } = server.coreData
const { repoRegistry } = server.repoData
const { driveSwarm } = server.p2p
const configKey = `config:${type}Registry`
// Check if we already have this exact registry
const existing = type === 'identity' ? server.identityRegistry
: type === 'namespace' ? server.namespaceRegistry
: type === 'deploy' ? server.deployRegistry
: null
if (existing) {
if (existing.bootstrapKeyHex === remoteKey) {
console.log(`[Peer] ${type} registry already shared (same key)`)
return
}
// Registry differs — check if local is empty (fresh boot, no real data)
// If empty, close it and join the remote one instead
let isEmpty = false
try {
if (type === 'identity') {
const users = await existing.listAllUsers()
isEmpty = users.length === 0
} else if (type === 'namespace') {
const groups = await existing.list()
isEmpty = groups.length === 0
} else if (type === 'deploy') {
const servers = await existing.listServers()
isEmpty = servers.length === 0
}
} catch { isEmpty = true }
if (isEmpty) {
// Local registry is empty — safe to replace with remote
console.log(`[Peer] ${type} registry is empty, joining remote: ${remoteKey.slice(0, 16)}...`)
try { await existing.close() } catch {}
// Clear the server reference so the join code below creates a new one
if (type === 'identity') server.identityRegistry = null
else if (type === 'namespace') server.namespaceRegistry = null
else if (type === 'deploy') server.deployRegistry = null
// Fall through to join code
} else {
// Local has data — don't replace, store mismatch for repair
console.log(`[Peer] ${type} registry differs — local: ${existing.bootstrapKeyHex?.slice(0, 16)}, remote: ${remoteKey.slice(0, 16)}`)
await registry.put(`peer-registry:${type}:${remoteInfo.serverKey}`, {
remoteKey,
localKey: existing.bootstrapKeyHex,
detectedAt: Date.now()
})
return
}
}
// With unified MetadataBase, all registries are always available as facades.
// Separate registry joining is no longer needed — MetadataBase replication handles everything.
console.log(`[Peer] Registry type '${type}' handled by unified MetadataBase — no separate join needed`)
}
/**
* GET /api/peer/list
* List all peered servers.
*/
async function handlePeerList (req, res, server) {
const { serverIdentity } = server.p2p
try {
const peers = await listPeers(server)
const selfKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
json(res, { peers, selfKey, count: peers.length })
} catch (err) {
console.error('[Peer] List error:', err)
json(res, { peers: [], selfKey: null, count: 0 })
}
}
/**
* GET /api/peer/status
* Detailed peering status including registry sharing state.
*/
async function handlePeerStatus (req, res, server) {
const { serverIdentity, gitSyncManager } = server.p2p
const { namespaceRegistry } = server.namespaceData
const { deployRegistry } = server.deployData
const { repos, repoRegistry } = server.repoData
try {
const selfKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
const hostname = (() => { try { return os.hostname() } catch { return 'unknown' } })()
const peers = await listPeers(server)
// Gather registry keys
const registries = {
identity: server.identityRegistry
? { key: server.identityRegistry.bootstrapKeyHex, status: 'active' }
: { key: null, status: 'missing' },
namespace: namespaceRegistry
? { key: namespaceRegistry.bootstrapKeyHex, status: 'active' }
: { key: null, status: 'missing' },
deploy: deployRegistry
? { key: deployRegistry.bootstrapKeyHex, status: 'active' }
: { key: null, status: 'missing' }
}
// Count repos
const repoCount = (repos || []).length
const reposDir = repoRegistry ? repoRegistry.gitReposPath : null
// GitSync status (Protomux-based, replaced ServerSync)
const syncActive = !!(gitSyncManager && gitSyncManager._peers.size > 0)
json(res, {
selfKey,
hostname,
peers,
peerCount: peers.length,
registries,
repoCount,
syncActive,
httpUrl: buildHttpUrl(server)
})
} catch (err) {
console.error('[Peer] Status error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* DELETE /api/peer/:serverKey
* Remove a peered server.
*/
async function handlePeerRemove (req, res, server) {
const { registry } = server.coreData
const { serverKey } = req.params
try {
const entry = await registry.get(`federation:${serverKey}`)
if (!entry) {
return notFound(res, 'Peered server not found')
}
// Mark as revoked
const record = entry.value
record.status = 'revoked'
record.revokedAt = Date.now()
await registry.put(`federation:${serverKey}`, record)
json(res, { success: true, serverKey })
} catch (err) {
console.error('[Peer] Remove error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* POST /api/peer/repair
* Detect and fix holes in partially merged peering state.
*/
async function handlePeerRepair (req, res, server) {
const { serverIdentity, gitSyncManager } = server.p2p
const { registry } = server.coreData
let body
try {
body = await readJson(req)
} catch {
body = {}
}
try {
const selfKey = serverIdentity
? serverIdentity.getPublicKeyHex()
: null
if (!selfKey) {
return badRequest(res, 'Server identity not initialized')
}
const peers = await listPeers(server)
const mirrorPeers = peers.filter(p => p.mode === 'mirror' && p.status === 'active')
if (mirrorPeers.length === 0) {
return json(res, {
success: false,
message: 'No active mirror peers found. Use pgh peer add --mirror first.',
repairs: []
})
}
const repairs = []
// Check federation entries
if (peers.length === 0) {
repairs.push({
type: 'federation',
status: 'missing',
description: 'No federation entries — ServerSync has no peers',
action: 'Create federation entries with HTTP URLs'
})
}
// Check each registry type for mismatches with mirror peers
for (const peer of mirrorPeers) {
// Read stored peer registry info
for (const regType of ['identity', 'namespace', 'deploy']) {
const peerRegEntry = await registry.get(`peer-registry:${regType}:${peer.serverKey}`)
if (peerRegEntry && peerRegEntry.value) {
const { remoteKey, localKey } = peerRegEntry.value
repairs.push({
type: regType,
status: 'independent',
description: `${regType} registry differs — local: ${localKey?.slice(0, 16)}..., remote: ${remoteKey.slice(0, 16)}...`,
action: `Adopt remote key (requires restart)`,
localKey,
remoteKey,
peerServerKey: peer.serverKey
})
}
}
// Check if peer has serverKey (needed for P2P sync)
if (!peer.serverKey) {
repairs.push({
type: 'federation-key',
status: 'missing',
description: `Peer ${(peer.displayName || 'unknown')} has no serverKey`,
action: 'Need serverKey for P2P sync via punch://'
})
}
}
// Check active GitSync
if (!gitSyncManager || gitSyncManager._peers.size === 0) {
repairs.push({
type: 'sync',
status: 'inactive',
description: 'Git sync has no connected peers',
action: 'Will activate when a federation peer connects via driveSwarm'
})
}
// Apply fixes for the simple cases
const applyFixes = body.apply !== false // default: apply
if (applyFixes) {
for (const repair of repairs) {
if (repair.type === 'identity' && repair.status === 'independent' && repair.remoteKey) {
// Adopt remote identity registry key
await registry.put('config:identityRegistry', {
bootstrapKey: repair.remoteKey,
adoptedFrom: repair.peerServerKey,
adoptedAt: Date.now()
})
repair.applied = true
repair.note = 'Restart required to load new identity registry'
}
if (repair.type === 'deploy' && repair.status === 'independent' && repair.remoteKey) {
// Adopt remote deploy registry key
await registry.put('config:deployRegistry', {
bootstrapKey: repair.remoteKey,
adoptedFrom: repair.peerServerKey,
adoptedAt: Date.now()
})
repair.applied = true
repair.note = 'Restart required to load new deploy registry'
}
// Clean up the peer-registry detection entry after applying
if (repair.applied && repair.peerServerKey) {
await registry.del(`peer-registry:${repair.type}:${repair.peerServerKey}`)
}
}
}
const needsRestart = repairs.some(r => r.applied)
json(res, {
success: true,
repairs,
needsRestart,
message: repairs.length === 0
? 'All registries are properly shared. No repairs needed.'
: `Found ${repairs.length} issue(s).${needsRestart ? ' Restart both servers to complete repair.' : ''}`
})
} catch (err) {
console.error('[Peer] Repair error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* POST /api/peer/sync
* Force immediate sync with all peers.
*/
async function handlePeerSync (req, res, server) {
const { gitSyncManager } = server.p2p
try {
if (!gitSyncManager) {
return json(res, { success: false, message: 'Git sync not initialized' })
}
// Trigger full-sync-request to all connected peers
for (const [peerKey, { server: syncServer }] of gitSyncManager._peers) {
try {
await gitSyncManager._initiateFullSync(syncServer, peerKey)
} catch (err) {
console.error(`[Peer] Sync failed for ${peerKey.slice(0, 8)}: ${err.message}`)
}
}
json(res, { success: true, message: `Sync triggered for ${gitSyncManager._peers.size} peer(s)` })
} catch (err) {
console.error('[Peer] Sync error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* POST /api/peer/sync/:repoName
* Push-triggered sync: notify all federation peers to fetch a specific repo.
*/
async function handlePeerSyncRepo (req, res, server) {
const { repoName } = req.params
if (!repoName) {
return json(res, { error: 'repoName required' }, 400)
}
try {
const { notifyPeersOfDeploy } = await import('../ci/deploy-hook.js')
await notifyPeersOfDeploy(server, repoName)
json(res, { success: true, message: `Sync notification sent for ${repoName}` })
} catch (err) {
console.error('[Peer] Sync repo error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* GET /api/peer/sync-log
* Return recent sync events (sent and received).
*/
async function handlePeerSyncLog (req, res, server) {
const { registry } = server.coreData
try {
const entries = []
const stream = registry.createReadStream({
gt: 'sync-log:',
lt: 'sync-log:\xff',
reverse: true,
limit: 50
})
for await (const entry of stream) {
entries.push(entry.value)
}
json(res, { entries, count: entries.length })
} catch (err) {
json(res, { entries: [], count: 0 })
}
}
/**
* POST /api/peer/:serverKey/grant
* Grant repo or group access to a peer (peer mode only).
* Body: { repos?: string[], groups?: string[] }
*/
async function handlePeerGrant (req, res, server) {
const { registry } = server.coreData
const { serverKey } = req.params
let body
try {
body = await readJson(req)
} catch (err) {
return badRequest(res, 'Invalid JSON: ' + err.message)
}
try {
const entry = await registry.get(`federation:${serverKey}`)
if (!entry) {
return notFound(res, 'Peered server not found')
}
const record = entry.value
if (record.mode !== 'peer') {
return badRequest(res, 'Grant only applies to peer mode (mirrors share everything)')
}
// Update granted repos/groups
if (!record.grantedRepos) record.grantedRepos = []
if (!record.grantedGroups) record.grantedGroups = []
if (body.repos) {
for (const repo of body.repos) {
if (!record.grantedRepos.includes(repo)) {
record.grantedRepos.push(repo)
}
}
}
if (body.groups) {
for (const group of body.groups) {
if (!record.grantedGroups.includes(group)) {
record.grantedGroups.push(group)
}
}
}
record.updatedAt = Date.now()
await registry.put(`federation:${serverKey}`, record)
json(res, {
success: true,
serverKey,
grantedRepos: record.grantedRepos,
grantedGroups: record.grantedGroups
})
} catch (err) {
console.error('[Peer] Grant error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* POST /api/peer/:serverKey/upgrade
* Upgrade a peer to mirror mode.
*/
async function handlePeerUpgrade (req, res, server) {
const { registry } = server.coreData
const { serverKey } = req.params
try {
const entry = await registry.get(`federation:${serverKey}`)
if (!entry) {
return notFound(res, 'Peered server not found')
}
const record = entry.value
if (record.mode === 'mirror') {
return json(res, { success: true, message: 'Already in mirror mode' })
}
// Upgrade to mirror
record.mode = 'mirror'
record.upgradedAt = Date.now()
await registry.put(`federation:${serverKey}`, record)
// Now share all registries that weren't shared before
// This requires the remote server to also have these keys,
// which will happen when they accept the upgrade or on next repair
console.log(`[Peer] Upgraded ${serverKey.slice(0, 16)}... to mirror mode`)
json(res, {
success: true,
serverKey,
mode: 'mirror',
message: 'Upgraded to mirror mode. Run pgh peer repair to sync registries.'
})
} catch (err) {
console.error('[Peer] Upgrade error:', err)
json(res, { error: err.message }, 500)
}
}
/**
* Helper: List all federation peers from Hyperbee
*/
async function listPeers (server) {
const { registry } = server.coreData
const peers = []
try {
for await (const entry of registry.createReadStream({
gt: 'federation:',
lt: 'federation:\xff'
})) {
const data = typeof entry.value === 'string' ? JSON.parse(entry.value) : entry.value
peers.push(data)
}
} catch {}
return peers
}