import test from 'brittle'
import fs from 'bare-fs'
import path from 'bare-path'
import os from 'bare-os'
import Protomux from 'protomux'
import SecretStream from '@hyperswarm/secret-stream'
import { GitSyncManager } from '../../lib/p2p/sync/git-sync-manager.js'
import { GitSyncClient, GitSyncServer } from '../../lib/p2p/sync/git-sync-channel.js'
import { GitSyncClientManager } from '../../lib/p2p/sync/git-sync-client-manager.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()
}
/**
* Create a mock replication stream that mimics corestore.replicate() return.
* The Protomux is at stream.noiseStream.userData (just like corestore does it).
*/
function createMockReplicateStream (mux, stream) {
return {
noiseStream: {
userData: mux,
opened: Promise.resolve()
},
_stream: stream
}
}
/**
* Create a mock registry with federation entries.
*/
function createMockRegistry (entries = []) {
return {
createReadStream (opts) {
return (async function * () {
for (const entry of entries) {
yield entry
}
})()
}
}
}
// ---------------------------------------------------------------------------
// Constructor
// ---------------------------------------------------------------------------
test('constructor: sets defaults', (t) => {
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: {}
})
t.is(mgr.reposDir, '/tmp/repos')
t.is(mgr._maxConcurrent, 5)
t.is(mgr._peers.size, 0)
})
test('constructor: accepts custom maxConcurrent', (t) => {
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: {},
maxConcurrent: 3
})
t.is(mgr._maxConcurrent, 3)
})
// ---------------------------------------------------------------------------
// attachToConnection creates GitSyncServer and tracks peer
// ---------------------------------------------------------------------------
test('attachToConnection creates server and tracks peer', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
const replicateStream = createMockReplicateStream(muxA, streamA)
// conn needs remotePublicKey for peer tracking
const conn = {
remotePublicKey: Buffer.from('a'.repeat(64), 'hex'),
on () {}
}
mgr.attachToConnection(conn, replicateStream)
t.is(mgr._peers.size, 1, 'peer tracked')
const peerKey = conn.remotePublicKey.toString('hex')
t.ok(mgr._peers.has(peerKey), 'peer stored by remote key')
const peerEntry = mgr._peers.get(peerKey)
t.ok(peerEntry.server, 'server created')
await mgr.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// attachToConnection handles connection close
// ---------------------------------------------------------------------------
test('attachToConnection cleans up on connection close', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
const replicateStream = createMockReplicateStream(muxA, streamA)
let closeHandler = null
const conn = {
remotePublicKey: Buffer.from('b'.repeat(64), 'hex'),
on (event, handler) {
if (event === 'close') closeHandler = handler
}
}
mgr.attachToConnection(conn, replicateStream)
t.is(mgr._peers.size, 1, 'peer tracked')
t.ok(closeHandler, 'close handler registered')
// Simulate connection close
closeHandler()
t.is(mgr._peers.size, 0, 'peer removed on close')
await mgr.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// broadcastRefUpdate sends to all connected peers
// ---------------------------------------------------------------------------
test('broadcastRefUpdate sends to all peers', async (t) => {
t.timeout(10000)
const pair1 = createMuxPair()
const pair2 = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
// Track messages received by clients
const received = []
let resolveAll
const allReceived = new Promise((r) => { resolveAll = r })
// Attach two connections
const conn1 = {
remotePublicKey: Buffer.from('c'.repeat(64), 'hex'),
on () {}
}
const conn2 = {
remotePublicKey: Buffer.from('d'.repeat(64), 'hex'),
on () {}
}
mgr.attachToConnection(conn1, createMockReplicateStream(pair1.muxA, pair1.streamA))
mgr.attachToConnection(conn2, createMockReplicateStream(pair2.muxA, pair2.streamA))
t.is(mgr._peers.size, 2, 'two peers tracked')
// Create clients on the other side of each pair
const client1 = new GitSyncClient(pair1.muxB, {
onRefUpdate (repo, refs, serverKey) {
received.push({ peer: 1, repo, refs, serverKey })
if (received.length === 2) resolveAll()
}
})
const client2 = new GitSyncClient(pair2.muxB, {
onRefUpdate (repo, refs, serverKey) {
received.push({ peer: 2, repo, refs, serverKey })
if (received.length === 2) resolveAll()
}
})
await client1.init()
await client2.init()
// Give the servers time to initialize
await new Promise(r => setTimeout(r, 100))
// Broadcast ref update
mgr.broadcastRefUpdate('my-app', { 'refs/heads/main': 'abc123' }, 'server-key-1')
await allReceived
t.is(received.length, 2, 'both clients received ref-update')
for (const msg of received) {
t.is(msg.repo, 'my-app')
t.alike(msg.refs, { 'refs/heads/main': 'abc123' })
t.is(msg.serverKey, 'server-key-1')
}
await client1.destroy()
await client2.destroy()
await mgr.destroy()
destroyPair(pair1)
destroyPair(pair2)
})
// ---------------------------------------------------------------------------
// broadcastRefUpdate with no peers is a no-op
// ---------------------------------------------------------------------------
test('broadcastRefUpdate with no peers does nothing', async (t) => {
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
// Should not throw
mgr.broadcastRefUpdate('my-app', { 'refs/heads/main': 'abc123' }, 'key')
t.pass('no error with zero peers')
await mgr.destroy()
})
// ---------------------------------------------------------------------------
// destroy cleans up all state
// ---------------------------------------------------------------------------
test('destroy cleans up all peers', async (t) => {
t.timeout(10000)
const pair1 = createMuxPair()
const pair2 = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
const conn1 = {
remotePublicKey: Buffer.from('e'.repeat(64), 'hex'),
on () {}
}
const conn2 = {
remotePublicKey: Buffer.from('f0'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn1, createMockReplicateStream(pair1.muxA, pair1.streamA))
mgr.attachToConnection(conn2, createMockReplicateStream(pair2.muxA, pair2.streamA))
t.is(mgr._peers.size, 2)
await mgr.destroy()
t.is(mgr._peers.size, 0, 'all peers cleared')
t.is(mgr._destroyed, true, 'marked as destroyed')
destroyPair(pair1)
destroyPair(pair2)
})
// ---------------------------------------------------------------------------
// handleSyncRequest: responds sync-complete when already synced
// ---------------------------------------------------------------------------
test('handleSyncRequest sends sync-complete when refs match', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
// We need a real repo for readRefs, but for unit tests we mock the internal
// by subclassing. Instead, let's test the full flow with a mock _readRefs.
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
// Override _readRefs to return known refs
mgr._readRefs = async (repoPath) => {
return {
HEAD: 'ref: refs/heads/main',
'refs/heads/main': 'abc123'
}
}
const conn = {
remotePublicKey: Buffer.from('aa'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn, createMockReplicateStream(muxA, streamA))
// Set up client on other side
let resolveComplete
const gotComplete = new Promise((r) => { resolveComplete = r })
const client = new GitSyncClient(muxB, {
onSyncComplete (repo, exitCode, refsUpdated) {
resolveComplete({ repo, exitCode, refsUpdated })
}
})
await client.init()
// Wait for server to be ready
await new Promise(r => setTimeout(r, 100))
// Client requests sync with matching refs
client.sendSyncRequest('my-app', { 'refs/heads/main': 'abc123' })
const result = await gotComplete
t.is(result.repo, 'my-app')
t.is(result.exitCode, 0)
t.alike(result.refsUpdated, [])
await client.destroy()
await mgr.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// handleSyncRequest: sends sync-error for unknown repo
// ---------------------------------------------------------------------------
test('handleSyncRequest sends sync-error for unknown repo', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
// Override _readRefs to throw ENOENT (repo doesn't exist)
mgr._readRefs = async (repoPath) => {
const err = new Error('ENOENT: no such file')
err.code = 'ENOENT'
throw err
}
const conn = {
remotePublicKey: Buffer.from('bb'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn, createMockReplicateStream(muxA, streamA))
let resolveError
const gotError = new Promise((r) => { resolveError = r })
const client = new GitSyncClient(muxB, {
onSyncError (repo, code, message) {
resolveError({ repo, code, message })
}
})
await client.init()
await new Promise(r => setTimeout(r, 100))
client.sendSyncRequest('nonexistent-repo', {})
const result = await gotError
t.is(result.repo, 'nonexistent-repo')
t.is(result.code, 'REPO_NOT_FOUND')
await client.destroy()
await mgr.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// handleSyncRequest: sends sync-ready when refs differ
// ---------------------------------------------------------------------------
test('handleSyncRequest sends sync-ready when refs differ', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
// Local has newer refs than peer
mgr._readRefs = async (repoPath) => {
return {
HEAD: 'ref: refs/heads/main',
'refs/heads/main': 'new-sha-222'
}
}
// Mock _spawnGitUploadPack to avoid needing real git
mgr._spawnGitUploadPack = async (repoName, repoPath, server) => {
// Simulate git-upload-pack completing
server.sendSyncComplete(repoName, 0, ['refs/heads/main'])
}
const conn = {
remotePublicKey: Buffer.from('cc'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn, createMockReplicateStream(muxA, streamA))
const steps = []
let resolveComplete
const gotComplete = new Promise((r) => { resolveComplete = r })
const client = new GitSyncClient(muxB, {
onSyncReady (repo, remoteRefs) {
steps.push('sync-ready')
t.is(repo, 'my-app')
t.alike(remoteRefs, {
HEAD: 'ref: refs/heads/main',
'refs/heads/main': 'new-sha-222'
})
},
onSyncComplete (repo, exitCode, refsUpdated) {
steps.push('sync-complete')
resolveComplete({ repo, exitCode, refsUpdated })
}
})
await client.init()
await new Promise(r => setTimeout(r, 100))
// Client has old refs
client.sendSyncRequest('my-app', { 'refs/heads/main': 'old-sha-111' })
const result = await gotComplete
t.is(result.repo, 'my-app')
t.is(result.exitCode, 0)
t.alike(steps, ['sync-ready', 'sync-complete'])
await client.destroy()
await mgr.destroy()
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Multiple connections: each gets its own server
// ---------------------------------------------------------------------------
test('multiple connections get independent servers', async (t) => {
t.timeout(10000)
const pair1 = createMuxPair()
const pair2 = createMuxPair()
const mgr = new GitSyncManager({
reposDir: '/tmp/repos',
registry: createMockRegistry()
})
const conn1 = {
remotePublicKey: Buffer.from('11'.repeat(32), 'hex'),
on () {}
}
const conn2 = {
remotePublicKey: Buffer.from('22'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn1, createMockReplicateStream(pair1.muxA, pair1.streamA))
mgr.attachToConnection(conn2, createMockReplicateStream(pair2.muxA, pair2.streamA))
const peer1 = mgr._peers.get(conn1.remotePublicKey.toString('hex'))
const peer2 = mgr._peers.get(conn2.remotePublicKey.toString('hex'))
t.ok(peer1.server !== peer2.server, 'different server instances')
t.is(mgr._peers.size, 2)
await mgr.destroy()
destroyPair(pair1)
destroyPair(pair2)
})
// ---------------------------------------------------------------------------
// _buildManifest reads all repos from reposDir
// ---------------------------------------------------------------------------
test('_buildManifest reads all local repos', async (t) => {
t.timeout(10000)
const tmp = path.join(os.tmpdir(), 'test-mgr-manifest-' + Date.now())
await fs.promises.mkdir(tmp, { recursive: true })
// Create two fake bare repos
const repoA = path.join(tmp, 'repo-a.git')
const repoB = path.join(tmp, 'repo-b.git')
await fs.promises.mkdir(path.join(repoA, 'refs', 'heads'), { recursive: true })
await fs.promises.mkdir(path.join(repoB, 'refs', 'heads'), { recursive: true })
await fs.promises.writeFile(path.join(repoA, 'HEAD'), 'ref: refs/heads/main\n')
await fs.promises.writeFile(path.join(repoB, 'HEAD'), 'ref: refs/heads/main\n')
await fs.promises.writeFile(
path.join(repoA, 'refs', 'heads', 'main'),
'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n'
)
await fs.promises.writeFile(
path.join(repoB, 'refs', 'heads', 'main'),
'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n'
)
const mgr = new GitSyncManager({
reposDir: tmp,
registry: createMockRegistry()
})
const manifest = await mgr._buildManifest()
t.is(Object.keys(manifest).length, 2, 'has 2 repos')
t.is(manifest['repo-a']['refs/heads/main'], 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')
t.is(manifest['repo-b']['refs/heads/main'], 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb')
await fs.promises.rm(tmp, { recursive: true })
await mgr.destroy()
})
// ---------------------------------------------------------------------------
// handleFullSyncRequest responds with local manifest
// ---------------------------------------------------------------------------
test('handleFullSyncRequest responds with full-sync-manifest', async (t) => {
t.timeout(10000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
const tmp = path.join(os.tmpdir(), 'test-mgr-fullsync-' + Date.now())
const repo = path.join(tmp, 'my-app.git')
await fs.promises.mkdir(path.join(repo, 'refs', 'heads'), { recursive: true })
await fs.promises.writeFile(path.join(repo, 'HEAD'), 'ref: refs/heads/main\n')
await fs.promises.writeFile(
path.join(repo, 'refs', 'heads', 'main'),
'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n'
)
const mgr = new GitSyncManager({
reposDir: tmp,
registry: createMockRegistry()
})
const conn = {
remotePublicKey: Buffer.from('aa'.repeat(32), 'hex'),
on () {}
}
mgr.attachToConnection(conn, createMockReplicateStream(muxA, streamA))
// Set up client on other side to send full-sync-request and receive manifest
let resolveManifest
const gotManifest = new Promise((r) => { resolveManifest = r })
const client = new GitSyncClient(muxB, {
onFullSyncManifest (repos) {
resolveManifest(repos)
}
})
await client.init()
await new Promise(r => setTimeout(r, 100))
// Client sends full-sync-request
client.sendFullSyncRequest({})
const manifest = await gotManifest
t.is(Object.keys(manifest).length, 1, 'manifest has 1 repo')
t.is(manifest['my-app']['refs/heads/main'], 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')
await client.destroy()
await mgr.destroy()
await fs.promises.rm(tmp, { recursive: true })
destroyPair({ streamA, streamB })
})
// ---------------------------------------------------------------------------
// Full sync: server sends full-sync-request, peer responds with manifest
// ---------------------------------------------------------------------------
test('full sync exchange between two GitSyncManagers', async (t) => {
t.timeout(15000)
const { muxA, muxB, streamA, streamB } = createMuxPair()
// Create two temp repo dirs
const tmpA = path.join(os.tmpdir(), 'test-fullsync-a-' + Date.now())
const tmpB = path.join(os.tmpdir(), 'test-fullsync-b-' + Date.now())
const repoA = path.join(tmpA, 'shared-repo.git')
const repoB = path.join(tmpB, 'shared-repo.git')
await fs.promises.mkdir(path.join(repoA, 'refs', 'heads'), { recursive: true })
await fs.promises.mkdir(path.join(repoB, 'refs', 'heads'), { recursive: true })
await fs.promises.writeFile(path.join(repoA, 'HEAD'), 'ref: refs/heads/main\n')
await fs.promises.writeFile(path.join(repoB, 'HEAD'), 'ref: refs/heads/main\n')
await fs.promises.writeFile(
path.join(repoA, 'refs', 'heads', 'main'),
'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n'
)
// B has different ref — should trigger sync
await fs.promises.writeFile(
path.join(repoB, 'refs', 'heads', 'main'),
'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n'
)
// Create client managers
const clientMgrA = new GitSyncClientManager({ reposDir: tmpA })
const clientMgrB = new GitSyncClientManager({ reposDir: tmpB })
// Mock _syncRepo to simulate successful sync
let syncCalledA = false
clientMgrA._syncRepo = async (repoName, repoPath, peerKey) => {
syncCalledA = true
// Update ref to match remote
await fs.promises.writeFile(
path.join(repoPath, 'refs', 'heads', 'main'),
'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\n'
)
}
let syncCalledB = false
clientMgrB._syncRepo = async (repoName, repoPath, peerKey) => {
syncCalledB = true
await fs.promises.writeFile(
path.join(repoPath, 'refs', 'heads', 'main'),
'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n'
)
}
// Create managers (no setTimeout, we'll trigger manually)
const mgrA = new GitSyncManager({
reposDir: tmpA,
registry: createMockRegistry(),
clientManager: clientMgrA
})
const mgrB = new GitSyncManager({
reposDir: tmpB,
registry: createMockRegistry(),
clientManager: clientMgrB
})
const connA = {
remotePublicKey: Buffer.from('bb'.repeat(32), 'hex'),
on () {}
}
const connB = {
remotePublicKey: Buffer.from('aa'.repeat(32), 'hex'),
on () {}
}
// Manually attach (skip setTimeout by not using attachToConnection)
// Instead, directly init servers and trigger sync
mgrA.attachToConnection(connA, createMockReplicateStream(muxA, streamA))
mgrB.attachToConnection(connB, createMockReplicateStream(muxB, streamB))
// Wait for servers to init
await new Promise(r => setTimeout(r, 200))
// Manually trigger full sync from A to B (simulating the setTimeout)
const peerKeyA = connA.remotePublicKey.toString('hex')
const serverA = mgrA._peers.get(peerKeyA).server
await mgrA._initiateFullSync(serverA, peerKeyA)
// Wait for the manifest exchange to complete
await new Promise(r => setTimeout(r, 500))
// A should have synced because B's manifest shows different refs
t.ok(syncCalledA, 'manager A synced from B')
await mgrA.destroy()
await mgrB.destroy()
await clientMgrA.destroy()
await clientMgrB.destroy()
await fs.promises.rm(tmpA, { recursive: true })
await fs.promises.rm(tmpB, { recursive: true })
destroyPair({ streamA, streamB })
})