hyper-note / test / unit / replication.test.js

import test from 'brittle'
import fs from 'bare-fs'
import path from 'bare-path'
import os from 'bare-os'
import Corestore from 'corestore'
import Hyperbee from 'hyperbee'
import { DriveManager } from '../../lib/drive-manager.js'
import { ReplicationManager } from '../../lib/replication.js'

function tmpSetup () {
  const dir = path.join(os.tmpdir(), 'hn-rep-' + Date.now() + '-' + Math.random().toString(36).slice(2))
  fs.mkdirSync(dir, { recursive: true })
  const store = new Corestore(dir)
  const registry = new Hyperbee(store.get({ name: 'registry' }), {
    keyEncoding: 'utf-8',
    valueEncoding: 'json'
  })
  const dm = new DriveManager(store, registry)
  return { store, registry, dm, dir }
}

function rmrf (dir) {
  try {
    for (const f of fs.readdirSync(dir)) {
      const fp = path.join(dir, f)
      const stat = fs.statSync(fp)
      if (stat.isDirectory()) rmrf(fp)
      else fs.unlinkSync(fp)
    }
    fs.rmdirSync(dir)
  } catch {}
}

test('start creates swarm', async (t) => {
  const { store, registry, dm, dir } = tmpSetup()
  await registry.ready()

  const rm = new ReplicationManager(store, dm)
  await rm.start()

  t.ok(rm.swarm, 'swarm is created')
  const status = rm.getStatus()
  t.is(status.swarmActive, true)
  t.is(status.peersConnected, 0)

  await rm.stop()
  await store.close()
  rmrf(dir)
})

test('stop destroys swarm', async (t) => {
  const { store, registry, dm, dir } = tmpSetup()
  await registry.ready()

  const rm = new ReplicationManager(store, dm)
  await rm.start()
  await rm.stop()

  t.is(rm.swarm, null)
  const status = rm.getStatus()
  t.is(status.swarmActive, false)

  await store.close()
  rmrf(dir)
})

test('announce / unannounce tracks topics', async (t) => {
  const { store, registry, dm, dir } = tmpSetup()
  await registry.ready()

  const rm = new ReplicationManager(store, dm)
  await rm.start()

  await dm.createDrive('work')

  await rm.announce('work')
  let status = rm.getStatus()
  t.ok(status.announced.includes('work'))

  // setShared should be true
  const info = await dm.getDriveInfo('work')
  t.is(info.isShared, true)

  await rm.unannounce('work')
  status = rm.getStatus()
  t.not(status.announced.includes('work'))

  // setShared should be false
  const info2 = await dm.getDriveInfo('work')
  t.is(info2.isShared, false)

  await rm.stop()
  await store.close()
  rmrf(dir)
})

test('getStatus returns correct info', async (t) => {
  const { store, registry, dm, dir } = tmpSetup()
  await registry.ready()

  const rm = new ReplicationManager(store, dm)

  // Before start
  let status = rm.getStatus()
  t.is(status.swarmActive, false)
  t.is(status.peersConnected, 0)
  t.is(status.announced.length, 0)

  await rm.start()
  await dm.createDrive('alpha')
  await dm.createDrive('beta')
  await rm.announce('alpha')
  await rm.announce('beta')

  status = rm.getStatus()
  t.is(status.swarmActive, true)
  t.is(status.announced.length, 2)
  t.ok(status.announced.includes('alpha'))
  t.ok(status.announced.includes('beta'))

  await rm.stop()
  await store.close()
  rmrf(dir)
})

test('announceAll re-announces shared drives', async (t) => {
  const { store, registry, dm, dir } = tmpSetup()
  await registry.ready()

  const rm = new ReplicationManager(store, dm)
  await rm.start()

  // Create drives and mark some as shared
  await dm.createDrive('shared1')
  await dm.setShared('shared1', true)
  await dm.createDrive('shared2')
  await dm.setShared('shared2', true)
  await dm.createDrive('private')

  await rm.announceAll()

  const status = rm.getStatus()
  t.is(status.announced.length, 2)
  t.ok(status.announced.includes('shared1'))
  t.ok(status.announced.includes('shared2'))
  t.not(status.announced.includes('private'))

  await rm.stop()
  await store.close()
  rmrf(dir)
})

test('two stores replicate via pipe', async (t) => {
  // Store A: write data
  const dirA = path.join(os.tmpdir(), 'hn-repA-' + Date.now() + '-' + Math.random().toString(36).slice(2))
  fs.mkdirSync(dirA, { recursive: true })
  const storeA = new Corestore(dirA)
  const registryA = new Hyperbee(storeA.get({ name: 'registry' }), {
    keyEncoding: 'utf-8',
    valueEncoding: 'json'
  })
  await registryA.ready()
  const dmA = new DriveManager(storeA, registryA)

  const driveA = await dmA.createDrive('test')
  await driveA.put('/metadata.json', Buffer.from(JSON.stringify({ slug: 'test', name: 'Test' })))
  await driveA.put('/notes/abc.json', Buffer.from(JSON.stringify({ id: 'abc', title: 'Hello', content: 'World' })))

  // Store B: replicate from A
  const dirB = path.join(os.tmpdir(), 'hn-repB-' + Date.now() + '-' + Math.random().toString(36).slice(2))
  fs.mkdirSync(dirB, { recursive: true })
  const storeB = new Corestore(dirB)
  const registryB = new Hyperbee(storeB.get({ name: 'registry' }), {
    keyEncoding: 'utf-8',
    valueEncoding: 'json'
  })
  await registryB.ready()
  const dmB = new DriveManager(storeB, registryB)

  // Add driveA's key as remote in B
  const driveAKey = driveA.key.toString('hex')
  const driveB = await dmB.addRemoteDrive(driveAKey, 'remote-test')

  // Pipe replication
  const s1 = storeA.replicate(true)
  const s2 = storeB.replicate(false)
  s1.pipe(s2).pipe(s1)

  // Wait for replication to propagate
  await new Promise(r => setTimeout(r, 500))

  // Verify B can read A's data
  const metaBuf = await driveB.get('/metadata.json')
  t.ok(metaBuf, 'metadata replicated')
  const meta = JSON.parse(metaBuf.toString())
  t.is(meta.slug, 'test')

  const noteBuf = await driveB.get('/notes/abc.json')
  t.ok(noteBuf, 'note replicated')
  const note = JSON.parse(noteBuf.toString())
  t.is(note.title, 'Hello')
  t.is(note.content, 'World')

  // Clean up
  s1.destroy()
  s2.destroy()
  await storeA.close()
  await storeB.close()
  rmrf(dirA)
  rmrf(dirB)
})