hyper-svc / lib / transports / swarm-rpc.js

/**
 * Swarm-RPC transport for hyper-svc.
 *
 * Adds a Protomux RPC channel on every Hyperswarm connection. Requests
 * and responses use the same length-prefixed JSON framing as UDS (see
 * frame.js), multiplexed over the Noise-encrypted swarm stream.
 *
 * This transport extends the existing SwarmTransport (replication) with
 * request/response dispatch. A single swarm connection carries:
 *   1. Hypercore/Autobase replication (corestore.replicate)
 *   2. Module mux channels (attachProtocol)
 *   3. RPC frames (this module)
 *
 * Wire protocol over Protomux:
 *   Channel: 'hyper-svc-rpc'
 *   Message type 0 (request):  { id, method, params }
 *   Message type 1 (response): { id, result } | { id, error }
 *
 * Auth: the caller's Noise public key is available on the connection.
 * For pear-git-v3, envelope auth (Phase 2a) is layered on top — the
 * RPC payload itself is a signed envelope, verified by the dispatch
 * handler. The transport just provides the Noise tunnel.
 */

import Protomux from 'protomux'
import c from 'compact-encoding'
import b4a from 'b4a'

const PROTOCOL = 'hyper-svc-rpc'

/**
 * Attach an RPC channel to a Protomux instance (one per swarm connection).
 *
 * Server-side: receives requests, calls dispatch, sends responses.
 *
 * @param {Protomux} mux — the protomux for this connection
 * @param {Function} dispatch — (method, params, ctx) => result
 * @param {object} [opts]
 * @param {Buffer} [opts.remotePublicKey] — peer's Noise pubkey (for ctx)
 * @returns {{ channel, close: () => void }}
 */
export function attachRpcServer (mux, dispatch, opts = {}) {
  const channel = mux.createChannel({
    protocol: PROTOCOL,
    onopen () {},
    onclose () {}
  })

  if (!channel) return null // protocol already claimed on this mux

  const requestType = channel.addMessage({
    encoding: c.json,
    onmessage (msg) {
      handleRequest(msg, channel, responseType, dispatch, opts)
    }
  })

  const responseType = channel.addMessage({
    encoding: c.json,
    onmessage () {} // server doesn't receive responses
  })

  channel.open()

  return { channel, close: () => channel.close() }
}

/**
 * Attach an RPC channel to a Protomux instance (client-side).
 *
 * Client-side: sends requests, receives responses.
 *
 * @param {Protomux} mux
 * @returns {{ call: (method, params) => Promise<any>, close: () => void }}
 */
export function attachRpcClient (mux) {
  let nextId = 1
  const pending = new Map() // id → { resolve, reject }

  const channel = mux.createChannel({
    protocol: PROTOCOL,
    onopen () {},
    onclose () {
      // Reject all pending calls on channel close.
      for (const [id, p] of pending) {
        p.reject(new Error('swarm-rpc channel closed'))
        pending.delete(id)
      }
    }
  })

  if (!channel) throw new Error('swarm-rpc: protocol already claimed')

  const requestType = channel.addMessage({
    encoding: c.json,
    onmessage () {} // client doesn't receive requests
  })

  const responseType = channel.addMessage({
    encoding: c.json,
    onmessage (msg) {
      const p = pending.get(msg.id)
      if (!p) return // stale or unknown id
      pending.delete(msg.id)
      if (msg.error) {
        const err = new Error(msg.error)
        err.code = msg.code || null
        p.reject(err)
      } else {
        p.resolve(msg.result)
      }
    }
  })

  channel.open()

  return {
    call (method, params) {
      return new Promise((resolve, reject) => {
        const id = nextId++
        pending.set(id, { resolve, reject })
        requestType.send({ id, method, params })
      })
    },
    close () {
      channel.close()
    }
  }
}

// --- internals ---

async function handleRequest (msg, channel, responseType, dispatch, opts) {
  const { id, method, params } = msg
  if (!id || !method) {
    responseType.send({ id: id || 0, error: 'invalid request (missing id or method)' })
    return
  }

  const ctx = {
    transport: 'swarm-rpc',
    identity: {
      type: 'swarm',
      remotePublicKey: opts.remotePublicKey
        ? b4a.toString(opts.remotePublicKey, 'hex')
        : null,
      trust: 'network'
    }
  }

  try {
    const result = await dispatch(method, params || {}, ctx)
    responseType.send({ id, result: result !== undefined ? result : null })
  } catch (err) {
    responseType.send({
      id,
      error: err.message || String(err),
      code: err.code || null
    })
  }
}