bare-operating-system/packages/bare-os-booter/test.bare-os-chat-roundtrip.js

/**
 * Integration: TCP + Protomux chat channel — verifies Protomux wiring + ingest when
 * socket.publicKey / remotePublicKey match the Hyperswarm Noise identity pattern.
 */
import test from 'brittle'
import net from 'net'
import Protomux from 'protomux'
import b4a from 'b4a'
import crypto from 'hypercore-crypto'
import { SwarmDisk } from './lib/p2p/swarm-disk.js'
import { createBareOsChatService } from './lib/services/bare-os-chat-service.js'

/** @returns {Promise<{ srvSock: net.Socket, cliSock: net.Socket }>} */
function tcpDuplexPair() {
  return new Promise((resolve, reject) => {
    const srv = net.createServer()
    srv.once('error', reject)
    srv.listen(0, '127.0.0.1', () => {
      const addr = srv.address()
      if (!addr || typeof addr === 'string') {
        reject(new Error('tcpDuplexPair: bad address'))
        return
      }
      const cliSock = net.connect(addr.port, '127.0.0.1')
      srv.once('connection', (srvSock) => {
        srv.close(() => {})
        resolve({ srvSock, cliSock })
      })
      cliSock.once('error', reject)
    })
  })
}

function wireNoiseLikeKeys(sockLocal, kpLocal, kpRemote) {
  sockLocal.publicKey = kpLocal.publicKey
  sockLocal.remotePublicKey = kpRemote.publicKey
}

/** Protomux 3.11 sets `peer.chatChan` only after `fullyOpened()`. */
async function waitForChatChannel(peer, ms = 1500) {
  const deadline = Date.now() + ms
  while (!peer.chatChan && Date.now() < deadline) {
    await new Promise((r) => setTimeout(r, 20))
  }
  const ch = peer.chatChan
  if (ch && typeof ch.fullyOpened === 'function') {
    await Promise.race([
      ch.fullyOpened(),
      new Promise((r) => setTimeout(r, 500))
    ])
  }
}

test('TCP + Protomux — chat event roundtrips Alice → Bob', async (t) => {
  process.env.BARE_OS_PROTOMUX_CHAT_CHANNEL = '1'

  const kpAlice = crypto.keyPair()
  const kpBob = crypto.keyPair()

  const { srvSock, cliSock } = await tcpDuplexPair()

  wireNoiseLikeKeys(cliSock, kpAlice, kpBob)
  wireNoiseLikeKeys(srvSock, kpBob, kpAlice)

  const diskAlice = new SwarmDisk()
  diskAlice.localNoiseWirePk = b4a.from(kpAlice.publicKey)

  const diskBob = new SwarmDisk()

  diskAlice.bareOsChatService = createBareOsChatService({
    env: /** @type {Record<string, string>} */ ({ USER: 'aliceuser' })
  })
  diskBob.bareOsChatService = createBareOsChatService({
    env: /** @type {Record<string, string>} */ ({ USER: 'bobuser' })
  })

  const muxAlice = new Protomux(cliSock)
  const muxBob = new Protomux(srvSock)

  const peerAliceToBob = {
    chan: /** @type {unknown} */ (null),
    mux: muxAlice,
    socket: cliSock,
    id: null,
    chatChan: null
  }

  const peerBobFromAlice = {
    chan: /** @type {unknown} */ (null),
    mux: muxBob,
    socket: srvSock,
    id: null,
    chatChan: null
  }

  diskAlice.bareOsChatService.pairOnMux(
    diskAlice,
    muxAlice,
    cliSock,
    peerAliceToBob
  )
  diskBob.bareOsChatService.pairOnMux(
    diskBob,
    muxBob,
    srvSock,
    peerBobFromAlice
  )

  diskAlice.peers.add(peerAliceToBob)
  diskBob.peers.add(peerBobFromAlice)

  await waitForChatChannel(peerAliceToBob)
  await waitForChatChannel(peerBobFromAlice)

  /** @type {unknown[]} */
  const bobRx = []
  diskBob.bareOsChatService.subscribe((ev) => bobRx.push(ev))

  await diskAlice.bareOsChatService.broadcastLocal(
    diskAlice,
    'hello-from-tcp',
    {
      displayName: 'aliceuser'
    }
  )

  const deadline = Date.now() + 1500
  while (bobRx.length < 1 && Date.now() < deadline) {
    await new Promise((r) => setTimeout(r, 50))
  }

  t.ok(bobRx.length >= 1, 'Bob received remote event')
  const last = /** @type {{ body?: string, local?: boolean }} */ (
    bobRx[bobRx.length - 1]
  )
  t.is(last.body, 'hello-from-tcp')
  t.absent(last.local)

  srvSock.destroy()
  cliSock.destroy()
})

test('pairBareOsChatExistingPeers wires chat after peers already on disk', async (t) => {
  process.env.BARE_OS_PROTOMUX_CHAT_CHANNEL = '1'

  const kpAlice = crypto.keyPair()
  const kpBob = crypto.keyPair()

  const { srvSock, cliSock } = await tcpDuplexPair()

  wireNoiseLikeKeys(cliSock, kpAlice, kpBob)
  wireNoiseLikeKeys(srvSock, kpBob, kpAlice)

  const diskAlice = new SwarmDisk()
  diskAlice.localNoiseWirePk = b4a.from(kpAlice.publicKey)

  const diskBob = new SwarmDisk()

  diskAlice.bareOsChatService = createBareOsChatService({
    env: /** @type {Record<string, string>} */ ({ USER: 'aliceuser' })
  })
  diskBob.bareOsChatService = createBareOsChatService({
    env: /** @type {Record<string, string>} */ ({ USER: 'bobuser' })
  })

  const muxAlice = new Protomux(cliSock)
  const muxBob = new Protomux(srvSock)

  const peerAliceToBob = {
    chan: /** @type {unknown} */ (null),
    mux: muxAlice,
    socket: cliSock,
    id: null,
    chatChan: null
  }

  const peerBobFromAlice = {
    chan: /** @type {unknown} */ (null),
    mux: muxBob,
    socket: srvSock,
    id: null,
    chatChan: null
  }

  diskAlice.peers.add(peerAliceToBob)
  diskBob.peers.add(peerBobFromAlice)

  await diskAlice.pairBareOsChatExistingPeers()
  await diskBob.pairBareOsChatExistingPeers()

  await waitForChatChannel(peerAliceToBob)
  await waitForChatChannel(peerBobFromAlice)

  /** @type {unknown[]} */
  const bobRx = []
  diskBob.bareOsChatService.subscribe((ev) => bobRx.push(ev))

  await diskAlice.bareOsChatService.broadcastLocal(
    diskAlice,
    'hello-after-late-pair',
    {
      displayName: 'aliceuser'
    }
  )

  const deadline = Date.now() + 1500
  while (bobRx.length < 1 && Date.now() < deadline) {
    await new Promise((r) => setTimeout(r, 50))
  }

  t.ok(bobRx.length >= 1, 'Bob received remote event')
  const last = /** @type {{ body?: string, local?: boolean }} */ (
    bobRx[bobRx.length - 1]
  )
  t.is(last.body, 'hello-after-late-pair')
  t.absent(last.local)

  srvSock.destroy()
  cliSock.destroy()
})