Expand pear-platform, observability, messaging, and network module APIs

Add batch parse, registry, log filtering, stream helpers, routing utilities, handshake management, and related clear/list/count methods across shorter modules.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Raven Scott
2026-05-21 03:12:42 -04:00
co-authored by Cursor
parent b663b5413c
commit c3b112f435
26 changed files with 278 additions and 4 deletions
@@ -36,7 +36,7 @@ class HyperP2PCreditLedger extends EventEmitter {
return this._apply(accountId, -Math.abs(amount), 'debit', reason) return this._apply(accountId, -Math.abs(amount), 'debit', reason)
} }
transfer (fromId, toId, amount) { transfer (fromId, toId, amount, reason = '') {
assertNonEmpty(fromId, 'fromId') assertNonEmpty(fromId, 'fromId')
assertNonEmpty(toId, 'toId') assertNonEmpty(toId, 'toId')
if (amount <= 0) throw new Error('amount must be positive') if (amount <= 0) throw new Error('amount must be positive')
@@ -44,9 +44,10 @@ class HyperP2PCreditLedger extends EventEmitter {
const to = this._accounts.get(toId) const to = this._accounts.get(toId)
if (!from || !to) throw new Error('unknown account') if (!from || !to) throw new Error('unknown account')
if (from.balance < amount) throw new Error('insufficient balance') if (from.balance < amount) throw new Error('insufficient balance')
this.debit(fromId, amount, `transfer to ${toId}`) const tag = reason || `transfer:${fromId}->${toId}`
this.credit(toId, amount, `transfer from ${fromId}`) this.debit(fromId, amount, tag)
return { from: fromId, to: toId, amount, at: Date.now() } this.credit(toId, amount, tag)
return { from: fromId, to: toId, amount, reason: tag, at: Date.now() }
} }
balance (accountId) { balance (accountId) {
@@ -115,6 +115,14 @@ class HyperP2PLeaderLease extends EventEmitter {
return Math.max(0, this._leaseUntil - Date.now()) return Math.max(0, this._leaseUntil - Date.now())
} }
currentTerm () {
return this._term
}
followerCount () {
return this._followers.size
}
_gossipSync () { _gossipSync () {
gossipSend(this, { gossipSend(this, {
type: 'leader-lease-sync', type: 'leader-lease-sync',
@@ -225,6 +225,26 @@ class CapabilityManager extends EventEmitter {
return caps.length return caps.length
} }
recordReceived (resource, cap) {
if (!resource || !cap) return false
if (!this.received.has(resource)) this.received.set(resource, [])
this.received.get(resource).push(cap)
return true
}
receivedCount (resource = null) {
if (resource) return (this.received.get(resource) || []).length
let n = 0
for (const caps of this.received.values()) n += caps.length
return n
}
clearReceived () {
const n = this.received.size
this.received.clear()
return n
}
getStats () { getStats () {
return { return {
...this._stats, ...this._stats,
@@ -45,6 +45,15 @@ class HyperP2PCompactCodecBridge extends EventEmitter {
} }
} }
encodeFrameBatch (protocolId, codecId, values) {
if (!Array.isArray(values)) throw new Error('values must be an array')
return values.map((v) => this.encodeFrame(protocolId, codecId, v))
}
registerCodec (codecId, codec) {
return this.registry.registerCodec(codecId, codec)
}
encodeBatch (codecId, values) { encodeBatch (codecId, values) {
if (!Array.isArray(values)) throw new Error('values must be an array') if (!Array.isArray(values)) throw new Error('values must be an array')
return values.map((v) => this.encode(codecId, v)) return values.map((v) => this.encode(codecId, v))
@@ -92,6 +92,20 @@ class HyperP2PSchemaValidator extends EventEmitter {
return s ? { ...s } : null return s ? { ...s } : null
} }
listSchemas () {
return [...this._schemas.keys()]
}
unregister (name) {
return this._schemas.delete(String(name))
}
clearAll () {
const n = this._schemas.size
this._schemas.clear()
return n
}
getStats () { getStats () {
return { ...this._stats, schemas: this._schemas.size, protocol: PROTOCOL } return { ...this._stats, schemas: this._schemas.size, protocol: PROTOCOL }
} }
@@ -81,6 +81,17 @@ class HyperP2PEntropyBeacon extends EventEmitter {
return { pool: b4a.toString(this._pool, 'hex'), contributions: this._contributions } return { pool: b4a.toString(this._pool, 'hex'), contributions: this._contributions }
} }
contributionCount () {
return this._contributions
}
restoreSnapshot (snap) {
if (!snap || !snap.pool) throw new Error('invalid snapshot')
this._pool = b4a.from(snap.pool, 'hex')
this._contributions = snap.contributions || 0
return this.poolHash()
}
async ready () { async ready () {
if (this.swarm || !this.topic) return this if (this.swarm || !this.topic) return this
await initModuleSwarm(this, { await initModuleSwarm(this, {
@@ -44,6 +44,16 @@ class HyperP2PVoidChannel extends EventEmitter {
return this._voids.delete(String(channel)) return this._voids.delete(String(channel))
} }
clearVoids () {
const n = this._voids.size
this._voids.clear()
return n
}
voidCount () {
return this._voids.size
}
subscriberCount () { subscriberCount () {
return this._subs.size return this._subs.size
} }
@@ -114,6 +114,17 @@ class HyperP2PTriePrefix extends EventEmitter {
return matches.reduce((a, b) => (a.length >= b.length ? a : b)) return matches.reduce((a, b) => (a.length >= b.length ? a : b))
} }
wordCount () {
return this._words.size
}
clearAll () {
const n = this._words.size
this._root = new TrieNode()
this._words.clear()
return n
}
getStats () { getStats () {
return { return {
...this._stats, ...this._stats,
@@ -100,6 +100,16 @@ class HyperP2PPeerSelectorStreaming extends EventEmitter {
return this._peers.size return this._peers.size
} }
peerIds () {
return [...this._peers.keys()]
}
clearAll () {
const n = this._peers.size
this._peers.clear()
return n
}
topRegions (limit = 5) { topRegions (limit = 5) {
const counts = new Map() const counts = new Map()
for (const p of this._peers.values()) { for (const p of this._peers.values()) {
@@ -71,6 +71,24 @@ class HyperP2PRetainedMessages extends EventEmitter {
return this._store.has(channel) return this._store.has(channel)
} }
countFor (channel) {
const list = this._store.get(channel)
return list ? list.length : 0
}
pruneOld (maxAgeMs = 86400000) {
const cutoff = Date.now() - maxAgeMs
let n = 0
for (const [, list] of this._store) {
const before = list.length
const kept = list.filter((e) => e.at >= cutoff)
n += before - kept.length
list.length = 0
kept.forEach((e) => list.push(e))
}
return n
}
getStats () { getStats () {
return { return {
...this._stats, ...this._stats,
@@ -97,6 +97,11 @@ class HyperP2PStreamBackpressure extends EventEmitter {
return { high: this.highWaterMark, low: this.lowWaterMark } return { high: this.highWaterMark, low: this.lowWaterMark }
} }
resetStats () {
this._stats = { writes: 0, paused: 0, resumed: 0 }
return this._stats
}
getStats () { getStats () {
return { return {
...this._stats, ...this._stats,
@@ -81,6 +81,16 @@ class HyperP2PStreamTee extends EventEmitter {
return out return out
} }
branchCount () {
return this._branches.size
}
clearAll () {
const n = this._branches.size
this._branches.clear()
return n
}
getStats () { getStats () {
return { ...this._stats, branchCount: this._branches.size, protocol: PROTOCOL } return { ...this._stats, branchCount: this._branches.size, protocol: PROTOCOL }
} }
@@ -82,6 +82,17 @@ class HyperP2PStreamTransform extends EventEmitter {
return n return n
} }
writeBatch (chunks) {
if (!Array.isArray(chunks)) throw new Error('chunks must be an array')
let ok = 0
for (const c of chunks) if (this.write(c)) ok++
return ok
}
hasTransform () {
return typeof this._fn === 'function'
}
getStats () { getStats () {
return { ...this._stats, pending: this._out.length, protocol: PROTOCOL } return { ...this._stats, pending: this._out.length, protocol: PROTOCOL }
} }
@@ -75,6 +75,16 @@ class HyperP2PAnycastSelector extends EventEmitter {
return n return n
} }
clearTags () {
const n = this._tags.size
this._tags.clear()
return n
}
latencyPeerIds () {
return [...this._latency.keys()]
}
getStats () { getStats () {
return { ...this._stats, tags: this._tags.size, latencyPeers: this._latency.size, protocol: PROTOCOL } return { ...this._stats, tags: this._tags.size, latencyPeers: this._latency.size, protocol: PROTOCOL }
} }
@@ -87,6 +87,16 @@ class HyperP2PMultipathFanout extends EventEmitter {
return n return n
} }
clearAll () {
const n = this._pending.size
this._pending.clear()
return n
}
pendingCount () {
return this._pending.size
}
getStats () { getStats () {
return { ...this._stats, pending: this._pending.size, protocol: PROTOCOL } return { ...this._stats, pending: this._pending.size, protocol: PROTOCOL }
} }
@@ -73,6 +73,16 @@ class HyperP2PProtocolHandshake extends EventEmitter {
return n return n
} }
agreedPeerIds () {
return [...this._agreed.keys()]
}
clearAgreed () {
const n = this._agreed.size
this._agreed.clear()
return n
}
getStats () { getStats () {
return { ...this._stats, pending: this._offers.size, agreed: this._agreed.size, protocol: PROTOCOL } return { ...this._stats, pending: this._offers.size, agreed: this._agreed.size, protocol: PROTOCOL }
} }
@@ -57,6 +57,20 @@ class HyperP2PLogGossip extends EventEmitter {
return out return out
} }
setMinLevel (level) {
if (!LEVELS.includes(level)) throw new Error(`unknown level: ${level}`)
this.minLevel = level
return this.minLevel
}
logCount () {
return this._logs.length
}
errors (limit = 50) {
return this.filterByLevel('error', limit)
}
_onGossip (data) { _onGossip (data) {
if (!data || data.type !== 'log' || !data.entry) return if (!data || data.type !== 'log' || !data.entry) return
this._stats.gossipIn++ this._stats.gossipIn++
@@ -77,6 +77,16 @@ class HyperP2PStatsExporter extends EventEmitter {
return this._sources.has(name) return this._sources.has(name)
} }
unregisterAll () {
const n = this._sources.size
this._sources.clear()
return n
}
latestSnapshot () {
return this._history.length ? { ...this._history[this._history.length - 1] } : null
}
getStats () { getStats () {
return { return {
...this._stats, ...this._stats,
@@ -83,6 +83,18 @@ class HyperPearApplinkConfig extends EventEmitter {
return this.load(pkg) return this.load(pkg)
} }
isLoaded () {
return this._config != null
}
hasFlag (name) {
return !!(this._config && Object.prototype.hasOwnProperty.call(this._config.flags, name))
}
flagNames () {
return this._config ? Object.keys(this._config.flags) : []
}
getStats () { getStats () {
return platformStats(this._stats, PROTOCOL, { loaded: !!this._config }) return platformStats(this._stats, PROTOCOL, { loaded: !!this._config })
} }
@@ -83,6 +83,16 @@ class HyperPearDeepLink extends EventEmitter {
return this.peek(link).route return this.peek(link).route
} }
openBatch (links) {
if (!Array.isArray(links)) throw new Error('links must be an array')
return links.map((link) => this.open(link))
}
setDefaultEntry (entry) {
this.defaultEntry = String(entry)
return this.defaultEntry
}
getStats () { getStats () {
return platformStats(this._stats, PROTOCOL, { history: this._history.length }) return platformStats(this._stats, PROTOCOL, { history: this._history.length })
} }
@@ -77,6 +77,15 @@ class HyperPearLinkResolver extends EventEmitter {
return this._cache.size return this._cache.size
} }
parseBatch (links) {
if (!Array.isArray(links)) throw new Error('links must be an array')
return links.map((link) => this.parse(link))
}
invalidate (href) {
return this._cache.delete(String(href))
}
getStats () { getStats () {
return platformStats(this._stats, PROTOCOL, { cached: this._cache.size }) return platformStats(this._stats, PROTOCOL, { cached: this._cache.size })
} }
@@ -63,6 +63,17 @@ class HyperPearRuntimeEmbed extends EventEmitter {
return this._lastPlan ? { ...this._lastPlan } : null return this._lastPlan ? { ...this._lastPlan } : null
} }
hasLastPlan () {
return this._lastPlan != null
}
reset () {
this._opened = false
this._lastPlan = null
this._stats = { opens: 0, runs: 0 }
return true
}
snapshot () { snapshot () {
return { return {
dir: this.dir, dir: this.dir,
@@ -87,6 +87,12 @@ class HyperPearStorageLayout extends EventEmitter {
return this._apps.size return this._apps.size
} }
clearAll () {
const n = this._apps.size
this._apps.clear()
return n
}
getStats () { getStats () {
return platformStats(this._stats, PROTOCOL, { apps: this._apps.size }) return platformStats(this._stats, PROTOCOL, { apps: this._apps.size })
} }
@@ -73,6 +73,15 @@ class HyperP2PRetryPolicy extends EventEmitter {
return n return n
} }
getRoute (id) {
const r = this._routes.get(String(id))
return r ? { ...r } : null
}
routeCount () {
return this._routes.size
}
resetStats () { resetStats () {
this._stats = { routes: this._routes.size, retries: 0, denied: 0 } this._stats = { routes: this._routes.size, retries: 0, denied: 0 }
return this._stats return this._stats
@@ -65,6 +65,14 @@ class HyperP2PStickySession extends EventEmitter {
return n return n
} }
sessionIds () {
return [...this._bindings.keys()]
}
bindingCount () {
return this._bindings.size
}
_onGossip (d) { _onGossip (d) {
if (!d || d.type !== 'sticky-session-sync' || !d.sessionId) return if (!d || d.type !== 'sticky-session-sync' || !d.sessionId) return
this._stats.gossipIn++ this._stats.gossipIn++
@@ -64,6 +64,23 @@ class HyperP2PDriveWatchNotify extends EventEmitter {
return n return n
} }
hasWatch (id) {
return this._watches.has(id)
}
watchCount () {
return this._watches.size
}
notifyBatch (changes) {
if (!Array.isArray(changes)) throw new Error('changes must be an array')
let total = 0
for (const c of changes) {
if (c && c.path) total += this.notifyChange(c.path, c.op || 'update')
}
return total
}
_onGossip (d) { _onGossip (d) {
if (!d || d.type !== 'drive-change') return if (!d || d.type !== 'drive-change') return
this._stats.gossipIn++ this._stats.gossipIn++