modules/media-streaming/hyper-p2p-media-chunker/index.js

require('bare-process/global')
const EventEmitter = require('bare-events')
const b4a = require('b4a')
const crypto = require('hypercore-crypto')
const { CHUNK_TYPES, assertStreamId, chunkKey, mediaStats } = require('../../_shared/media-streaming-base.js')

const PROTOCOL = 'media-chunker/v1'

class HyperP2PMediaChunker extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.chunkSize = opts.chunkSize ?? 256 * 1024
    this.keyframeAlign = opts.keyframeAlign !== false
    this._chunks = new Map()
    this._stats = { segmented: 0, bytes: 0 }
  }

  segment (payload, opts = {}) {
    if (payload == null) throw new Error('payload required')
    const streamId = assertStreamId(opts.streamId || 'default')
    const buf = b4a.isBuffer(payload) ? payload : b4a.from(payload)
    const type = opts.keyframe ? CHUNK_TYPES.KEYFRAME : (opts.type || CHUNK_TYPES.DELTA)
    const size = opts.chunkSize ?? this.chunkSize
    const out = []
    let seq = opts.startSeq ?? 0
    for (let off = 0; off < buf.length; off += size) {
      const slice = buf.subarray(off, Math.min(off + size, buf.length))
      const id = chunkKey(streamId, seq)
      const chunk = {
        id,
        streamId,
        seq,
        type,
        data: slice,
        byteLength: slice.length,
        keyframe: type === CHUNK_TYPES.KEYFRAME || (this.keyframeAlign && seq === 0),
        at: Date.now()
      }
      this._chunks.set(id, chunk)
      out.push(chunk)
      seq++
    }
    this._stats.segmented += out.length
    this._stats.bytes += buf.length
    this.emit('segment', { streamId, count: out.length })
    return out
  }

  getChunk (chunkId) {
    const c = this._chunks.get(chunkId)
    return c ? { ...c, data: b4a.from(c.data) } : null
  }

  listChunks (streamId) {
    const sid = assertStreamId(streamId)
    return [...this._chunks.values()].filter((c) => c.streamId === sid).sort((a, b) => a.seq - b.seq)
  }

  hashChunk (chunkId) {
    const c = this._chunks.get(chunkId)
    if (!c) return null
    return b4a.toString(crypto.hash(c.data), 'hex')
  }

  keyframeIndices (streamId) {
    return this.listChunks(streamId)
      .filter((c) => c.keyframe || c.type === CHUNK_TYPES.KEYFRAME)
      .map((c) => c.seq)
  }

  totalBytes (streamId) {
    return this.listChunks(streamId).reduce((s, c) => s + c.byteLength, 0)
  }

  pruneBefore (streamId, seq) {
    const sid = assertStreamId(streamId)
    let n = 0
    for (const [id, c] of this._chunks) {
      if (c.streamId === sid && c.seq < seq) {
        this._chunks.delete(id)
        n++
      }
    }
    if (n) this.emit('prune', { streamId: sid, before: seq, count: n })
    return n
  }

  getStats () {
    return mediaStats(this._stats, PROTOCOL, { stored: this._chunks.size })
  }

  async ready () { return this }

  async close () {
    this._chunks.clear()
    this.emit('closed')
  }
}

module.exports = { HyperP2PMediaChunker, PROTOCOL, CHUNK_TYPES }