hyper-deploy / lib / protocol.js

/**
 * Deploy Protocol — Protomux channel for release distribution
 *
 * Messages:
 *   0: fileList    (server→client)  JSON array of { path, contentHash, size }
 *   1: fileRequest (client→server)  JSON { path }
 *   2: chunkInfo   (server→client)  JSON { path, size, chunks: [{ hash, offset, length }] }
 *   3: chunkData   (server→client)  4-byte LE index + raw chunk bytes
 *   4: control     (bidirectional)  JSON { type, ... }
 *
 * Backpressure: ondrain callback on Protomux channel.
 */

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

const PROTOCOL_NAME = 'hyper-deploy'
const PROTOCOL_VERSION = 2

/**
 * Attach the deploy protocol to a stream.
 *
 * @param {Stream} stream — encrypted stream (SecretStream or Hyperswarm connection)
 * @returns {{ channel, msgs, send, onMessage, destroy } | null}
 */
export function attachProtocol (stream) {
  const mux = Protomux.from(stream)
  if (!mux) return null

  let drainResolve = null

  const channel = mux.createChannel({
    protocol: PROTOCOL_NAME,
    ondrain () {
      if (drainResolve) {
        const r = drainResolve
        drainResolve = null
        r()
      }
    }
  })

  if (!channel) return null

  const fileListMsg = channel.addMessage({ encoding: c.raw })
  const fileRequestMsg = channel.addMessage({ encoding: c.raw })
  const chunkInfoMsg = channel.addMessage({ encoding: c.raw })
  const chunkDataMsg = channel.addMessage({ encoding: c.raw })
  const controlMsg = channel.addMessage({ encoding: c.raw })

  channel.open()

  async function waitDrain () {
    await new Promise(r => { drainResolve = r })
  }

  async function sendRaw (msg, buf) {
    if (!msg.send(buf)) await waitDrain()
  }

  return {
    channel,
    version: PROTOCOL_VERSION,

    // ── Senders ──────────────────────────────────────────────

    async sendFileList (files) {
      await sendRaw(fileListMsg, Buffer.from(JSON.stringify(files)))
    },

    async sendFileRequest (path) {
      await sendRaw(fileRequestMsg, Buffer.from(JSON.stringify({ path })))
    },

    async sendChunkInfo (info) {
      await sendRaw(chunkInfoMsg, Buffer.from(JSON.stringify(info)))
    },

    async sendChunkData (index, data) {
      const header = Buffer.alloc(4)
      header.writeUInt32LE(index)
      await sendRaw(chunkDataMsg, Buffer.concat([header, data]))
    },

    async sendControl (obj) {
      await sendRaw(controlMsg, Buffer.from(JSON.stringify(obj)))
    },

    // ── Receivers ─────────────────────────────────────────────

    onFileList (fn) {
      fileListMsg.onmessage = (raw) => fn(JSON.parse(raw.toString()))
    },

    onFileRequest (fn) {
      fileRequestMsg.onmessage = (raw) => fn(JSON.parse(raw.toString()))
    },

    onChunkInfo (fn) {
      chunkInfoMsg.onmessage = (raw) => fn(JSON.parse(raw.toString()))
    },

    onChunkData (fn) {
      chunkDataMsg.onmessage = (raw) => {
        const index = raw.readUInt32LE(0)
        const data = raw.subarray(4)
        fn(index, data)
      }
    },

    onControl (fn) {
      controlMsg.onmessage = (raw) => fn(JSON.parse(raw.toString()))
    },

    destroy () {
      channel.close()
    }
  }
}

export { PROTOCOL_NAME, PROTOCOL_VERSION }