import test from 'brittle'
import Protomux from 'protomux'
import SecretStream from '@hyperswarm/secret-stream'
import { GitSyncServer, GitSyncClient } from '../../lib/p2p/sync/git-sync-channel.js'
/**
* Create a pair of connected Protomux instances over SecretStream.
* Returns { muxA, muxB, streamA, streamB }.
*/
function createMuxPair () {
const a = new SecretStream(true)
const b = new SecretStream(false)
a.rawStream.pipe(b.rawStream).pipe(a.rawStream)
const muxA = Protomux.from(a)
const muxB = Protomux.from(b)
return { muxA, muxB, streamA: a, streamB: b }
}
function destroyPair ({ streamA, streamB }) {
streamA.destroy()
streamB.destroy()
}
// ---------------------------------------------------------------------------
// Control channel: ref-update
// ---------------------------------------------------------------------------
test('control channel exchanges ref-update', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const received = t.test('ref-update received')
received.plan(4)
const server = new GitSyncServer(muxA)
const client = new GitSyncClient(muxB, {
onRefUpdate (repo, refs, serverKey) {
received.is(repo, 'my-app')
received.alike(refs, { 'refs/heads/main': 'abc123' })
received.is(serverKey, 'deadbeef')
received.pass('ref-update received')
}
})
await server.init()
await client.init()
server.sendRefUpdate('my-app', { 'refs/heads/main': 'abc123' }, 'deadbeef')
await received
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Control channel: sync-request
// ---------------------------------------------------------------------------
test('control channel exchanges sync-request', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const received = t.test('sync-request received')
received.plan(3)
const server = new GitSyncServer(muxA, {
onSyncRequest (repo, localRefs) {
received.is(repo, 'my-app')
received.alike(localRefs, { 'refs/heads/main': '000aaa' })
received.pass('sync-request received')
}
})
const client = new GitSyncClient(muxB)
await server.init()
await client.init()
client.sendSyncRequest('my-app', { 'refs/heads/main': '000aaa' })
await received
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Full sync flow: sync-request -> sync-ready -> sync-complete
// ---------------------------------------------------------------------------
test('full sync flow: request -> ready -> complete', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const steps = []
let resolveAll
const allDone = new Promise((resolve) => { resolveAll = resolve })
const server = new GitSyncServer(muxA, {
onSyncRequest (repo, localRefs) {
steps.push('server:sync-request')
// Respond with sync-ready
server.sendSyncReady(repo, { 'refs/heads/main': 'bbb222' })
// After a tick, send sync-complete
setTimeout(() => {
server.sendSyncComplete(repo, 0, ['refs/heads/main'])
}, 50)
}
})
const client = new GitSyncClient(muxB, {
onSyncReady (repo, remoteRefs) {
steps.push('client:sync-ready')
t.is(repo, 'my-app')
t.alike(remoteRefs, { 'refs/heads/main': 'bbb222' })
},
onSyncComplete (repo, exitCode, refsUpdated) {
steps.push('client:sync-complete')
t.is(repo, 'my-app')
t.is(exitCode, 0)
t.alike(refsUpdated, ['refs/heads/main'])
resolveAll()
}
})
await server.init()
await client.init()
client.sendSyncRequest('my-app', { 'refs/heads/main': 'aaa111' })
await allDone
t.alike(steps, ['server:sync-request', 'client:sync-ready', 'client:sync-complete'])
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Error handling: sync-error
// ---------------------------------------------------------------------------
test('error handling: sync-error', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const received = t.test('sync-error received')
received.plan(4)
const server = new GitSyncServer(muxA)
const client = new GitSyncClient(muxB, {
onSyncError (repo, code, message) {
received.is(repo, 'my-app')
received.is(code, 'FETCH_FAILED')
received.is(message, 'git fetch exit 128')
received.pass('sync-error received')
}
})
await server.init()
await client.init()
server.sendSyncError('my-app', 'FETCH_FAILED', 'git fetch exit 128')
await received
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Data channel: open/close
// ---------------------------------------------------------------------------
test('data channel opens and closes', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const server = new GitSyncServer(muxA)
const client = new GitSyncClient(muxB)
await server.init()
await client.init()
// Open data channels on both sides (they pair by protocol+id)
const serverData = server.openDataChannel('my-app')
const clientData = client.openDataChannel('my-app')
t.ok(serverData, 'server data channel created')
t.ok(clientData, 'client data channel created')
t.ok(serverData.channel, 'server has channel')
t.ok(clientData.channel, 'client has channel')
// Wait for channels to open
await serverData.channel.fullyOpened()
await clientData.channel.fullyOpened()
// Exchange binary data
let serverRecv = null
let clientRecv = null
let resolveServer, resolveClient
const serverGot = new Promise((r) => { resolveServer = r })
const clientGot = new Promise((r) => { resolveClient = r })
serverData.message.onmessage = (data) => {
serverRecv = data
resolveServer()
}
clientData.message.onmessage = (data) => {
clientRecv = data
resolveClient()
}
clientData.message.send(Buffer.from('hello from client'))
serverData.message.send(Buffer.from('hello from server'))
await serverGot
await clientGot
t.alike(serverRecv, Buffer.from('hello from client'))
t.alike(clientRecv, Buffer.from('hello from server'))
// Close data channels
server.closeDataChannel('my-app')
client.closeDataChannel('my-app')
t.absent(server._dataChannels.get('my-app'), 'server data channel removed')
t.absent(client._dataChannels.get('my-app'), 'client data channel removed')
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Full sync: full-sync-request -> full-sync-manifest (server-to-server)
// ---------------------------------------------------------------------------
test('full-sync-request and full-sync-manifest exchange (server-to-server)', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
let resolveManifest
const gotManifest = new Promise((r) => { resolveManifest = r })
// Server A sends full-sync-request, receives full-sync-manifest
const serverA = new GitSyncServer(muxA, {
onFullSyncManifest (repos) {
resolveManifest(repos)
}
})
// Server B receives full-sync-request, sends full-sync-manifest
const serverB = new GitSyncServer(muxB, {
onFullSyncRequest (repos) {
t.alike(repos, {
'repo-a': { 'refs/heads/main': 'aaa111' },
'repo-b': { 'refs/heads/main': 'bbb222' }
})
serverB.sendFullSyncManifest({
'repo-a': { 'refs/heads/main': 'aaa111' },
'repo-b': { 'refs/heads/main': 'bbb222' },
'repo-c': { 'refs/heads/main': 'ccc333' }
})
}
})
await serverA.init()
await serverB.init()
// Server A initiates full sync
serverA.sendFullSyncRequest({
'repo-a': { 'refs/heads/main': 'aaa111' },
'repo-b': { 'refs/heads/main': 'bbb222' }
})
const manifest = await gotManifest
t.is(Object.keys(manifest).length, 3, 'manifest has 3 repos')
t.is(manifest['repo-a']['refs/heads/main'], 'aaa111')
t.is(manifest['repo-b']['refs/heads/main'], 'bbb222')
t.is(manifest['repo-c']['refs/heads/main'], 'ccc333')
await serverA.destroy()
await serverB.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Full sync: client sends full-sync-request, receives full-sync-manifest
// ---------------------------------------------------------------------------
test('client full-sync-request and full-sync-manifest', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
let resolveManifest
const gotManifest = new Promise((r) => { resolveManifest = r })
const server = new GitSyncServer(muxA, {
onFullSyncRequest (repos) {
t.alike(repos, { 'my-app': { 'refs/heads/main': 'old111' } })
server.sendFullSyncManifest({
'my-app': { 'refs/heads/main': 'new222' }
})
}
})
const client = new GitSyncClient(muxB, {
onFullSyncManifest (repos) {
resolveManifest(repos)
}
})
await server.init()
await client.init()
client.sendFullSyncRequest({ 'my-app': { 'refs/heads/main': 'old111' } })
const manifest = await gotManifest
t.is(manifest['my-app']['refs/heads/main'], 'new222')
await server.destroy()
await client.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Destroy cleans up
// ---------------------------------------------------------------------------
test('destroy cleans up without errors', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const server = new GitSyncServer(muxA)
const client = new GitSyncClient(muxB)
await server.init()
await client.init()
// Open a data channel before destroying
server.openDataChannel('cleanup-test')
client.openDataChannel('cleanup-test')
await server.destroy()
await client.destroy()
t.is(server._dataChannels.size, 0, 'server data channels cleaned up')
t.is(client._dataChannels.size, 0, 'client data channels cleaned up')
t.pass('both destroyed without errors')
destroyPair({ streamA, streamB })
})