/**
* GitSyncManager — Orchestrates git sync over Protomux channels
*
* Attaches a GitSyncServer to every driveSwarm connection alongside
* corestore.replicate(). Handles incoming sync requests by reading
* local refs, diffing against the peer's refs, and piping
* git-upload-pack through a Protomux data channel.
*
* Also supports broadcasting ref-update notifications to all
* connected federation peers after a deploy.
*
* Usage:
* const mgr = new GitSyncManager({ reposDir, registry, maxConcurrent: 5 })
*
* // In driveSwarm connection handler:
* swarm.on('connection', (conn) => {
* const stream = store.replicate(conn)
* mgr.attachToConnection(conn, stream)
* })
*
* // After deploy:
* mgr.broadcastRefUpdate('my-app', refs, serverKey)
*/
import path from 'bare-path'
import fs from 'bare-fs'
import Protomux from 'protomux'
import { GitSyncServer } from './git-sync-channel.js'
import { readRefs } from '../../git/ref-reader.js'
import { diffRefs } from '../../git/ref-diff.js'
import { createGitPipe } from './git-sync-pipe.js'
export class GitSyncManager {
constructor (opts = {}) {
this.reposDir = opts.reposDir
this.registry = opts.registry
this.serverKey = opts.serverKey || null // Server identity key (hex) for punch:// URLs
this.clientManager = opts.clientManager || null
this.linkedRepos = opts.linkedRepos || null // Map<repoName, LinkedRepo> for issue Autobase keys
this.metadataKey = opts.metadataKey || null // MetadataBase bootstrap key (hex) — sent in manifests for federation
this._maxConcurrent = opts.maxConcurrent || 5
this._peers = new Map() // remotePublicKey hex -> { server, conn }
this._destroyed = false
}
/**
* Attach git sync to an existing driveSwarm connection.
*
* Obtains the Protomux from the replication stream (at
* stream.noiseStream.userData, as corestore sets it up),
* creates a GitSyncServer on that mux, and registers
* handlers for sync-request and ref-update messages.
*
* @param {object} conn — Hyperswarm connection (has remotePublicKey)
* @param {object} replicateStream — return value of corestore.replicate(conn)
*/
attachToConnection (conn, replicateStream) {
if (this._destroyed) return
// Get or create Protomux — Protomux.from() is idempotent.
const noiseStream = replicateStream.noiseStream || replicateStream
const mux = Protomux.from(noiseStream)
if (!mux) {
console.error('[GitSync] Failed to get Protomux from stream, skipping')
return
}
const peerKey = conn.remotePublicKey.toString('hex')
const server = new GitSyncServer(mux, {
onChannelOpen: () => {
// Channel is open on both sides — initiate full sync (no timers)
if (this._destroyed) return
if (!this.clientManager) return
this._initiateFullSync(server, peerKey).catch(err => {
console.error(`[GitSync] Failed to initiate full sync with ${peerKey.slice(0, 8)}: ${err.message}`)
})
},
onSyncRequest: (repo, localRefs) => {
this._handleSyncRequest(repo, localRefs, server).catch(err => {
console.error(`[GitSync] Error handling sync-request for ${repo}: ${err.message}`)
try {
server.sendSyncError(repo, 'INTERNAL_ERROR', err.message)
} catch {}
})
},
onRefUpdate: (repo, refs, remoteServerKey) => {
// Received ref-update from a peer — forward to client manager for sync
// Use the serverKey from the message (server identity key for punch:// URLs),
// NOT the driveSwarm connection peerKey (different key, wrong DHT topic)
const syncKey = remoteServerKey || peerKey
if (this.serverKey && syncKey === this.serverKey) return // self-sync guard
console.log(`[GitSync] Received ref-update from ${syncKey.slice(0, 8)} for ${repo}`)
if (this.clientManager) {
// Pass the Protomux server so sync can use the existing connection
// instead of spawning a git fetch subprocess
this.clientManager.handleRefUpdate(repo, refs, syncKey, server).catch(err => {
console.error(`[GitSync] Client sync error for ${repo}: ${err.message}`)
})
}
},
onFullSyncRequest: (remoteRepos, remoteServerKey, remoteMetadataKey) => {
this._handleFullSyncRequest(remoteRepos, server, peerKey, remoteServerKey).catch(err => {
console.error(`[GitSync] Error handling full-sync-request from ${peerKey.slice(0, 8)}: ${err.message}`)
})
},
onFullSyncManifest: (remoteRepos, remoteServerKey, remoteMetadataKey) => {
// Received manifest response from peer — use remoteServerKey for punch:// URLs
if (this.clientManager) {
const syncKey = remoteServerKey || peerKey
this.clientManager.handleFullSyncManifest(remoteRepos, syncKey).catch(err => {
console.error(`[GitSync] Error handling full-sync-manifest from ${syncKey.slice(0, 8)}: ${err.message}`)
})
}
}
})
server.init().catch(err => {
console.error(`[GitSync] Failed to init server for peer ${peerKey.slice(0, 8)}: ${err.message}`)
})
this._peers.set(peerKey, { server, conn })
// Clean up on connection close
conn.on('close', () => {
const entry = this._peers.get(peerKey)
if (entry) {
entry.server.destroy().catch(() => {})
this._peers.delete(peerKey)
}
if (this.clientManager) {
this.clientManager._initialSyncDone.delete(peerKey)
}
})
// Full sync triggered by channel onopen (no timers — same pattern as Hyperdrive)
// When both sides open the pear-git-sync channel, Protomux fires onopen
}
/**
* Handle an incoming sync-request from a peer.
*
* 1. Read local refs for the requested repo
* 2. Diff against the peer's declared refs
* 3. If synced, send sync-complete immediately
* 4. If not synced, send sync-ready with our refs, then spawn
* git-upload-pack and pipe it through the data channel
*
* @param {string} repo — repository name (without .git suffix)
* @param {object} peerRefs — peer's local refs { refName: sha }
* @param {GitSyncServer} server — the server to send responses on
*/
async _handleSyncRequest (repo, peerRefs, server) {
const repoPath = path.join(this.reposDir, repo + '.git')
// Read our local refs
let localRefs
try {
localRefs = await this._readRefs(repoPath)
} catch (err) {
if (err.code === 'ENOENT') {
server.sendSyncError(repo, 'REPO_NOT_FOUND', 'Repository not found: ' + repo)
return
}
throw err
}
// Diff: what does the peer need from us?
const diff = diffRefs(peerRefs, localRefs)
if (diff.synced) {
// Already in sync
server.sendSyncComplete(repo, 0, [])
return
}
// Refs differ — send sync-ready with our full ref set
server.sendSyncReady(repo, localRefs)
// Spawn git-upload-pack and pipe through data channel
await this._spawnGitUploadPack(repo, repoPath, server)
}
/**
* Spawn git-upload-pack and pipe through a Protomux data channel.
* Separated for testability.
*
* @param {string} repo — repository name
* @param {string} repoPath — path to the bare git repo
* @param {GitSyncServer} server — the server instance
*/
async _spawnGitUploadPack (repo, repoPath, server) {
const dataEntry = server.openDataChannel(repo)
if (!dataEntry) {
server.sendSyncError(repo, 'CHANNEL_ERROR', 'Failed to open data channel')
return
}
const pipe = createGitPipe('git-upload-pack', repoPath, dataEntry, {
timeout: 120000
})
const result = await pipe.exitPromise
// Close data channel after transfer
server.closeDataChannel(repo)
// Report completion
const refsUpdated = result.exitCode === 0
? Object.keys(await this._readRefs(repoPath)).filter(k => k !== 'HEAD')
: []
server.sendSyncComplete(repo, result.exitCode, refsUpdated)
}
/**
* Read refs from a bare git repository.
* Wrapped in a method for testability (tests can override).
*
* @param {string} repoPath — path to the bare git repo
* @returns {Promise<Object>} ref name -> SHA hash
*/
async _readRefs (repoPath) {
return readRefs(repoPath)
}
/**
* Initiate a full sync with a peer by sending our manifest.
* Only fires once per peer connection (tracked by clientManager._initialSyncDone).
*
* @param {GitSyncServer} server — the server to send through
* @param {string} peerKey — hex public key of the peer
*/
async _initiateFullSync (server, peerKey) {
if (!this.clientManager) return
if (this.serverKey && peerKey === this.serverKey) return // self-sync guard
if (this.clientManager._initialSyncDone.has(peerKey)) return
this.clientManager._initialSyncDone.set(peerKey, true)
const localManifest = await this._buildManifest()
console.log(`[GitSync] Sending full-sync-request to ${peerKey.slice(0, 8)} with ${Object.keys(localManifest).length} repos`)
server.sendFullSyncRequest(localManifest, this.serverKey, this.metadataKey)
}
/**
* Handle a full-sync-request from a peer.
*
* Builds a manifest of all local repos and their refs, then
* sends it back as a full-sync-manifest response.
*
* @param {object} remoteRepos — peer's repo manifest { repoName: { ref: sha, ... }, ... }
* @param {GitSyncServer} server — the server to respond on
* @param {string} peerKey — peer's public key hex
*/
async _handleFullSyncRequest (remoteRepos, server, peerKey, remoteServerKey) {
console.log(`[GitSync] Received full-sync-request from ${(remoteServerKey || peerKey).slice(0, 8)} with ${Object.keys(remoteRepos).length} repos`)
const localManifest = await this._buildManifest()
server.sendFullSyncManifest(localManifest, this.serverKey, this.metadataKey)
console.log(`[GitSync] Sent full-sync-manifest to ${(remoteServerKey || peerKey).slice(0, 8)} with ${Object.keys(localManifest).length} repos`)
// Forward remote manifest to client manager with the correct serverKey for punch:// URLs
if (this.clientManager && remoteServerKey) {
this.clientManager.handleFullSyncManifest(remoteRepos, remoteServerKey).catch(err => {
console.error(`[GitSync] Error handling remote manifest from ${remoteServerKey.slice(0, 8)}: ${err.message}`)
})
}
}
/**
* Build a manifest of all local repos and their refs.
* Scans reposDir for .git directories and reads refs from each.
*
* @returns {Promise<Object>} { repoName: { refName: sha, ... }, ... }
*/
async _buildManifest () {
const manifest = {}
try {
const entries = await fs.promises.readdir(this.reposDir)
for (const entry of entries) {
if (!entry.endsWith('.git')) continue
const repoPath = path.join(this.reposDir, entry)
// Skip repos with .pear-git-ignore marker
const ignored = await fs.promises.stat(path.join(repoPath, '.pear-git-ignore')).catch(() => null)
if (ignored) continue
const repoName = entry.replace(/\.git$/, '')
try {
const refs = await this._readRefs(repoPath)
const hashRefs = {}
for (const [ref, val] of Object.entries(refs)) {
if (ref === 'HEAD') continue
if (/^[0-9a-f]{40}$/.test(val)) hashRefs[ref] = val
}
// Include issue Autobase bootstrap key for federation
const linked = this.linkedRepos ? this.linkedRepos.get(repoName) : null
manifest[repoName] = {
refs: hashRefs,
issueKey: linked ? linked.bootstrapKeyHex : null
}
} catch {}
}
} catch {}
return manifest
}
/**
* Broadcast a ref-update notification to all connected federation peers.
* Called after a deploy completes to notify peers of new refs.
*
* @param {string} repoName — repository name
* @param {object} refs — { refName: sha, ... }
* @param {string} serverKey — this server's public key
*/
broadcastRefUpdate (repoName, refs, serverKey) {
for (const [peerKey, entry] of this._peers) {
try {
entry.server.sendRefUpdate(repoName, refs, serverKey)
} catch (err) {
console.error(`[GitSync] Failed to send ref-update to ${peerKey.slice(0, 8)}: ${err.message}`)
}
}
}
/**
* Destroy the manager: clean up all peer connections.
*/
async destroy () {
this._destroyed = true
for (const [peerKey, entry] of this._peers) {
try {
await entry.server.destroy()
} catch {}
}
this._peers.clear()
}
}