const ReadyResource = require('ready-resource');
const safetyCatch = require('safety-catch');
const Hyperswarm = require('hyperswarm');
const crypto = require('hypercore-crypto');
const fs = require('fs');
const path = require('path');
const { spawn } = require('child_process');
const process = require('process');
const { scanRepos, repoInfo, readRefs } = require('./lib/repos');
const { deriveTopic } = require('./lib/topics');
const { browserMethods } = require('./lib/browser');
module.exports = class HyperSync extends ReadyResource {
constructor (opts = {}) {
super();
this._reposDir = opts.repos || path.join(require('os').homedir(), 'git-repos');
// OFFLINE mode: `swarm: false` means NO NETWORK — do not inject one, do
// not create one. Distinct from `swarm: null/undefined`, which still means
// "make your own". Without this distinction a caller had no way to say
// "scan and serve locally, but never touch the DHT": `_open()` below
// creates a Hyperswarm whenever none was injected, so passing null (as
// pear-git-v3 did when its own swarm was disabled) silently produced a
// live DHT-connected swarm that pulled from real peers.
//
// Offline keeps everything local intact — scan(), the repos map, info(),
// _repoPath, handleInfoRefs — so HTTP git serving still works. Only
// announce/join/connect is suppressed.
this._offline = opts.swarm === false;
this._swarm = this._offline ? null : (opts.swarm || null);
this._ownSwarm = !this._offline && !this._swarm;
this._keypair = opts.keypair || null;
// Optional push authorization hook. git-upload-pack (read) is always
// allowed; any write service (git-receive-pack) is rejected UNLESS
// this hook approves it. Signature:
// async ({ service, repo, remotePublicKey }) => boolean
// When absent, behavior is read-only (back-compat — the prior code
// hard-rejected everything that wasn't git-upload-pack).
this._authorize = typeof opts.authorize === 'function' ? opts.authorize : null;
this.repos = new Map();
this.remotes = new Map();
this._topics = new Map();
this._serving = false;
this.ready().catch(safetyCatch);
}
async _open () {
// Generate or load keypair for server identity
if (!this._keypair) {
this._keypair = crypto.keyPair();
}
// Scan repos
this.repos = scanRepos(this._reposDir);
this.emit('scan', this.repos);
// Create swarm if not injected — unless explicitly offline.
if (!this._swarm && !this._offline) {
this._swarm = new Hyperswarm({ keyPair: this._keypair });
}
if (this._swarm) {
this._swarm.on('connection', (conn, info) => this._onConnection(conn, info));
}
}
async _close () {
// Unannounce all topics
for (const [name, entry] of this._topics) {
try {
await entry.discovery.destroy();
} catch {}
}
this._topics.clear();
if (this._ownSwarm && this._swarm) {
await this._swarm.destroy();
this._swarm = null;
}
}
// ── Properties ────────────────────────────────────────────
get serverKey () {
return this._keypair ? this._keypair.publicKey.toString('hex') : null;
}
get serving () {
return this._topics.size;
}
// ── Repo Management ───────────────────────────────────────
async scan () {
// Pass the previous map so unchanged repos skip the tree walk.
this.repos = scanRepos(this._reposDir, { previous: this.repos });
this.emit('scan', this.repos);
return this.repos;
}
status () {
const repoList = [];
for (const [name, info] of this.repos) {
repoList.push({
name,
branches: info.branchCount,
packSize: info.packSize,
head: info.head ? info.head.slice(0, 12) : null,
serving: this._topics.has(name),
lastUpdated: info.lastUpdated
});
}
return {
repos: repoList,
repoCount: this.repos.size,
serving: this._topics.size,
remotes: this.remotes.size,
serverKey: this.serverKey
};
}
info (name) {
const repo = this.repos.get(name);
if (!repo) return null;
// Re-read refs for fresh data
const refs = readRefs(repo.path);
return {
...repo,
refs,
serving: this._topics.has(name),
topic: this._topics.has(name)
? this._topics.get(name).topic.toString('hex')
: null
};
}
// ── Serving ───────────────────────────────────────────────
async serve (name) {
if (name) return this._serveOne(name);
// Serve all repos
const names = [...this.repos.keys()];
for (const n of names) {
await this._serveOne(n);
}
// Single flush for all topics
if (this._swarm) await this._swarm.flush();
this._serving = true;
this.emit('serving', this._topics.size);
}
async _serveOne (name) {
if (this._topics.has(name)) return;
const repo = this.repos.get(name);
if (!repo) throw new Error('Repo not found: ' + name);
const topic = deriveTopic(this._keypair.publicKey, name);
// Offline: derive + record the topic (callers read it, and `serve` events
// drive downstream indexes) but never ANNOUNCE it. No discovery handle.
const discovery = this._offline
? null
: this._swarm.join(topic, { server: true, client: false });
this._topics.set(name, { topic, discovery, name });
this.emit('serve', name, topic.toString('hex'));
}
async unserve (name) {
// `discovery` is null in offline mode (nothing was announced, so there is
// nothing to unannounce) — guard rather than relying on a thrown TypeError.
if (name) {
const entry = this._topics.get(name);
if (entry) {
if (entry.discovery) await entry.discovery.destroy();
this._topics.delete(name);
}
return;
}
// Unserve all
for (const [n, entry] of this._topics) {
if (entry.discovery) await entry.discovery.destroy();
}
this._topics.clear();
this._serving = false;
}
// ── Remotes ───────────────────────────────────────────────
addRemote (name, serverKey) {
this.remotes.set(name, {
name,
serverKey: typeof serverKey === 'string' ? serverKey : serverKey.toString('hex')
});
}
removeRemote (name) {
this.remotes.delete(name);
}
// ── Sync (Pull) ───────────────────────────────────────────
async pull (remoteName, repoName) {
const remote = this.remotes.get(remoteName);
if (!remote) throw new Error('Remote not found: ' + remoteName);
const repos = repoName
? [repoName]
: [...this.repos.keys()];
const result = { synced: [], failed: [], skipped: [], elapsed: 0 };
const start = Date.now();
for (const name of repos) {
const repoPath = path.join(this._reposDir, name + '.git');
const packDir = path.join(repoPath, 'objects', 'pack');
const fetchUrl = 'punch://' + remote.serverKey + '/' + name;
// Ensure repo exists locally
if (!fs.existsSync(repoPath)) {
try {
await this._exec('git', ['init', '--bare', repoPath]);
} catch (err) {
result.failed.push({ name, error: err.message });
continue;
}
}
// Pre-fetch cleanup: an interrupted transfer leaves
// `objects/pack/tmp_pack_XXX` orphaned; these accumulate
// indefinitely (seen 37 on hyper-overlay) and slow every
// subsequent operation. Clean before each attempt so state is
// deterministic.
this._cleanTmpPacks(packDir);
this.emit('pull', remoteName, name, 'start');
// Retry on transient failures. Bounded attempts + exponential
// backoff + per-attempt tmp_pack cleanup so a failure's garbage
// can't contaminate the next try.
const maxAttempts = 3;
let lastErr = null;
let synced = false;
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
try {
await this._gitFetch(repoPath, fetchUrl);
synced = true;
break;
} catch (err) {
lastErr = err;
if (attempt < maxAttempts) {
this._cleanTmpPacks(packDir);
await new Promise(r => setTimeout(r, 1000 * attempt));
}
}
}
if (synced) {
// fsck — even if fetch claimed success, the object graph might
// be broken (refs point at objects that aren't in any local
// pack, because previous failed fetches left stale refs or
// the force-refspec updated refs without their full closure).
// Plain `git fetch` is a no-op in this state (refs match
// remote), so corruption stays forever unless we escalate.
// Check fsck, and if broken, try `git fetch --refetch` —
// git's own primitive for "redownload everything".
const fsckBroken = await this._fsckMissingObjects(repoPath);
if (fsckBroken) {
this.emit('pull', remoteName, name, 'refetch');
try {
this._cleanTmpPacks(packDir);
await this._gitRefetch(repoPath, fetchUrl);
// Re-check after refetch. If still broken, report
// failure — don't destroy local state.
const stillBroken = await this._fsckMissingObjects(repoPath);
if (stillBroken) synced = false;
} catch (err) {
synced = false;
lastErr = err;
}
}
}
if (synced) {
result.synced.push(name);
this.emit('pull', remoteName, name, 'complete');
} else {
// Failure = report it. Don't destroy local state. Git's own
// tools (fetch/fetch --refetch/fsck/repack) have already had
// their shot. If those didn't recover, the caller decides
// what to do — not this code. Previous revisions nuked the
// bare repo under various conditions; that was never the
// plan and has been removed.
result.failed.push({ name, error: lastErr ? lastErr.message : 'unknown error', attempts: maxAttempts });
this.emit('pull', remoteName, name, 'failed');
}
}
result.elapsed = Date.now() - start;
return result;
}
// ── Helpers for pull recovery ─────────────────────────────
/**
* Run `git fsck --full --no-dangling` and detect missing objects
* or "could not read" errors. Returns true if the object graph is
* broken. False means healthy (or fsck couldn't run — treated as
* healthy so we don't trigger refetch on every pull).
*/
_fsckMissingObjects (repoPath) {
return new Promise((resolve) => {
const proc = spawn('git', ['-C', repoPath, 'fsck', '--full', '--no-dangling'], {
stdio: ['ignore', 'pipe', 'pipe']
});
const out = [];
const err = [];
proc.stdout.on('data', (d) => out.push(d));
proc.stderr.on('data', (d) => err.push(d));
proc.on('exit', (code) => {
const combined = (Buffer.concat(out).toString() + Buffer.concat(err).toString()).toLowerCase();
// git fsck uses 'missing', 'could not read', and 'broken link'
// for the cases we care about. Exit code != 0 is another
// signal but not always reliable (some warnings return 0).
const broken = /missing (blob|commit|tree|tag)|could not read|broken link/.test(combined);
resolve(broken);
});
proc.on('error', () => resolve(false));
});
}
/**
* `git fetch --refetch` — git's own primitive (>=2.41) for
* "redownload everything regardless of what I already have."
* Heavy but surgical. Only called when fsck detected missing
* objects and plain fetch would be a no-op.
*/
_gitRefetch (repoPath, fetchUrl) {
return new Promise((resolve, reject) => {
const proc = spawn('git', [
'-C', repoPath, 'fetch', '--refetch', fetchUrl, '+refs/*:refs/*', '--no-tags'
], { stdio: ['ignore', 'pipe', 'pipe'], env: this._gitEnv() });
const errChunks = [];
proc.stderr.on('data', (d) => errChunks.push(d));
proc.on('exit', (code) => {
if (code === 0) resolve();
else reject(new Error('git fetch --refetch exit ' + code + ': ' + Buffer.concat(errChunks).toString('utf-8').trim()));
});
proc.on('error', reject);
});
}
_cleanTmpPacks (packDir) {
try {
for (const f of fs.readdirSync(packDir)) {
if (f.startsWith('tmp_pack_')) {
try { fs.unlinkSync(path.join(packDir, f)) } catch {}
}
}
} catch {}
}
// ── Connection Handler ────────────────────────────────────
_onConnection (conn, info) {
const peerKey = conn.remotePublicKey
? conn.remotePublicKey.toString('hex').slice(0, 12)
: 'unknown';
// Determine which repo this connection is for
// by matching the topic against our announced topics
let repoName = null;
for (const [name, entry] of this._topics) {
// Topics match by the swarm's internal routing
repoName = name;
break;
}
conn.on('error', safetyCatch);
this.emit('connection', peerKey, repoName);
// Handle git protocol — read header, spawn git-upload-pack
let headerReceived = false;
let buffer = Buffer.alloc(0);
const headerHandler = (chunk) => {
if (headerReceived) return;
buffer = Buffer.concat([buffer, chunk]);
const firstNull = buffer.indexOf(0);
if (firstNull === -1) return;
const secondNull = buffer.indexOf(0, firstNull + 1);
if (secondNull === -1) return;
headerReceived = true;
conn.off('data', headerHandler);
const service = buffer.slice(0, firstNull).toString('utf-8');
const repo = buffer.slice(firstNull + 1, secondNull).toString('utf-8');
const remaining = buffer.slice(secondNull + 1);
const repoPath = path.join(this._reposDir, repo + '.git');
if (!fs.existsSync(repoPath)) {
conn.destroy();
return;
}
// git-upload-pack (read) is always open — clone/fetch is unrestricted.
if (service === 'git-upload-pack') {
this._spawnGitService(conn, service, repoPath, remaining);
return;
}
// Any write service (git-receive-pack) requires the authorize hook
// to approve. Without a hook configured, stay read-only (back-compat).
if (!this._authorize) {
conn.destroy();
return;
}
// Pause until the (async) decision lands so no pack bytes are dropped
// in the gap; _spawnGitService resumes the conn once it attaches.
conn.pause();
Promise.resolve(this._authorize({ service, repo, remotePublicKey: conn.remotePublicKey }))
.then((ok) => {
if (ok) this._spawnGitService(conn, service, repoPath, remaining);
else conn.destroy();
})
.catch(() => conn.destroy());
};
conn.on('data', headerHandler);
}
_spawnGitService (conn, service, repoPath, initialData) {
const proc = spawn(service, [repoPath], {
stdio: ['pipe', 'pipe', 'pipe']
});
if (initialData.length > 0) {
proc.stdin.write(initialData);
}
// Bidirectional proxy with backpressure
conn.on('data', (chunk) => {
if (!proc.stdin.write(chunk)) conn.pause();
});
proc.stdin.on('drain', () => conn.resume());
proc.stdout.on('data', (chunk) => {
if (!conn.write(chunk)) proc.stdout.pause();
});
conn.on('drain', () => proc.stdout.resume());
if (typeof conn.resume === 'function') conn.resume();
conn.on('end', () => { try { proc.stdin.end(); } catch {} });
// Error teardown — mirrors hyperbeam (canonical Duplex-over-DHT
// pattern): conn errors destroy the parent stream rather than
// getting silently swallowed. Without this, a transient libudx /
// secret-stream error mid-transfer leaves proc.stdout paused
// awaiting a 'drain' that the broken conn will never emit → 30s
// idle stall on the client side. With teardown, the error
// promptly tears down both ends and the client's retry loop can
// act on it.
const teardown = (err) => {
if (proc.exitCode === null) {
try { proc.kill('SIGTERM'); } catch {}
}
if (!conn.destroyed) {
try { conn.destroy(err); } catch {}
}
};
conn.on('error', teardown);
proc.stdin.on('error', teardown);
proc.stdout.on('error', teardown);
// Use 'close' not 'exit'. 'exit' fires when the process terminates
// but BEFORE all of proc.stdout has been emitted to our 'data'
// listener — calling conn.end() at that point truncates the tail
// of the pack. 'close' fires after all stdio streams have ended,
// guaranteeing every byte git-upload-pack wrote has reached
// conn.write() before we FIN the conn.
proc.on('close', () => {
if (!conn.destroyed) conn.end();
});
conn.on('close', () => {
if (proc.exitCode === null) proc.kill();
});
}
// ── Helpers ───────────────────────────────────────────────
// git-remote-punch is the remote helper that makes `git fetch punch://...`
// work — a SEPARATE bare app, not something git ships. Rather than rely on it
// being hand-installed on every peer's PATH (it usually isn't, and is
// off-PATH even on machines that have it), self-provision: locate our bundled
// git-remote-punch + the running bare, write a tiny shim into a helper dir,
// and prepend that dir to the PATH of the `git` child we spawn. Idempotent +
// cached; returns the helper dir, or null when we can't provision (git then
// falls back to whatever `git-remote-punch` is already on PATH — no regression
// for hosts that already have it).
_ensureRemoteHelper () {
// Cache ONLY a successful dir. A null/failed result is NOT cached, so we
// RETRY on the next pull — the first call can fire very early in boot before
// module resolution is ready (returning null); caching that null (the old
// `!== undefined` guard) stuck the daemon helper-less forever → org joiners
// got refs but zero objects even with the resolution fix in place.
if (this._remoteHelperDir) return this._remoteHelperDir;
try {
// Resolve git-remote-punch's bin. ORDER MATTERS: prefer the MODULE-RESOLVED
// (hoisted top-level) copy. In content-addressed pgh bundles, hyper-sync's
// NESTED node_modules/git-remote-punch is deduped/pruned — its `bin/` can be
// ABSENT while the package dir lingers, so a shim pointing at that missing
// file makes bare abort (SIGABRT) on EVERY fetch (org joiners then get refs
// but zero objects). The hoisted top-level copy is the complete one.
const candidates = [];
try {
// Follows node_modules resolution to the hoisted top-level copy.
// (Available under Node; may be absent under bare — hence the walk below.)
const pkg = require.resolve('git-remote-punch/package.json');
candidates.push(path.join(path.dirname(pkg), 'bin', 'git-remote-punch'));
} catch {}
// bare-safe fallback: walk ancestors for a HOISTED node_modules/git-remote-punch
// (the top-level, COMPLETE copy). Does not rely on require.resolve.
for (let d = __dirname, prev = null; d !== prev; prev = d, d = path.dirname(d)) {
candidates.push(path.join(d, 'node_modules', 'git-remote-punch', 'bin', 'git-remote-punch'));
}
// Dev tree (sibling) + immediate-nested fallbacks.
candidates.push(path.join(__dirname, '..', 'git-remote-punch', 'bin', 'git-remote-punch'));
candidates.push(path.join(__dirname, 'node_modules', 'git-remote-punch', 'bin', 'git-remote-punch'));
let helperBin = null;
for (const c of candidates) {
// Verify the actual BIN FILE exists (isFile) — a pruned package dir must
// NOT be selected, or we shim a missing path and crash. Fall through, and
// finally to null (git uses whatever git-remote-punch is already on PATH).
try { if (fs.existsSync(c) && fs.statSync(c).isFile()) { helperBin = c; break; } } catch {}
}
if (!helperBin) return null;
// git-remote-punch is `#!/usr/bin/env bare`, so the shim runs it under the
// bare running us. Bare.argv[0] is authoritative under bare; execPath is
// the fallback.
const barePath = (typeof Bare !== 'undefined' && Bare.argv && Bare.argv[0])
? Bare.argv[0]
: process.execPath;
if (!barePath) return null;
// Shim lives under the repos dir but NOT as a *.git, so repo scans ignore
// it. Rewritten each boot (cheap) so bare/helper paths stay current.
const dir = path.join(this._reposDir, '.remote-helpers');
fs.mkdirSync(dir, { recursive: true });
const shimPath = path.join(dir, 'git-remote-punch');
fs.writeFileSync(shimPath, '#!/bin/sh\nexec "' + barePath + '" "' + helperBin + '" "$@"\n');
fs.chmodSync(shimPath, 0o755);
this._remoteHelperDir = dir;
return dir;
} catch {
return null; // not cached — retry next call
}
}
// Env for a `git` child, with the self-provisioned git-remote-punch helper
// dir prepended to PATH so `git fetch punch://...` resolves the helper.
_gitEnv () {
const dir = this._ensureRemoteHelper();
const env = Object.assign({}, process.env);
if (dir) env.PATH = dir + path.delimiter + (env.PATH || '');
return env;
}
_gitFetch (repoPath, fetchUrl) {
return new Promise((resolve, reject) => {
// '+refs/*:refs/*' — leading '+' means force-update. Lets us
// recover when the remote rewrote history (e.g. filter-repo
// strip of node_modules). Without it, git fetch rejects
// non-fast-forward refs and the mirror stays stuck on old
// SHAs forever. Old objects become unreachable on the next
// fetch and get reclaimed by the next gc.
const proc = spawn('git', [
'-C', repoPath, 'fetch', fetchUrl, '+refs/*:refs/*', '--no-tags'
], { stdio: ['ignore', 'pipe', 'pipe'], env: this._gitEnv() });
const errChunks = [];
proc.stderr.on('data', (d) => errChunks.push(d));
proc.on('exit', (code) => {
if (code === 0) resolve();
else reject(new Error('git fetch exit ' + code + ': ' + Buffer.concat(errChunks).toString('utf-8').trim()));
});
proc.on('error', reject);
});
}
_exec (cmd, args) {
return new Promise((resolve, reject) => {
const proc = spawn(cmd, args, { stdio: ['ignore', 'pipe', 'pipe'] });
const chunks = [];
proc.stdout.on('data', (c) => chunks.push(c));
proc.on('exit', (code) => {
if (code === 0) resolve(Buffer.concat(chunks).toString('utf-8'));
else reject(new Error(cmd + ' exited with code ' + code));
});
proc.on('error', reject);
});
}
};
// Mix in browser methods (listFiles, readFile, getCommits, getDiff, etc.)
Object.assign(module.exports.prototype, browserMethods);