/**
* Git Sync Channel — Protomux control + data channels for git sync
*
* Multiplexes git sync messages over an existing Protomux instance
* (typically obtained from driveSwarm's corestore.replicate() connection).
*
* Control channel (protocol: 'pear-git-sync'):
* Message 0 — JSON control messages (ref-update, sync-request, sync-ready, sync-complete, sync-error)
*
* Data channels (protocol: 'pear-git-sync-data', id: repoName):
* Message 0 — binary git protocol data per-repo
*
* Usage:
* const mux = Protomux.from(conn.noiseStream)
* const server = new GitSyncServer(mux, { onSyncRequest, onRefUpdate })
* await server.init()
* server.sendRefUpdate('my-repo', { 'refs/heads/main': 'abc123' }, serverKey)
*/
import c from 'compact-encoding'
import b4a from 'b4a'
const CONTROL_PROTOCOL = 'pear-git-sync'
const DATA_PROTOCOL = 'pear-git-sync-data'
// -- JSON encoding for Protomux messages (compact-encoding compatible) ------
const jsonEncoding = {
preencode (state, val) {
const str = JSON.stringify(val)
c.string.preencode(state, str)
},
encode (state, val) {
const str = JSON.stringify(val)
c.string.encode(state, str)
},
decode (state) {
const str = c.string.decode(state)
return JSON.parse(str)
}
}
// -- Base class for shared logic -------------------------------------------
class GitSyncBase {
constructor (mux, callbacks = {}) {
this._mux = mux
this._controlChannel = null
this._controlMessage = null
this._dataChannels = new Map() // repoName -> { channel, message }
this._callbacks = callbacks
}
/**
* Create and open the control channel.
*
* Uses mux.pair() to register a handler BEFORE creating the channel.
* This matches the Hypercore replicator pattern (replicator.js:2986-3008)
* and prevents the timing race where the remote opens the channel before
* the local side has called createChannel(). Without pair(), Protomux
* rejects incoming opens for unknown protocols (_requestSession line 690).
*/
async _initControl () {
// Register handler for incoming control channel opens from remote.
// If remote opens first, this callback fires and we create our channel.
// If we open first, createChannel + open below handles it directly.
this._mux.pair({ protocol: CONTROL_PROTOCOL }, () => {
this._ensureControlChannel()
})
// Create and open our side of the channel
this._ensureControlChannel()
}
_ensureControlChannel () {
if (this._controlChannel) return
this._controlChannel = this._mux.createChannel({
protocol: CONTROL_PROTOCOL,
onopen: () => {
console.log('[GitSync] Control channel OPENED')
if (this._callbacks.onChannelOpen) this._callbacks.onChannelOpen()
},
onclose: () => {
console.log('[GitSync] Control channel CLOSED')
this._controlChannel = null
this._controlMessage = null
if (this._callbacks.onChannelClose) this._callbacks.onChannelClose()
}
})
if (!this._controlChannel) return
this._controlMessage = this._controlChannel.addMessage({
encoding: jsonEncoding,
onmessage: (msg) => this._onControlMessage(msg)
})
this._controlChannel.open()
}
/**
* Open a per-repo binary data channel.
* Both sides must call openDataChannel(repoName) for the channels to pair.
*/
openDataChannel (repoName) {
if (this._dataChannels.has(repoName)) {
return this._dataChannels.get(repoName)
}
const id = b4a.from(repoName, 'utf-8')
const channel = this._mux.createChannel({
protocol: DATA_PROTOCOL,
id,
onopen () {},
onclose () {}
})
if (!channel) return null
const message = channel.addMessage({
encoding: c.raw,
onmessage () {} // caller overrides message.onmessage
})
channel.open()
const entry = { channel, message }
this._dataChannels.set(repoName, entry)
return entry
}
/**
* Close a per-repo data channel.
*/
closeDataChannel (repoName) {
const entry = this._dataChannels.get(repoName)
if (!entry) return
entry.channel.close()
this._dataChannels.delete(repoName)
}
/**
* Destroy everything: close all data channels, close control channel.
*/
async destroy () {
for (const [name] of this._dataChannels) {
this.closeDataChannel(name)
}
if (this._controlChannel) {
this._controlChannel.close()
this._controlChannel = null
}
}
_onControlMessage (msg) {
// override in subclass
}
}
// -- Server -----------------------------------------------------------------
export class GitSyncServer extends GitSyncBase {
constructor (mux, callbacks = {}) {
super(mux, callbacks)
}
async init () {
await this._initControl()
}
// -- Send methods (server -> client) --
sendRefUpdate (repo, refs, serverKey) {
this._controlMessage.send({
type: 'ref-update',
repo,
refs,
serverKey
})
}
sendSyncReady (repo, remoteRefs) {
this._controlMessage.send({
type: 'sync-ready',
repo,
remoteRefs
})
}
sendSyncComplete (repo, exitCode, refsUpdated) {
this._controlMessage.send({
type: 'sync-complete',
repo,
exitCode,
refsUpdated
})
}
// Server can also request sync from the remote (bidirectional).
// Used when the client manager wants to pull data through the
// existing Protomux connection instead of spawning git fetch.
sendSyncRequest (repo, localRefs) {
this._controlMessage.send({
type: 'sync-request',
repo,
localRefs
})
}
sendSyncError (repo, code, message) {
this._controlMessage.send({
type: 'sync-error',
repo,
code,
message
})
}
sendFullSyncManifest (repos, serverKey, metadataKey) {
this._controlMessage.send({
type: 'full-sync-manifest',
repos,
serverKey,
metadataKey: metadataKey || null
})
}
sendFullSyncRequest (repos, serverKey, metadataKey) {
this._controlMessage.send({
type: 'full-sync-request',
repos,
serverKey,
metadataKey: metadataKey || null
})
}
// -- Receive dispatch --
_onControlMessage (msg) {
switch (msg.type) {
case 'sync-request':
if (this._callbacks.onSyncRequest) {
this._callbacks.onSyncRequest(msg.repo, msg.localRefs)
}
break
case 'ref-update':
// Server can also receive ref-updates from peers
if (this._callbacks.onRefUpdate) {
this._callbacks.onRefUpdate(msg.repo, msg.refs, msg.serverKey)
}
break
case 'full-sync-request':
if (this._callbacks.onFullSyncRequest) {
this._callbacks.onFullSyncRequest(msg.repos, msg.serverKey, msg.metadataKey)
}
break
case 'full-sync-manifest':
if (this._callbacks.onFullSyncManifest) {
this._callbacks.onFullSyncManifest(msg.repos, msg.serverKey, msg.metadataKey)
}
break
}
}
}
// -- Client -----------------------------------------------------------------
export class GitSyncClient extends GitSyncBase {
constructor (mux, callbacks = {}) {
super(mux, callbacks)
}
async init () {
await this._initControl()
}
// -- Send methods (client -> server) --
sendSyncRequest (repo, localRefs) {
this._controlMessage.send({
type: 'sync-request',
repo,
localRefs
})
}
sendFullSyncRequest (repos, metadataKey) {
this._controlMessage.send({
type: 'full-sync-request',
repos,
metadataKey: metadataKey || null
})
}
// -- Receive dispatch --
_onControlMessage (msg) {
switch (msg.type) {
case 'ref-update':
if (this._callbacks.onRefUpdate) {
this._callbacks.onRefUpdate(msg.repo, msg.refs, msg.serverKey)
}
break
case 'sync-ready':
if (this._callbacks.onSyncReady) {
this._callbacks.onSyncReady(msg.repo, msg.remoteRefs)
}
break
case 'sync-complete':
if (this._callbacks.onSyncComplete) {
this._callbacks.onSyncComplete(msg.repo, msg.exitCode, msg.refsUpdated)
}
break
case 'sync-error':
if (this._callbacks.onSyncError) {
this._callbacks.onSyncError(msg.repo, msg.code, msg.message)
}
break
case 'full-sync-manifest':
if (this._callbacks.onFullSyncManifest) {
this._callbacks.onFullSyncManifest(msg.repos, msg.serverKey, msg.metadataKey)
}
break
}
}
}