dat-thang/tests/interop/cabal/cable_hyperswarm.js

'use strict'

// Cable's SDL-facing bridge speaks S-expressions on stdin/stdout. Cabal Core
// owns the network boundary and continues to exchange its canonical JSON
// Hypercore blocks over Hyperswarm.
const fs = require('fs')
const { createRequire } = require('module')
const os = require('os')
const path = require('path')
const readline = require('readline')

function loadCabalCore () {
  const configured = process.env.CABLE_CABAL_CORE || process.env.CABAL_JS_MODULE
  if (configured) {
    const entry = require.resolve(configured)
    const requireCabal = createRequire(entry)
    return { Cabal: requireCabal(entry), requireCabal }
  }
  try {
    const entry = require.resolve('cabal-core')
    const requireCabal = createRequire(entry)
    return { Cabal: requireCabal(entry), requireCabal }
  } catch (_) {}
  const home = os.homedir()
  const candidates = [
    path.join(home, 'AppData', 'Roaming', 'npm', 'node_modules', 'cabal-cli', 'node_modules', 'cabal-core'),
    path.join(home, '.npm-global', 'lib', 'node_modules', 'cabal-cli', 'node_modules', 'cabal-core'),
    '/usr/local/lib/node_modules/cabal-cli/node_modules/cabal-core',
    '/usr/lib/node_modules/cabal-cli/node_modules/cabal-core'
  ]
  for (const candidate of candidates) {
    if (fs.existsSync(path.join(candidate, 'index.js'))) {
      const entry = path.join(candidate, 'index.js')
      const requireCabal = createRequire(entry)
      return { Cabal: requireCabal(entry), requireCabal }
    }
  }
  throw new Error('Cabal Core not found; install cabal-cli or set CABLE_CABAL_CORE')
}

function parseBootstrap (requireCabal) {
  const setting = process.env.CABLE_BOOTSTRAP || ''
  const configPath = path.join(os.homedir(), '.config', 'dat-gpt', 'cable-bootstrap')
  let saved = ''
  try { saved = fs.readFileSync(configPath, 'utf8') } catch (err) {
    if (err.code !== 'ENOENT') throw err
  }
  const endpoints = (setting + '\n' + saved).split(/[;,\s]+/).filter(Boolean)
  if (!endpoints.length) return requireCabal('./bootstrap_nodes')
  return Array.from(new Set(endpoints))
}

// Mirrors Cabal Core's swarm.js replication and peer-id extension, while
// allowing the same DHT bootstrap list to be selected by the local client.
async function startHyperswarm (cabal, networkOptions = {}) {
  const loaded = loadCabalCore()
  const DHT = loaded.requireCabal('hyperdht')
  const Swarm = loaded.requireCabal('hyperswarm')
  const crypto = loaded.requireCabal('hypercore-crypto')
  const pump = loaded.requireCabal('pump')
  const localKey = await new Promise((resolve, reject) =>
    cabal.getLocalKey((err, key) => err ? reject(err) : resolve(key)))
  const dhtOptions = { bootstrap: parseBootstrap(loaded.requireCabal) }
  const host = networkOptions.host || process.env.CABLE_DHT_HOST
  const port = networkOptions.port || process.env.CABLE_DHT_PORT
  if (host) dhtOptions.host = host
  if (port) {
    const numericPort = Number(port)
    if (!Number.isInteger(numericPort) || numericPort < 1 || numericPort > 65535)
      throw new Error('CABLE_DHT_PORT must be an integer from 1 to 65535')
    dhtOptions.port = numericPort
  }
  const advertiseHost = networkOptions.advertiseHost || process.env.CABLE_DHT_ADVERTISE_HOST
  if (advertiseHost) dhtOptions.firewalled = false
  const dht = new DHT(dhtOptions)
  if (advertiseHost && Number.isInteger(dhtOptions.port) && dht._nat && typeof dht._nat.add === 'function')
    dht._nat.add(advertiseHost, dhtOptions.port)
  const swarm = new Swarm({ dht })
  mediaExtensions.clear()
  mediaSeen.clear()
  swarm.on('error', err => {
    if (process.env.CABLE_DEBUG) process.stderr.write('Cable Hyperswarm error: ' + err.message + '\n')
  })
  const discoveryKey = crypto.discoveryKey(Buffer.from(cabal.key, 'hex'))
  if (process.env.CABLE_DEBUG) process.stderr.write('Cable swarm pid=' + process.pid + ' room=' + cabal.key +
    ' discovery=' + discoveryKey.toString('hex') + '\n')
  const discovery = swarm.join(discoveryKey, { server: true, client: true })
  let shuttingDown = false
  swarm.__cabalDiscovery = discovery
  swarm.__markShutdown = () => { shuttingDown = true }
  swarm.__shutdown = async function () {
    swarm.__markShutdown()
    await discovery.destroy()
    await swarm.destroy()
  }
  swarm.on('connection', (socket, info) => {
    if (process.env.CABLE_DEBUG) process.stderr.write('Cable Hyperswarm connection opened pid=' + process.pid +
      ' client=' + Boolean(info.client) + ' peer=' + (info.publicKey ? info.publicKey.toString('hex') : 'unknown') + '\n')
    socket.on('error', err => {
      if (process.env.CABLE_DEBUG) process.stderr.write('Cable Hyperswarm socket error: ' + err.message + '\n')
    })
    let remoteKey
    const replication = cabal.replicate(info.client)
    if (process.env.CABLE_DEBUG) replication.on('error', err => process.stderr.write('Cable Hypercore replication error: ' + err.message + '\n'))
    pump(socket, replication, socket, err => {
      if (err && !shuttingDown && err.message !== 'Readable stream closed' && err.code !== 'ECONNRESET')
        process.stderr.write('Cable replication: ' + err.message + '\n')
      if (remoteKey) cabal._removeConnection(remoteKey)
    })
    const extension = replication.registerExtension('peer-id', {
      encoding: 'json',
      onmessage (message) {
        if (remoteKey || !message || !message.id) return
        const key = Buffer.from(message.id, 'hex')
        if (key.length !== 32) return
        remoteKey = message.id
        cabal._addConnection(remoteKey)
        if (process.env.CABLE_DEBUG) process.stderr.write('Cable Cabal peer id ' + remoteKey + '\n')
      }
    })
    const liveExtension = replication.registerExtension('cable-live', {
      onmessage (message) {
        const packet = Buffer.isBuffer(message) ? message : Buffer.from(message)
        if (packet.length > 4096) return
        relayMediaPacket(packet, liveExtension)
        if (packet.length >= 42 && packet.toString('ascii', 0, 4) === 'CBLM')
          emit('live', remoteKey || '', packet.toString('base64'))
      }
    })
    mediaExtensions.add(liveExtension)
    socket.once('close', () => mediaExtensions.delete(liveExtension))
    extension.send({ id: localKey })
    if (process.env.CABLE_DEBUG) process.stderr.write('Cable local peer id sent: ' + localKey + '\n')
  })
  cabal._swarm = swarm
  if (networkOptions.waitForBootstrap || process.env.CABLE_WAIT_DHT === '1') {
    await dht.fullyBootstrapped()
    await swarm.flush()
  }
  if (!(await discovery.flushed())) {
    await swarm.__shutdown()
    throw new Error('Hyperswarm topic discovery did not complete')
  }
  return swarm
}

