#!/usr/bin/env node
'use strict'
//
// atomspace-p2p-bridge — CogServer over Hyperswarm
//
// Identity-first design: uses hyperident profiles for persistent,
// deterministic P2P identity. Both server and client derive the
// same DHT topic from a profile name — no manual key exchange.
//
// Derivation chain:
// hyperident mnemonic → SLIP-21(profileName) → seed
// seed + "atomspace-bridge" salt → Ed25519 keypair
// discoveryKey(publicKey) → DHT topic
//
// Server: runs alongside CogServer, announces on DHT
// Client: creates local TCP proxy, connects via DHT lookup
//
// CogStorageNode ─TCP─► bridge client ─Hyperswarm─► bridge server ─TCP─► CogServer
//
const Hyperswarm = require('hyperswarm')
const createTestnet = require('@hyperswarm/testnet')
const Protomux = require('protomux')
const c = require('compact-encoding')
const net = require('net')
const crypto = require('hypercore-crypto')
const b4a = require('b4a')
const fs = require('fs')
const path = require('path')
// hyperident is ESM — loaded dynamically
let _identityManagerPromise = null
function getIdentityManager () {
if (!_identityManagerPromise) {
_identityManagerPromise = import('hyperident/lib/identity-manager.js')
.then(m => m.IdentityManager)
.catch(() => null)
}
return _identityManagerPromise
}
const PROTOCOL = 'atomspace-proxy'
const SALT = 'atomspace-bridge' // app-specific namespace
// ---- Identity derivation ----
// Derives a deterministic Ed25519 keypair from a hyperident profile.
// The salt ensures this keypair is unique to the atomspace bridge
// (other apps using the same profile get different keys).
async function keyPairFromProfile (profileName) {
const IdentityManager = await getIdentityManager()
if (!IdentityManager) {
console.error('Error: hyperident not available')
console.error('Run: npm install hyperident')
process.exit(1)
}
const profileTopic = await IdentityManager.deriveTopicFromProfile(profileName)
const seed = crypto.hash(b4a.concat([profileTopic, b4a.from(SALT, 'utf8')]))
return crypto.keyPair(seed)
}
// Fallback: keypair from raw 32-byte seed
function keyPairFromSeed (seed) {
return crypto.keyPair(seed)
}
// ---- Session message encoding ----
function packSession (id, data) {
const buf = b4a.allocUnsafe(4 + data.length)
buf.writeUInt32BE(id, 0)
b4a.copy(data, buf, 4)
return buf
}
function unpackSession (buf) {
return {
id: buf.readUInt32BE(0),
data: buf.subarray(4)
}
}
// ---- Testnet bootstrap ----
async function resolveBootstrap (testnetFile) {
if (!testnetFile) return null
const testnetPath = path.resolve(testnetFile)
if (fs.existsSync(testnetPath)) {
const data = JSON.parse(fs.readFileSync(testnetPath, 'utf8'))
console.log('Loaded testnet: ' + data.bootstrap.length + ' bootstrap nodes')
return data.bootstrap
}
const testnet = await createTestnet(3)
const bootstrap = testnet.bootstrap.map(n => ({ host: n.host, port: n.port }))
fs.writeFileSync(testnetPath, JSON.stringify({ bootstrap }, null, 2) + '\n')
console.log('Created testnet: ' + bootstrap.length + ' bootstrap nodes → ' + testnetPath)
return bootstrap
}
// ---- CLI ----
const args = process.argv.slice(2)
const mode = args[0]
function getFlag (name, fallback) {
const i = args.indexOf(name)
return (i !== -1 && i + 1 < args.length) ? args[i + 1] : fallback
}
if (!mode || !['server', 'client'].includes(mode)) {
console.log('AtomSpace P2P Bridge — CogServer over Hyperswarm')
console.log('')
console.log('Usage:')
console.log(' node bridge.js server --profile <name> [options]')
console.log(' node bridge.js client --profile <name> [options]')
console.log(' node bridge.js client <public-key> [options]')
console.log('')
console.log('Identity (pick one):')
console.log(' --profile <name> Hyperident profile (recommended)')
console.log(' --seed <hex> Raw 64-char hex seed (fallback)')
console.log('')
console.log('Server options:')
console.log(' --port <n> CogServer port (default: 17001)')
console.log(' --testnet <file> Use local testnet for testing')
console.log('')
console.log('Client options:')
console.log(' --proxy-port <n> Local proxy port (default: 17002)')
console.log(' --testnet <file> Use local testnet (must match server)')
console.log('')
console.log('Examples:')
console.log(' # Both sides use same profile — no key exchange needed:')
console.log(' node bridge.js server --profile tabertron')
console.log(' node bridge.js client --profile tabertron')
console.log('')
console.log(' # Manual key (no hyperident):')
console.log(' node bridge.js server --seed <64-hex-chars>')
console.log(' node bridge.js client <public-key>')
console.log('')
console.log(' # Then in Guile:')
console.log(' (define csn (CogStorageNode "cog://localhost:17002/"))')
console.log(' (cog-open csn)')
process.exit(0)
}
if (mode === 'server') runServer()
else runClient()
// ============================================================
// SERVER
// ============================================================
async function runServer () {
const cogPort = parseInt(getFlag('--port', '17001'))
const profileName = getFlag('--profile', null)
const seedHex = getFlag('--seed', null)
const testnetFile = getFlag('--testnet', null)
const bootstrap = await resolveBootstrap(testnetFile)
// Resolve identity — profile is primary, seed is fallback
let keyPair
if (profileName) {
keyPair = await keyPairFromProfile(profileName)
} else if (seedHex) {
keyPair = keyPairFromSeed(b4a.from(seedHex, 'hex'))
} else {
console.error('Error: identity required')
console.error('Use --profile <name> (recommended) or --seed <hex>')
process.exit(1)
}
const topic = crypto.discoveryKey(keyPair.publicKey)
const swarmOpts = { keyPair }
if (bootstrap) swarmOpts.bootstrap = bootstrap
const swarm = new Hyperswarm(swarmOpts)
let peerCount = 0
let totalSessions = 0
console.log('=== AtomSpace P2P Bridge: Server ===')
if (profileName) console.log('Profile: ' + profileName)
if (bootstrap) console.log('Testnet: ' + bootstrap.length + ' nodes')
console.log('Public key: ' + b4a.toString(keyPair.publicKey, 'hex'))
console.log('Topic: ' + b4a.toString(topic, 'hex'))
console.log('CogServer: localhost:' + cogPort)
const discovery = swarm.join(topic, { server: true, client: false })
await discovery.flushed()
console.log('DHT announce complete — waiting for peers...\n')
swarm.on('connection', (socket, info) => {
const peerId = b4a.toString(info.publicKey, 'hex').slice(0, 12)
peerCount++
console.log('[+] Peer ' + peerId + ' (total: ' + peerCount + ')')
const mux = Protomux.from(socket)
const sessions = new Map()
const channel = mux.createChannel({
protocol: PROTOCOL,
onopen () {
console.log(' Channel opened with ' + peerId)
},
onclose () {
console.log(' Channel closed with ' + peerId)
for (const [, tcp] of sessions) tcp.destroy()
sessions.clear()
}
})
// Message 0: open session (client → server)
channel.addMessage({
encoding: c.uint32,
onmessage (sessionId) {
totalSessions++
const tcp = net.connect(cogPort, '127.0.0.1')
tcp.on('connect', () => {
console.log(' Session ' + sessionId + ' → CogServer:' + cogPort +
' (total: ' + totalSessions + ')')
})
tcp.on('data', (buf) => {
try { dataMsg.send(packSession(sessionId, buf)) } catch (e) {}
})
tcp.on('error', (err) => {
console.log(' Session ' + sessionId + ' TCP error: ' + err.message)
try { closeMsg.send(sessionId) } catch (e) {}
sessions.delete(sessionId)
})
tcp.on('close', () => {
try { closeMsg.send(sessionId) } catch (e) {}
sessions.delete(sessionId)
})
sessions.set(sessionId, tcp)
}
})
// Message 1: data (bidirectional)
const dataMsg = channel.addMessage({
encoding: c.raw,
onmessage (buf) {
const msg = unpackSession(buf)
const tcp = sessions.get(msg.id)
if (tcp && !tcp.destroyed) tcp.write(msg.data)
}
})
// Message 2: close session (bidirectional)
const closeMsg = channel.addMessage({
encoding: c.uint32,
onmessage (sessionId) {
const tcp = sessions.get(sessionId)
if (tcp) {
tcp.destroy()
sessions.delete(sessionId)
}
}
})
channel.open()
socket.on('close', () => {
peerCount--
console.log('[-] Peer ' + peerId + ' (total: ' + peerCount + ')')
for (const [, tcp] of sessions) tcp.destroy()
sessions.clear()
})
socket.on('error', (err) => {
console.log('[!] Peer ' + peerId + ' error: ' + err.message)
})
})
}
// ============================================================
// CLIENT
// ============================================================
async function runClient () {
const profileName = getFlag('--profile', null)
const seedHex = getFlag('--seed', null)
const proxyPort = parseInt(getFlag('--proxy-port', '17002'))
const testnetFile = getFlag('--testnet', null)
const bootstrap = await resolveBootstrap(testnetFile)
// Resolve server identity — profile derives the same key the server uses
let serverKey
if (profileName) {
const keyPair = await keyPairFromProfile(profileName)
serverKey = keyPair.publicKey
} else if (seedHex) {
const keyPair = keyPairFromSeed(b4a.from(seedHex, 'hex'))
serverKey = keyPair.publicKey
} else {
// Raw public key as positional argument
const serverKeyHex = args[1]
if (!serverKeyHex || serverKeyHex.length !== 64 || serverKeyHex.startsWith('--')) {
console.error('Error: identity required')
console.error('Use --profile <name> (recommended) or provide <public-key>')
process.exit(1)
}
serverKey = b4a.from(serverKeyHex, 'hex')
}
const topic = crypto.discoveryKey(serverKey)
const swarmOpts = {}
if (bootstrap) swarmOpts.bootstrap = bootstrap
const swarm = new Hyperswarm(swarmOpts)
swarm.join(topic, { server: false, client: true })
console.log('=== AtomSpace P2P Bridge: Client ===')
if (profileName) console.log('Profile: ' + profileName)
if (bootstrap) console.log('Testnet: ' + bootstrap.length + ' nodes')
console.log('Server key: ' + b4a.toString(serverKey, 'hex').slice(0, 24) + '...')
console.log('Topic: ' + b4a.toString(topic, 'hex').slice(0, 24) + '...')
console.log('Proxy port: ' + proxyPort)
console.log('Looking up server on DHT...\n')
let channel = null
let openMsg = null
let dataMsg = null
let closeMsg = null
const localSessions = new Map()
let nextSessionId = 1
const pending = []
swarm.on('connection', (socket, info) => {
const peerId = b4a.toString(info.publicKey, 'hex').slice(0, 12)
console.log('[+] Connected to server ' + peerId)
const mux = Protomux.from(socket)
channel = mux.createChannel({
protocol: PROTOCOL,
onopen () {
console.log(' Channel ready — proxy active')
for (const fn of pending) fn()
pending.length = 0
},
onclose () {
console.log(' Channel closed')
channel = null
openMsg = null
dataMsg = null
closeMsg = null
for (const [, tcp] of localSessions) tcp.destroy()
localSessions.clear()
}
})
openMsg = channel.addMessage({
encoding: c.uint32,
onmessage () {}
})
dataMsg = channel.addMessage({
encoding: c.raw,
onmessage (buf) {
const msg = unpackSession(buf)
const tcp = localSessions.get(msg.id)
if (tcp && !tcp.destroyed) tcp.write(msg.data)
}
})
closeMsg = channel.addMessage({
encoding: c.uint32,
onmessage (sessionId) {
const tcp = localSessions.get(sessionId)
if (tcp) {
tcp.destroy()
localSessions.delete(sessionId)
}
}
})
channel.open()
socket.on('close', () => {
console.log('[-] Server disconnected, waiting for reconnect...')
channel = null
openMsg = null
dataMsg = null
closeMsg = null
})
socket.on('error', (err) => {
console.log('[!] Server error: ' + err.message)
})
})
// Local TCP proxy — CogStorageNode connects here
const proxy = net.createServer((tcp) => {
const sessionId = nextSessionId++
localSessions.set(sessionId, tcp)
if (channel && openMsg) {
openMsg.send(sessionId)
wireSession(sessionId, tcp)
console.log(' Session ' + sessionId + ' → server')
} else {
console.log(' Session ' + sessionId + ' queued (no server yet)')
pending.push(() => {
if (tcp.destroyed) return
openMsg.send(sessionId)
wireSession(sessionId, tcp)
console.log(' Session ' + sessionId + ' → server (dequeued)')
})
}
})
function wireSession (sessionId, tcp) {
tcp.on('data', (buf) => {
try {
if (dataMsg) dataMsg.send(packSession(sessionId, buf))
} catch (e) {}
})
tcp.on('error', () => {
try { if (closeMsg) closeMsg.send(sessionId) } catch (e) {}
localSessions.delete(sessionId)
})
tcp.on('close', () => {
try { if (closeMsg) closeMsg.send(sessionId) } catch (e) {}
localSessions.delete(sessionId)
})
}
proxy.listen(proxyPort, () => {
console.log('Proxy listening on port ' + proxyPort)
console.log('')
console.log('Use in Guile:')
console.log(' (use-modules (opencog persist-cog))')
console.log(' (define csn (CogStorageNode "cog://localhost:' + proxyPort + '/"))')
console.log(' (cog-open csn)')
console.log('')
})
}
// ---- Graceful shutdown ----
process.on('SIGINT', () => {
console.log('\nShutting down...')
process.exit(0)
})
process.on('uncaughtException', (err) => {
console.error('Uncaught:', err.message)
})