const { getEncoding: e } = require("./spec/hyperschema"),
s = require("hypercore-crypto"),
n = require("hyperdht"),
t = require("ready-resource"),
i = require("b4a"),
o = require("protomux"),
r = require("compact-encoding"),
c = require("hyperdb"),
[h] = s.namespace("hyperdiscovery", 1),
u = 6e4,
a = e("@hyperdiscovery/announce"),
l = e("@hyperdiscovery/announce-to-swarm"),
d = e("@hyperdiscovery/subscribe-to-swarm"),
y = e("@hyperdiscovery/unsubscribe-to-swarm"),
m = e("@hyperdiscovery/bump-from-swarm")
class p extends t {
constructor(e, { bootstrap: s, dht: t = new n({ bootstrap: s }) } = {}) {
;(super(),
(this.dht = t),
(this.db = c.rocks(e, require("./spec/hyperdb"))),
(this.subs = new Map()),
(this._flushTimeout = null),
(this._flushing = null),
(this._server = null),
(this._manifest = null),
this.ready().catch(w))
}
get publicKey() {
return this._server && this._server.address().publicKey
}
async _open() {
;(await this._init(),
this._flushBackground().catch(w),
(this._server = this.dht.createServer()),
this._server.on("connection", this.addStream.bind(this)),
await this._server.listen(s.keyPair(this._manifest.seed)))
}
async _init() {
;((this._manifest = await this.db.get("@hyperdiscovery/manifest")),
this._manifest ||
((this._manifest = { version: 0, seed: s.randomBytes(32) }),
await this.db.insert("@hyperdiscovery/manifest", this._manifest),
await this.db.flush()))
}
async _close() {
;(this._flushTimeout && clearTimeout(this._flushTimeout),
this._flushing && (await this._flushing),
(this._flushTimeout = null),
this._server && (await this._server.close()),
await this.db.close())
}
async _flushBackground() {
for (; !this.closing; ) {
if (
(await new Promise((e) => {
this._flushTimeout = setTimeout(e, 12e4)
}),
(this._flushTimeout = null),
this.closing)
)
return
;((this._flushing = this.db.flush()), await this._flushing, (this._flushing = null))
}
}
_addSub(e, s, n) {
let t = this.subs.get(s)
;(t || ((t = new Map()), this.subs.set(s, t)), t.set(e, n))
}
_removeSub(e, s) {
const n = this.subs.get(s)
n && (n.delete(e), n.size || this.subs.delete(s))
}
addStream(e) {
const s = b(e),
n = this,
t = new Set()
function o() {
const e = s.createChannel({
protocol: "hyperdiscovery",
messages: [
{
encoding: d,
onmessage: async function (s) {
const o = i.toString(s.publicKey, "hex")
;(t.add(o), n._addSub(e, o, s.since))
const r = await n.lookup(s.publicKey)
r && r.bumped >= s.since && e.messages[3].send(r)
},
},
{
encoding: y,
onmessage: function (s) {
const o = i.toString(s.publicKey, "hex")
;(t.delete(o), n._removeSub(e, o))
},
},
{
encoding: l,
onmessage: function (s) {
return n.announce(s, e)
},
},
{ encoding: m, onmessage: g },
],
onclose() {
for (const s of t) n._removeSub(e, s)
},
})
e && e.open()
}
;(s.pair({ protocol: "hyperdiscovery" }, o), o())
}
async lookup(e) {
return await this.db.get("@hyperdiscovery/swarms", { publicKey: e })
}
async announce(e, n) {
const t = { buffer: null, start: 0, end: 0 }
if (
(r.fixed32.preencode(t, h),
a.preencode(t, e.announce),
(t.buffer = i.allocUnsafe(t.end)),
r.fixed32.encode(t, h),
a.encode(t, e.announce),
Date.now() + u > e.announce.bumped)
)
return !1
if (!s.verify(t.buffer, e.signature, e.publicKey)) return !1
const o = await this.db.get("@hyperdiscovery/swarms", e.publicKey)
if (o && o.bumped >= e.announce.bump) return !1
const c = {
publicKey: e.publicKey,
bumped: e.announce.bump,
updated: Date.now(),
signature: e.signature,
}
await this.db.insert("@hyperdiscovery/swarms", c)
const l = this.subs.get(i.toString(e.publicKey, "hex"))
if (l) for (const [e, s] of l) e !== n && (c.bumped < s || e.messages[3].send(c))
return !0
}
}
class f extends t {
constructor(e, { bootstrap: s, dht: t = new n({ bootstrap: s }) } = {}) {
;(super(),
(this.remotePublicKey = e),
(this.dht = t),
(this.connection = null),
(this.muxer = null),
(this.channel = null),
(this.suspended = !1),
(this.connecting = !1),
(this._retryTimeout = null),
(this._retryResolve = null),
(this._shouldReconnect = !1))
}
async suspend() {
;((this.suspended = !0),
this._clearRetry(),
this.connection &&
(this.connection.destroy(),
(this.connecting = null),
(this.channel = null),
(this.muxer = null),
this.connection.destroyed || (await new Promise((e) => this.connection.on("close", e)))))
}
resume() {
;((this.suspended = !1), this._shouldReconnect && !this.closing && this._reconnect())
}
connect() {
this.channel || this._reconnect()
}
async _reconnect() {
if (this.connecting) return this.channel
;((this.connecting = !0), (this._shouldReconnect = !0))
let e = 0
for (; !this.closing && !this.suspended; ) {
const s = (this.connection = this.dht.connect(this.remotePublicKey))
if (
((this.muxer = b(this.connection)),
(this.channel = this.muxer.createChannel({
userData: this,
protocol: "hyperdiscovery",
messages: [
{ encoding: d, onmessage: g },
{ encoding: y, onmessage: g },
{ encoding: l, onmessage: g },
{ encoding: m, onmessage: _ },
],
})),
!this.channel)
)
break
if (
(this.channel.open(),
await new Promise((e) => {
function n() {
;(s.off("connect", n), s.off("error", n), e())
}
;(s.on("connect", n), s.on("error", n))
}),
this.closing || this.suspended)
) {
;(s.on("error", w), s.destroy())
break
}
if (!s.destroyed && !s.destroying) {
s.on("close", this._reconnect.bind(this))
break
}
;(e++,
await new Promise((s) => {
;((this._retryResolve = s), (this._retryTimeout = setTimeout(s, e < 3 ? 5e3 : 15e3)))
}),
(this._retryResolve = null),
(this._retryTimeout = null))
}
return (
(this.connecting = !1),
this.closing || this.suspended || this.emit("connect"),
this.channel
)
}
_close() {
;(this.connection && this.connection.destroy(), this._clearRetry())
}
_clearRetry() {
this._retryTimeout &&
(clearTimeout(this._retryTimeout),
this._retryResolve(),
(this._retryResolve = null),
(this._retryTimeout = null))
}
_onbump(e) {
this.emit("announce", e)
}
_getChannel() {
return (this.channel || this._reconnect(), this.channel)
}
cork() {
const e = this._getChannel()
e && e.cork()
}
uncork() {
const e = this._getChannel()
e && e.uncork()
}
subscribe(e, { since: s = 0 } = {}) {
const n = this._getChannel()
n && n.messages[0].send({ publicKey: e, since: s })
}
unsubscribe(e) {
const s = this._getChannel()
s && s.messages[1].send({ publicKey: e })
}
async announce(e, { bump: n = Date.now() } = {}) {
const t = this._getChannel()
if (!t) return
const o = { bump: n },
c = { buffer: null, start: 0, end: 0 }
;(r.fixed32.preencode(c, h),
a.preencode(c, o),
(c.buffer = i.allocUnsafe(c.end)),
r.fixed32.encode(c, h),
a.encode(c, o))
const u = s.sign(c.buffer, e.secretKey),
l = { publicKey: e.publicKey, announce: o, signature: u }
t.messages[2].send(l)
}
}
function b(e) {
if (o.isProtomux(e)) return e
if (e.noiseStream.userData) return e.noiseStream.userData
const s = o.from(e.noiseStream)
return ((e.noiseStream.userData = s), s)
}
function _(e, s) {
s.userData._onbump(e)
}
function g() {
throw new Error("Method not supported")
}
function w() {}
module.exports = { HyperDiscovery: p, HyperDiscoveryClient: f }