function quote (value) {
  const text = String(value == null ? '' : value)
  let quoted = '"'
  for (const character of text) {
    switch (character) {
      case '"': quoted += '\\"'; break
      case '\\': quoted += '\\\\'; break
      case '\b': quoted += '\\b'; break
      case '\f': quoted += '\\f'; break
      case '\n': quoted += '\\n'; break
      case '\r': quoted += '\\r'; break
      case '\t': quoted += '\\t'; break
      default: quoted += character
    }
  }
  return quoted + '"'
}

function emit (kind, ...fields) {
  process.stdout.write('(event ' + kind + (fields.length ? ' ' + fields.map(value =>
    typeof value === 'number' ? String(value) : quote(value)).join(' ') : '') + ')\n')
}

function parse (source) {
  let at = 0
  function white () { while (/\s/.test(source[at] || '')) at++ }
  function value () {
    white()
    if (source[at] === '(') {
      at++
      const list = []
      while (true) {
        white()
        if (source[at] === ')') { at++; return list }
        if (at >= source.length) throw new Error('unterminated S-expression')
        list.push(value())
      }
    }
    if (source[at] === '"') {
      at++
      let text = ''
      while (at < source.length) {
        const c = source[at++]
        if (c === '"') return text
        if (c !== '\\') { text += c; continue }
        if (at >= source.length) throw new Error('incomplete S-expression escape')
        const escape = source[at++]
        switch (escape) {
          case '"': text += '"'; break
          case '\\': text += '\\'; break
          case 'b': text += '\b'; break
          case 'f': text += '\f'; break
          case 'n': text += '\n'; break
          case 'r': text += '\r'; break
          case 't': text += '\t'; break
          default: throw new Error('unsupported S-expression escape: \\' + escape)
        }
      }
      throw new Error('unterminated S-expression string')
    }
    const begin = at
    while (at < source.length && !/[\s()]/.test(source[at])) at++
    if (begin === at) throw new Error('expected S-expression atom')
    const atom = source.slice(begin, at)
    if (/^-?[0-9]+$/.test(atom)) return Number(atom)
    return atom
  }
  const result = value()
  white()
  if (at !== source.length) throw new Error('trailing S-expression data')
  return result
}

let cabal = null
let localKey = null
let starting = false
const names = new Map()
const pending = []
const mediaExtensions = new Set()
const mediaSeen = new Map()

function markMediaPacket (packet) {
  if (!Buffer.isBuffer(packet) || packet.length < 42 || packet.length > 4096 ||
      packet.toString('ascii', 0, 4) !== 'CBLM' || packet[4] !== 1 ||
      packet[5] < 1 || packet[5] > 6 || packet.readUInt16BE(40) !== packet.length - 42) return false
  const id = packet.subarray(8, 32).toString('hex')
  if (mediaSeen.has(id)) return false
  mediaSeen.set(id, true)
  if (mediaSeen.size > 2048) mediaSeen.delete(mediaSeen.keys().next().value)
  return true
}

