atomspace-gpu / experiments / m0-uds-bridge / daemon-ab.js

// hyper-atomspace daemon (Bare) — M4: Autobase multi-writer op-log.
//
// Two agents (daemons), each its own Autobase writer, replicate over a peer UDS
// and converge to ONE linearized log -> ONE AtomSpace projection. This is the
// device-as-agent step: the production form runs this over Hyperswarm (already
// proven in pear-git-v3/test/keet-pattern-e2e.js); here the peer transport is a
// direct UDS pipe for a deterministic gold check.
//
// usage:
//   bare daemon-ab.js <client-sock> <storage-dir> [--bootstrap <hex>]
//                     [--peer-listen <path> | --peer-connect <path>]
//
// Startup prints (for orchestration):
//   AB key=<autobase-key> local=<local-writer-key> writable=<bool>
//
// Client protocol — newline-framed over the client UDS (superset of M0-M3, minus
// clear which an append-only multi-writer log does not support):
//   store <sexpr>\n     append an atom op to THIS writer. no reply (fire-and-forget).
//   load\n              reply: (<sexpr>…)\n   — the linearized view.
//   loadat <K>\n        reply: (<sexpr0>…<sexpr_{K-1}>)\n
//   seqof <sexpr>\n     reply: <seq>\n
//   count\n             reply: <view.length>\n
//   barrier\n           await base.update(); reply: ok\n
//   addwriter <hex>\n   append an add-writer op for that local key; reply: ok\n
//   writable\n          reply: true|false\n

import net from 'bare-net'
import fs from 'bare-fs'
import process from 'bare-process'
import Corestore from 'corestore'
import Autobase from 'autobase'
import b4a from 'b4a'

const argv = process.argv.slice(2)
const clientSock = argv[0]
const storageDir = argv[1]
function flag (name) { const i = argv.indexOf(name); return i === -1 ? null : argv[i + 1] }
const bootstrapHex = flag('--bootstrap')
const peerListen = flag('--peer-listen')
const peerConnect = flag('--peer-connect')

if (!clientSock || !storageDir) {
  console.error('usage: bare daemon-ab.js <client-sock> <storage-dir> [--bootstrap hex] [--peer-listen p | --peer-connect p]')
  process.exit(1)
}
try { fs.unlinkSync(clientSock) } catch (_) {}

const store = new Corestore(storageDir)
const base = new Autobase(store, bootstrapHex ? b4a.from(bootstrapHex, 'hex') : null, {
  open: (s) => s.get('atomlog', { valueEncoding: 'utf-8' }),
  apply: async (nodes, view, host) => {
    for (const node of nodes) {
      const op = node.value
      if (op.t === 'add-writer') {
        await host.addWriter(b4a.from(op.key, 'hex'), { indexer: !!op.indexer })
        continue
      }
      await view.append(op.s)
    }
  },
  valueEncoding: 'json',
  optimistic: true,
  ackInterval: 1000,
  ackThreshold: 4
})
await base.ready()

// ── peer replication (cross-process, over a UDS pipe) ─────────────────────────
function wireReplication (sock, initiator) {
  const s = base.replicate(initiator)
  s.on('error', () => {})
  sock.on('error', () => {})
  s.pipe(sock).pipe(s)
}
if (peerListen) {
  try { fs.unlinkSync(peerListen) } catch (_) {}
  const rserver = net.createServer({ allowHalfOpen: false }, (sock) => wireReplication(sock, false))
  rserver.listen(peerListen)
}
if (peerConnect) {
  const connect = () => {
    const sock = net.createConnection(peerConnect)
    sock.on('error', () => setTimeout(connect, 100)) // peer may not be listening yet
    sock.on('connect', () => wireReplication(sock, true))
  }
  connect()
}

// ── replay helper ─────────────────────────────────────────────────────────────
async function replay (upto) {
  await base.update()
  const v = base.view
  const n = Math.max(0, Math.min(upto, v.length))
  let body = ''
  for (let i = 0; i < n; i++) body += await v.get(i)
  return '(' + body + ')\n'
}

async function handleLine (line, sock) {
  if (line.length === 0) return
  const sp = line.indexOf(' ')
  const verb = sp === -1 ? line : line.slice(0, sp)
  const rest = sp === -1 ? '' : line.slice(sp + 1)

  switch (verb) {
    case 'store':
      if (!base.writable) return // silently drop; harness checks `writable` first
      await base.append({ t: 'atom', s: rest })
      return
    case 'load':
      sock.write(await replay(Infinity))
      return
    case 'loadat':
      sock.write(await replay(parseInt(rest, 10) || 0))
      return
    case 'seqof': {
      await base.update()
      const v = base.view
      let found = -1
      for (let i = 0; i < v.length; i++) { if (await v.get(i) === rest) { found = i; break } }
      sock.write(String(found) + '\n')
      return
    }
    case 'count':
      await base.update()
      sock.write(String(base.view.length) + '\n')
      return
    case 'barrier':
      await base.update()
      sock.write('ok\n')
      return
    case 'addwriter':
      await base.append({ t: 'add-writer', key: rest, indexer: false })
      sock.write('ok\n')
      return
    case 'writable':
      sock.write((base.writable ? 'true' : 'false') + '\n')
      return
    default:
      sock.write('err unknown-verb ' + verb + '\n')
  }
}

const server = net.createServer({ allowHalfOpen: false }, (sock) => {
  let buf = ''
  let tail = Promise.resolve()
  sock.on('data', (chunk) => {
    buf += chunk.toString()
    let nl
    while ((nl = buf.indexOf('\n')) !== -1) {
      const line = buf.slice(0, nl)
      buf = buf.slice(nl + 1)
      tail = tail.then(() => handleLine(line, sock)).catch((e) => console.error(e))
    }
  })
  sock.on('end', () => sock.destroy())
  sock.on('error', () => sock.destroy())
})

server.listen(clientSock, () => {
  console.log('AB key=' + b4a.toString(base.key, 'hex') +
    ' local=' + b4a.toString(base.local.key, 'hex') +
    ' writable=' + base.writable + ' sock=' + clientSock)
})

async function shutdown () {
  try { server.close() } catch (_) {}
  try { await base.close() } catch (_) {}
  try { fs.unlinkSync(clientSock) } catch (_) {}
  try { if (peerListen) fs.unlinkSync(peerListen) } catch (_) {}
  process.exit(0)
}
process.on('SIGINT', shutdown)
process.on('SIGTERM', shutdown)