Expand media, network-stack, pear-platform, and scheduling module APIs

Add batch helpers, clear/count utilities, and management methods across content protection, FEC, flow shaper, circuit loom, trust gate, peer scheduler, and related modules.

Co-authored-by: Cursor <[email protected]>

encoding-wire/hyper-p2p-message-envelope/index.js

@@ -94,6 +94,16 @@ class HyperP2PMessageEnvelope extends EventEmitter {
     return { ...envelope, payload, at: envelope.at || Date.now() }
   }
 
+  wrapBatch (payloads, opts = {}) {
+    if (!Array.isArray(payloads)) throw new Error('payloads must be an array')
+    return payloads.map((p) => this.wrap(p, opts))
+  }
+
+  unwrapBatch (envelopes) {
+    if (!Array.isArray(envelopes)) throw new Error('envelopes must be an array')
+    return envelopes.map((e) => this.unwrap(e))
+  }
+
   resetStats () {
     this._stats = { wrapped: 0, unwrapped: 0, encoded: 0, decoded: 0, failed: 0 }
     return this._stats

measurement-rate-control/hyper-p2p-sla-budget/index.js

@@ -124,6 +124,15 @@ class HyperP2PSlaBudget extends EventEmitter {
     return true
   }
 
+  serviceCount () {
+    return this._services.size
+  }
+
+  allocateBatch (allocations) {
+    if (!Array.isArray(allocations)) throw new Error('allocations must be an array')
+    return allocations.map((a) => this.allocate(a.service, a.amount))
+  }
+
   getStats () {
     return {
       ...this._stats,

media-streaming/hyper-p2p-content-protection/index.js

@@ -81,6 +81,19 @@ class HyperP2PContentProtection extends EventEmitter {
     return payloads.map((p) => this.encryptSegment(p, keyId))
   }
 
+  decryptBatch (segments, keyId = null) {
+    if (!Array.isArray(segments)) throw new Error('segments array required')
+    return segments.map((s) => this.decryptSegment(s.data || s, keyId))
+  }
+
+  hasKey (keyId) {
+    return this._keys.has(String(keyId))
+  }
+
+  keyCount () {
+    return this._keys.size
+  }
+
   clearKeys () {
     const n = this._keys.size
     this._keys.clear()

media-streaming/hyper-p2p-fec-video/index.js

@@ -93,6 +93,16 @@ class HyperP2PFecVideo extends EventEmitter {
     return entries.map((e) => this.encodeGroup(e.groupId, e.shards))
   }
 
+  clearAll () {
+    const n = this._groups.size
+    this._groups.clear()
+    return n
+  }
+
+  groupCount () {
+    return this._groups.size
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { groups: this._groups.size })
   }

media-streaming/hyper-p2p-latency-optimizer/index.js

@@ -87,6 +87,15 @@ class HyperP2PLatencyOptimizer extends EventEmitter {
     return n
   }
 
+  pathCount () {
+    return this._paths.size
+  }
+
+  measureBatch (entries) {
+    if (!Array.isArray(entries)) throw new Error('entries must be an array')
+    return entries.map((e) => this.measurePath(e.pathId, e.hops || []))
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { paths: this._paths.size })
   }

media-streaming/hyper-p2p-origin-hybrid-bridge/index.js

@@ -90,6 +90,16 @@ class HyperP2POriginHybridBridge extends EventEmitter {
     return this._stats.originBytes + this._stats.p2pBytes
   }
 
+  clearStream (streamId) {
+    return this._origins.delete(assertStreamId(streamId))
+  }
+
+  clearAll () {
+    const n = this._origins.size
+    this._origins.clear()
+    return n
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, {
       offloadRatio: this.p2pOffloadRatio(),

messaging-pubsub/hyper-p2p-qos-topic/index.js

@@ -94,6 +94,17 @@ class HyperP2PQosTopic extends EventEmitter {
     return n
   }
 
+  handlerCount () {
+    return this._handlers.size
+  }
+
+  peekHighest () {
+    for (let q = MAX_QOS; q >= 0; q--) {
+      if (this._queues[q].length) return { ...this._queues[q][0] }
+    }
+    return null
+  }
+
   _onGossip (data) {
     if (!data || data.type !== 'qos-publish') return
     this._stats.gossipIn++

messaging-streams/hyper-p2p-stream-chunker/index.js

@@ -87,6 +87,15 @@ class HyperP2PStreamChunker extends EventEmitter {
     return { tail, data: out }
   }
 
+  pendingBytes () {
+    return this._pending.length
+  }
+
+  writeBatch (chunks) {
+    if (!Array.isArray(chunks)) throw new Error('chunks must be an array')
+    return chunks.map((c) => this.write(c))
+  }
+
   getStats () {
     return {
       ...this._stats,

network-stack/hyper-p2p-circuit-loom/index.js

@@ -98,6 +98,15 @@ class HyperP2PCircuitLoom extends EventEmitter {
     return n
   }
 
+  circuitCount () {
+    return this._circuits.size
+  }
+
+  hopCount (circuitId) {
+    const c = this._circuits.get(circuitId)
+    return c ? c.hops.length : 0
+  }
+
   getStats () {
     return { ...this._stats, open: this.openCount(), total: this._circuits.size, protocol: PROTOCOL }
   }

network-stack/hyper-p2p-flow-shaper/index.js

@@ -79,6 +79,23 @@ class HyperP2PFlowShaper extends EventEmitter {
     for (const p of PRIOS) this._tokens[p] = Math.min(this._rates[p], this._tokens[p] + Math.floor(this._rates[p] / 10))
   }
 
+  clearQueues () {
+    let n = 0
+    for (const p of PRIOS) {
+      n += this._queues[p].length
+      this._queues[p] = []
+    }
+    return n
+  }
+
+  setRate (priority, rate) {
+    if (!PRIOS.includes(priority)) throw new Error('priority must be control|data|bulk')
+    if (rate < 0) throw new Error('rate must be non-negative')
+    this._rates[priority] = rate
+    this._tokens[priority] = rate
+    return rate
+  }
+
   getStats () {
     return {
       ...this._stats,

network-transport/hyper-p2p-secret-stream-pair/index.js

@@ -82,6 +82,16 @@ class HyperP2PSecretStreamPair {
     this._stats = { pairs: 0 }
   }
 
+  publicKeyHex () {
+    return require('b4a').toString(this.keyPair.publicKey, 'hex')
+  }
+
+  rotateKeyPair () {
+    this.destroyActive()
+    this.keyPair = hypercoreCrypto.keyPair()
+    return this.keyPair
+  }
+
   getStats () {
     return { ...this._stats, protocol: PROTOCOL }
   }

observability/hyper-p2p-health-probe/index.js

@@ -67,6 +67,20 @@ class HyperP2PHealthProbe extends EventEmitter {
     return n
   }
 
+  reportBatch (entries) {
+    if (!Array.isArray(entries)) throw new Error('entries must be an array')
+    return entries.map((e) => this.report(e.peerId, e.status || e))
+  }
+
+  isHealthy (peerId) {
+    const r = this.get(peerId)
+    return !!(r && r.ok)
+  }
+
+  peerCount () {
+    return this._reports.size
+  }
+
   _onGossip (data) {
     if (!data || data.type !== 'health' || !data.entry) return
     this._stats.gossipIn++

pear-platform/hyper-bare-addon-resolve/index.js

@@ -85,6 +85,16 @@ class HyperBareAddonResolve extends EventEmitter {
     return n
   }
 
+  registerBatch (entries) {
+    if (!Array.isArray(entries)) throw new Error('entries must be an array')
+    return entries.map((e) => this.register(e.name, e.spec || e))
+  }
+
+  resolveBatch (names) {
+    if (!Array.isArray(names)) throw new Error('names must be an array')
+    return names.map((n) => this.resolve(n))
+  }
+
   getStats () {
     return platformStats(this._stats, PROTOCOL, {
       addons: this._addons.size,

pear-platform/hyper-bare-bundle-bridge/index.js

@@ -68,6 +68,15 @@ class HyperBareBundleBridge extends EventEmitter {
     return n
   }
 
+  resolveBatch (ids) {
+    if (!Array.isArray(ids)) throw new Error('ids must be an array')
+    return ids.map((id) => this.resolve(id))
+  }
+
+  bundleCount () {
+    return this._bundles.size
+  }
+
   _gossip (data) {
     if (!this._peerMsgs) return
     gossipSend(this, data)

pear-platform/hyper-bare-target-matrix/index.js

@@ -92,6 +92,16 @@ class HyperBareTargetMatrix extends EventEmitter {
     return this._targets.some((t) => t.id === id)
   }
 
+  resetToDefaults () {
+    this._targets = [...DEFAULT_TARGETS]
+    this.host = this._targets[0]?.id || 'darwin-arm64'
+    return this.list()
+  }
+
+  targetCount () {
+    return this._targets.length
+  }
+
   getStats () {
     return platformStats(this._stats, PROTOCOL, {
       host: this.host,

pear-platform/hyper-pear-preflight-sync/index.js

@@ -122,6 +122,21 @@ class HyperPearPreflightSync extends EventEmitter {
     return [...this._jobs.keys()]
   }
 
+  failTimedOut () {
+    let n = 0
+    for (const job of this.listPending()) {
+      if (this.timedOut(job.id)) {
+        this.fail(job.id, 'warmup timeout')
+        n++
+      }
+    }
+    return n
+  }
+
+  jobCount () {
+    return this._jobs.size
+  }
+
   getStats () {
     return platformStats(this._stats, PROTOCOL, {
       jobs: this._jobs.size,

pear-platform/hyper-pear-trust-gate/index.js

@@ -94,6 +94,16 @@ class HyperPearTrustGate extends EventEmitter {
     return n
   }
 
+  checkBatch (keys, autoTrust = false) {
+    if (!Array.isArray(keys)) throw new Error('keys must be an array')
+    return keys.map((k) => this.check(k, autoTrust))
+  }
+
+  trustBatch (keys) {
+    if (!Array.isArray(keys)) throw new Error('keys must be an array')
+    return keys.map((k) => this.trust(k))
+  }
+
   getStats () {
     return platformStats(this._stats, PROTOCOL, { trusted: this._trusted.size })
   }

routing-advanced/hyper-p2p-circuit-breaker/index.js

@@ -123,6 +123,24 @@ class HyperP2PCircuitBreaker extends EventEmitter {
     return n
   }
 
+  failuresFor (id) {
+    const c = this._circuits.get(id)
+    return c ? c.failures : 0
+  }
+
+  ensure (id) {
+    if (!this._circuits.has(id)) {
+      this._circuits.set(id, {
+        state: 'closed',
+        failures: 0,
+        successes: 0,
+        openedAt: 0,
+        halfOpenTrials: 0
+      })
+    }
+    return this.getState(id)
+  }
+
   getStats () {
     return {
       ...this._stats,

scheduling-queues/hyper-p2p-peer-scheduler/index.js

@@ -70,6 +70,15 @@ class HyperP2PPeerScheduler extends EventEmitter {
     return ids.length
   }
 
+  jobIds () {
+    return [...this._jobs.keys()]
+  }
+
+  scheduleBatch (entries) {
+    if (!Array.isArray(entries)) throw new Error('entries must be an array')
+    return entries.map((e) => this.schedule(e.expr, e.id, { shard: e.shard }))
+  }
+
   msUntil (jobId, now = Date.now()) {
     const job = this._jobs.get(jobId)
     if (!job) return null

trust-security/hyper-p2p-key-rotation/index.js

@@ -87,6 +87,14 @@ class HyperP2PKeyRotation extends EventEmitter {
     return list.reduce((a, b) => (a.activateAt < b.activateAt ? a : b))
   }
 
+  getActive (keyId) {
+    return this._active.get(String(keyId)) || null
+  }
+
+  pendingKeyIds () {
+    return [...this._pending.keys()]
+  }
+
   _maybeActivate (keyId, now = Date.now()) {
     const list = this._pending.get(keyId) || []
     for (const e of list) {