function relayMediaPacket (packet, source) {
  if (!markMediaPacket(packet)) return
  for (const extension of mediaExtensions) {
    if (extension === source) continue
    try { extension.send(packet) } catch (_) {}
  }
}

function channelName (msg) {
  return msg && msg.value && msg.value.content && msg.value.content.channel
}

async function start (roomKey, storage, nick, channel) {
  if (starting || cabal) throw new Error('Cabal room already started')
  starting = true
  try {
    if (process.env.CABLE_DEBUG) process.stderr.write('Cable Core storage: ' + storage + '\n')
    const { Cabal } = loadCabalCore()
    fs.mkdirSync(storage, { recursive: true })
    cabal = Cabal(storage, roomKey)
    await new Promise((resolve, reject) => cabal.ready(err => err ? reject(err) : resolve()))
    localKey = await new Promise((resolve, reject) => cabal.getLocalKey((err, key) => err ? reject(err) : resolve(key)))

    cabal.on('peer-added', key => emit('peer-up', key))
    cabal.on('peer-dropped', key => emit('peer-down', key))
    cabal.kcore.api.users.events.on('update', (key, msg) => {
      if (key === localKey) return
      const name = msg && msg.value && msg.value.content && msg.value.content.name
      if (typeof name === 'string') names.set(key, name)
      emit('nick', key, names.get(key) || key.slice(0, 8))
    })
    cabal.kcore.api.messages.events.on('message', msg => {
      const val = msg && msg.value
      const content = val && val.content
      const key = msg && msg.key
      if (!val || !content || key === localKey || !key) return
      if (val.type === 'chat/text' && typeof content.channel === 'string' && typeof content.text === 'string') {
        emit('message', key, names.get(key) || key.slice(0, 8), content.channel,
          content.text, val.timestamp || 0)
      } else if (val.type === 'channel/join' && typeof content.channel === 'string') {
        emit('joined', key, names.get(key) || key.slice(0, 8), content.channel)
      }
    })

    await startHyperswarm(cabal)
    emit('ready', localKey)
    await new Promise((resolve, reject) => cabal.publishNick(nick, err => err ? reject(err) : resolve()))
    await publish({ type: 'channel/join', content: { channel } })
    emit('joined-local', channel)
    for (const command of pending.splice(0)) await handle(command)
  } finally {
    starting = false
  }
}

function publish (message) {
  return new Promise((resolve, reject) => cabal.publish(message, err => err ? reject(err) : resolve()))
}

async function handle (command) {
  if (!Array.isArray(command) || !command.length) throw new Error('expected a command list')
  const [verb, ...args] = command
  if (verb === 'start') return start(...args)
  if (verb === 'close') {
    if (cabal && cabal._swarm && cabal._swarm.__shutdown) await cabal._swarm.__shutdown()
    if (cabal) await cabal.close()
    cabal = null
    localKey = null
    emit('closed')
    return
  }
  if (!cabal) {
    if (starting) { pending.push(command); return }
    throw new Error('Cabal room is not started')
  }
  if (verb === 'send') {
    const [channel, text] = args
    if (typeof channel !== 'string' || typeof text !== 'string') throw new Error('send expects channel and text strings')
    return publish({ type: 'chat/text', content: { channel, text } })
  }
  if (verb === 'nick') {
    const [nick] = args
    if (typeof nick !== 'string') throw new Error('nick expects a string')
    return new Promise((resolve, reject) => cabal.publishNick(nick, err => err ? reject(err) : resolve()))
  }
  if (verb === 'join') {
    const [channel] = args
    if (typeof channel !== 'string') throw new Error('join expects a channel string')
    return publish({ type: 'channel/join', content: { channel } })
  }
  if (verb === 'live') {
    const [encoded] = args
    if (typeof encoded !== 'string' || encoded.length > 5500)
      throw new Error('live expects one bounded base64 media packet')
    const packet = Buffer.from(encoded, 'base64')
    if (!markMediaPacket(packet)) throw new Error('invalid or repeated Cable live packet')
    for (const extension of mediaExtensions) {
      try { extension.send(packet) } catch (_) {}
    }
    return
  }
  throw new Error('unknown Cable bridge command: ' + verb)
}

module.exports = { startHyperswarm, loadCabalCore, quote, parse }

if (require.main === module) {
  process.stdout.on('error', err => {
    if (err && err.code === 'EPIPE') process.exit(0)
    throw err
  })
  const input = readline.createInterface({ input: process.stdin, crlfDelay: Infinity })
  input.on('line', line => {
    if (!line.trim()) return
    let result
    try { result = handle(parse(line)) } catch (err) { emit('error', err.message); return }
    Promise.resolve(result).catch(err => emit('error', err && err.message ? err.message : String(err)))
  })
  input.on('close', async () => {
    if (cabal && cabal._swarm && cabal._swarm.__shutdown) { try { await cabal._swarm.__shutdown() } catch (_) {} }
    if (cabal) { try { await cabal.close() } catch (_) {} }
    process.exit(0)
  })
}