pgh-sidecar / test / overlay-e2e.test.js

/**
 * E2E test: overlay publisher + consumer via in-process Duplex pair.
 *
 * Two sidecar-shaped overlay stacks (publisher + consumer) are paired
 * over a streamx Duplex (mirroring upstream hyper-overlay's e2e style —
 * faster + more reliable than spinning up a real testnet swarm).
 *
 * Publisher registers a working-tree file with its OverlayManager.
 * The hyper-overlay/v2 protocol auto-syncs: publisher offers, consumer
 * requests, chunks stream, file lands at consumer's repoRoot. Consumer
 * then verifies the bytes match the original via crypto.data hash.
 *
 * This is the in-process equivalent of the tabertron ↔ stortron e2e
 * test in `docs/overlay-tabertron-stortron-e2e.md` — same code paths
 * (MultiRepoOverlayProtocolV2 + SyncEngine + TransferManager + the
 * unified OverlayManager), just no real network.
 */

import test from 'brittle'
import Corestore from 'corestore'
import Protomux from 'protomux'
import { Duplex } from 'streamx'
import crypto from 'hypercore-crypto'
import fs from '#node:fs'
import path from '#node:path'
import os from '#node:os'

import { OverlayManager } from '../lib/overlay-manager.js'
import { SyncEngine } from 'hyper-overlay/sync-engine'
import { TransferManager } from 'hyper-overlay/transfer'
import { MultiRepoOverlayProtocolV2 } from '../lib/multi-repo-overlay-protocol.js'

function tmpDir (prefix) {
  return fs.mkdtempSync(path.join(os.tmpdir(), prefix + '-'))
}

function makeDuplex () {
  let aWrite, bWrite
  const a = new Duplex({ write (d, cb) { bWrite(d); cb() }, read () {} })
  const b = new Duplex({ write (d, cb) { aWrite(d); cb() }, read () {} })
  aWrite = (d) => a.push(d)
  bWrite = (d) => b.push(d)
  return [a, b]
}

async function makeNode (label, t) {
  const storeRoot = tmpDir('overlay-' + label + '-store')
  const store = new Corestore(storeRoot)
  await store.ready()
  const filePaths = new Map()
  const om = new OverlayManager(store, { filePaths })
  const fileIndex = await om.getFileIndex()
  const syncEngine = new SyncEngine(fileIndex, store.namespace('overlay-sync-v2'))
  await syncEngine.ready()
  const transferManager = new TransferManager(fileIndex)
  const protocol = new MultiRepoOverlayProtocolV2(
    syncEngine, transferManager, om, { filePaths }
  )
  om.attachSyncEngine(syncEngine)
  await om.warmRepoCache()
  await om.populateFilePaths()
  t.teardown(async () => {
    protocol.destroy()
    await syncEngine.close()
    await om.close()
    await store.close()
    fs.rmSync(storeRoot, { recursive: true, force: true })
  })
  return { storeRoot, store, om, syncEngine, transferManager, protocol }
}

test('publisher serves a file by oid; consumer auto-syncs and verifies', async (t) => {
  const pub = await makeNode('pub', t)
  const cons = await makeNode('cons', t)

  // Each side registers the same logical repo with its own working tree.
  const pubRepo = tmpDir('overlay-pub-repo')
  const consRepo = tmpDir('overlay-cons-repo')
  t.teardown(() => fs.rmSync(pubRepo, { recursive: true, force: true }))
  t.teardown(() => fs.rmSync(consRepo, { recursive: true, force: true }))

  await pub.om.initRepo('demo', pubRepo)
  await cons.om.initRepo('demo', consRepo)
  await pub.om.warmRepoCache()
  await cons.om.warmRepoCache()

  // Publisher: write a file + register it.
  const data = Buffer.from('hello v2 protocol '.repeat(500))
  const oid = crypto.data(data).toString('hex')
  fs.writeFileSync(path.join(pubRepo, 'asset.bin'), data)
  await pub.om.putFile('demo', 'asset.bin', {
    oid, size: data.length, mtime: Date.now()
  })
  t.is(pub.syncEngine.feed.length, 1, 'publisher feed has 1 entry after putFile')
  t.is(
    pub.om.filePaths.get('demo/asset.bin'),
    path.join(pubRepo, 'asset.bin'),
    'publisher filePaths populated'
  )

  // Pair via in-memory Duplex.
  const [streamA, streamB] = makeDuplex()
  pub.protocol.attach(Protomux.from(streamA))
  cons.protocol.attach(Protomux.from(streamB))

  // Wait for transfer.
  const expected = path.join(consRepo, 'asset.bin')
  const start = Date.now()
  while (Date.now() - start < 15000) {
    if (fs.existsSync(expected)) {
      const got = fs.readFileSync(expected)
      if (got.equals(data)) break
    }
    await new Promise(r => setTimeout(r, 100))
  }

  t.ok(fs.existsSync(expected), 'consumer received file at expected path')
  const got = fs.readFileSync(expected)
  t.is(got.length, data.length, 'consumer file size matches publisher')
  t.is(
    crypto.data(got).toString('hex'),
    oid,
    'consumer file oid matches publisher (full byte match)'
  )
  t.is(cons.syncEngine.feed.length, 1, 'consumer feed got logPut on receive')
})