pear-git / lib / cli / p2p-client.js

/**
 * P2P Client for pgh admin commands
 *
 * Connects to the pear-git server's __admin__ topic over Hyperswarm
 * and sends JSON commands via the pgh-admin protocol.
 *
 * Uses the same persistent Noise keypair as git-remote-punch,
 * so the server recognizes this peer by its identity registration.
 */

import { TopicConnector } from '../p2p/discovery/topic-manager.js'

export class P2PClient {
  constructor (options = {}) {
    this.serverKey = options.serverKey
    this.timeout = options.timeout || 30000
    this.connector = null
  }

  /**
   * Initialize the P2P connector (loads persistent keypair)
   */
  async init () {
    this.connector = new TopicConnector({ timeout: this.timeout })
    await this.connector.init()
  }

  /**
   * Get this peer's Noise public key (hex)
   */
  getPublicKeyHex () {
    return this.connector ? this.connector.getPublicKeyHex() : null
  }

  /**
   * Send an admin command to the server and get a response
   *
   * @param {Object} request - Command object (must include { command })
   * @returns {Promise<Object>} Server response
   */
  async command (request) {
    // Connect to __admin__ topic
    const conn = await this.connector.connect(this.serverKey, '__admin__')

    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        conn.destroy()
        reject(new Error('Command timeout'))
      }, this.timeout)

      // Send pgh-admin header + JSON request as single write
      // (avoids race between header handler removal and admin data handler setup)
      const header = Buffer.concat([
        Buffer.from('pgh-admin', 'utf-8'),
        Buffer.from([0]),
        Buffer.from('admin', 'utf-8'),
        Buffer.from([0]),
        Buffer.from(JSON.stringify(request) + '\n', 'utf-8')
      ])

      conn.write(header)

      // Accumulate response until newline
      let responseBuffer = Buffer.alloc(0)

      conn.on('data', (chunk) => {
        responseBuffer = Buffer.concat([responseBuffer, chunk])

        const newlineIdx = responseBuffer.indexOf(10) // \n
        if (newlineIdx !== -1) {
          clearTimeout(timeout)
          const responseStr = responseBuffer.slice(0, newlineIdx).toString('utf-8')
          try {
            resolve(JSON.parse(responseStr))
          } catch (err) {
            reject(new Error(`Invalid server response: ${responseStr.slice(0, 100)}`))
          }
        }
      })

      conn.on('error', (err) => {
        clearTimeout(timeout)
        reject(err)
      })

      conn.on('close', () => {
        clearTimeout(timeout)
        // If we haven't resolved yet, check if we have data
        if (responseBuffer.length > 0) {
          try {
            resolve(JSON.parse(responseBuffer.toString('utf-8').trim()))
          } catch {
            reject(new Error('Connection closed before response'))
          }
        }
      })
    })
  }

  /**
   * Clean up resources
   */
  async destroy () {
    if (this.connector) {
      await this.connector.destroy()
      this.connector = null
    }
  }
}