pear-git / lib / p2p / federation / server-sync.js

/**
 * ServerSync — Periodic git fetch from federated peers via P2P
 *
 * Reads federation entries from the Hyperbee registry and runs
 * `git fetch` for each repo from each active peer using punch:// URLs.
 * This keeps git objects in sync across federated pear-git servers
 * over Hyperswarm — no HTTP, no DNS, no hostname resolution needed.
 *
 * Respects peering mode:
 *   mirror — sync ALL repos
 *   peer   — sync only grantedRepos (if specified)
 */

import { spawn } from 'bare-subprocess'
import path from 'bare-path'
import fs from 'bare-fs'

export class ServerSync {
  constructor ({ reposDir, registry, interval }) {
    this.reposDir = reposDir
    this.registry = registry // Hyperbee instance
    this.interval = interval || 30 * 60 * 1000 // 30 min fallback (primary sync is push-triggered)
    this._timer = null
    this._running = false
  }

  start () {
    if (this._timer) return
    console.log(`[ServerSync] Starting periodic sync every ${this.interval / 1000}s`)
    setTimeout(() => this._tick(), 30000) // initial delay
    this._timer = setInterval(() => this._tick(), this.interval)
  }

  stop () {
    if (this._timer) {
      clearInterval(this._timer)
      this._timer = null
    }
  }

  async _tick () {
    if (this._running) return
    this._running = true
    try {
      const peers = await this._buildSyncList()
      if (peers.length === 0) return
      const allRepos = await this._listRepos()
      if (allRepos.length === 0) return

      for (const peer of peers) {
        // Determine which repos to sync based on peering mode
        const repos = this._filterReposForPeer(allRepos, peer)
        // Process repos in parallel batches of 5
        for (let i = 0; i < repos.length; i += 5) {
          const batch = repos.slice(i, i + 5)
          await Promise.allSettled(
            batch.map(repo =>
              this._syncRepo(repo, peer).catch(err => {
                if (!err.message.includes('exit 128')) {
                  console.error(`[ServerSync] Failed ${repo} from ${peer.serverKey.slice(0, 8)}: ${err.message}`)
                }
              })
            )
          )
        }
      }
    } catch (err) {
      console.error(`[ServerSync] Tick error: ${err.message}`)
    } finally {
      this._running = false
    }
  }

  /**
   * Filter repos based on peer mode.
   * mirror: all repos
   * peer: only grantedRepos (or all if no grants specified — backwards compat)
   */
  _filterReposForPeer (allRepos, peer) {
    // Mirror mode or legacy (no mode) — sync everything
    if (!peer.mode || peer.mode === 'mirror') {
      return allRepos
    }

    // Peer mode with specific grants — only sync those
    if (peer.mode === 'peer' && peer.grantedRepos && peer.grantedRepos.length > 0) {
      const granted = new Set(peer.grantedRepos)
      return allRepos.filter(r => granted.has(r))
    }

    // Peer mode with no grants — sync everything (backwards compat)
    return allRepos
  }

  /**
   * Build fetch URL for a repo from a peer.
   * Prefers punch:// (P2P via Hyperswarm), falls back to HTTP if no serverKey.
   */
  _buildFetchUrl (repoName, peer) {
    // P2P via punch:// — uses git-remote-punch helper over Hyperswarm
    if (peer.serverKey) {
      return `punch://${peer.serverKey}/${repoName}`
    }
    // HTTP fallback (legacy peers without serverKey)
    if (peer.httpUrl) {
      return `${peer.httpUrl}/${repoName}.git`
    }
    return null
  }

  async _buildSyncList () {
    const peers = []
    const stream = this.registry.createReadStream({ gt: 'federation:', lt: 'federation:\xff' })
    for await (const entry of stream) {
      try {
        const data = typeof entry.value === 'string' ? JSON.parse(entry.value) : entry.value
        // Require serverKey for P2P sync, or httpUrl as fallback
        if (data.status === 'active' && (data.serverKey || data.httpUrl)) {
          peers.push(data)
        }
      } catch {}
    }
    return peers
  }

  async _listRepos () {
    try {
      const entries = await fs.promises.readdir(this.reposDir)
      return entries.filter(e => e.endsWith('.git')).map(e => e.replace(/\.git$/, ''))
    } catch {
      return []
    }
  }

  async _syncRepo (repoName, peer) {
    const repoPath = path.join(this.reposDir, `${repoName}.git`)
    const fetchUrl = this._buildFetchUrl(repoName, peer)
    if (!fetchUrl) return

    return new Promise((resolve, reject) => {
      let done = false
      const child = spawn('git', [
        '-C', repoPath, 'fetch', fetchUrl,
        'refs/*:refs/*', '--no-tags'
      ])

      // bare-subprocess doesn't support timeout — implement manually
      const timer = setTimeout(() => {
        if (done) return
        done = true
        try { child.kill('SIGKILL') } catch {}
        reject(new Error(`git fetch timeout after 60s for ${repoName}`))
      }, 60000)

      let stderr = ''
      child.stderr.on('data', (d) => { stderr += d.toString() })
      child.on('exit', (code) => {
        if (done) return
        done = true
        clearTimeout(timer)
        if (code === 0) resolve()
        else reject(new Error(`git fetch exit ${code}: ${stderr.trim()}`))
      })
      child.on('error', (err) => {
        if (done) return
        done = true
        clearTimeout(timer)
        reject(err)
      })
    })
  }
}