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

// hyper-atomspace daemon (Bare) — M1: Hypercore-backed op-log.
//
// The op-log is a real Hypercore (append-only, durable on disk). store appends
// one s-expr per atom; load replays the whole log; loadat replays a prefix (a
// read-only checkout at seq K = "cheap fork": read fewer blocks, get an earlier
// graph, no reprocessing). This is the M1 backend swap from M0's in-memory array.
//
// usage: bare daemon.js <socket-path> <storage-dir>
//
// Wire protocol — newline-framed request lines over a UNIX-domain socket:
//   store <sexpr>\n   append <sexpr> to the log. NO reply (fire-and-forget).
//   load\n            reply: (<sexpr0>…<sexpr_{len-1}>)\n   — whole log.
//   loadat <K>\n      reply: (<sexpr0>…<sexpr_{K-1}>)\n     — prefix/checkout @K.
//   seqof <sexpr>\n    reply: <seq>\n  — log position this op entered at, or -1.
//                     (log-level provenance; the C++ side composes causal lineage.)
//   clear\n           truncate the log to 0. reply: ok\n
//   barrier\n         sync point. reply: ok\n
//   count\n           reply: <len>\n
//
// All commands run on a single ordered promise-chain, so by the time a load or
// barrier executes, every prior store's append has resolved.

import net from 'bare-net'
import fs from 'bare-fs'
import process from 'bare-process'
import Hypercore from 'hypercore'

const sockPath = process.argv[2]
const storageDir = process.argv[3]
if (!sockPath || !storageDir) {
  console.error('usage: bare daemon.js <socket-path> <storage-dir>')
  process.exit(1)
}

// Fresh socket bind: a stale socket file makes listen() fail.
try { fs.unlinkSync(sockPath) } catch (_) { /* not there — fine */ }

const core = new Hypercore(storageDir, { valueEncoding: 'utf-8' })
await core.ready()

// Read blocks [0, upto) and wrap them in one outer paren — the exact shape the
// C++ decode loop expects.
async function replay (upto) {
  const n = Math.max(0, Math.min(upto, core.length))
  let body = ''
  for (let i = 0; i < n; i++) body += await core.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':
      await core.append(rest)
      return // no reply
    case 'load':
      sock.write(await replay(core.length))
      return
    case 'loadat':
      sock.write(await replay(parseInt(rest, 10) || 0))
      return
    case 'seqof': {
      // Exact-match scan for the op that entered the log as `rest`. The log is
      // append-only so the first (only) match is the birth position. A seq index
      // is a later optimisation; M3's log is small.
      let found = -1
      for (let i = 0; i < core.length; i++) {
        if (await core.get(i) === rest) { found = i; break }
      }
      sock.write(String(found) + '\n')
      return
    }
    case 'clear':
      await core.truncate(0)
      sock.write('ok\n')
      return
    case 'barrier':
      sock.write('ok\n')
      return
    case 'count':
      sock.write(String(core.length) + '\n')
      return
    default:
      sock.write('err unknown-verb ' + verb + '\n')
  }
}

const server = net.createServer({ allowHalfOpen: false }, (sock) => {
  let buf = ''
  // Single ordered chain: guarantees appends apply in arrival order and that a
  // later load/barrier observes all prior stores.
  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())   // release fd (hyper-svc leak fix)
  sock.on('error', () => sock.destroy())
})

server.listen(sockPath, () => {
  console.log('hyper-atomspace listening ' + sockPath + ' store=' + storageDir +
    ' len=' + core.length)
})

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