const { Readable: e } = require("streamx"),
s = require("hypercore"),
t = require("hypercore-id-encoding"),
r = require("safety-catch"),
o = require("b4a"),
{ StreamsMessage: n, PeerPresenceMessage: i } = require("../lib/messages"),
{ PRESENCE_EVENTS: a } = require("../lib/common"),
c = "keet-call-presence/v3"
function d(e, s) {
return e.audioMuted === s.audioMuted && e.videoMuted === s.videoMuted
}
function l(e) {
let t = s.getProtocolMuxer(e)
return (t || (s.createProtocolStream(e), (t = s.getProtocolMuxer(e))), t)
}
module.exports = class {
constructor(e, s, r) {
;((this.server = e),
(this.swarm = e.swarm),
(this.onerror = e.onerror),
(this.firewall = r),
(this.activeFirewall = null),
(this.callId = s),
(this.closed = !1),
(this.topic = t.decode(s)),
(this._description = { protocol: c, id: this.topic }),
(this._directPresenceIds = new Set()),
(this._relayedPresenceBySwarmId = new Map()),
(this._streamsBySwarmId = new Map()),
(this._localStreams = null),
(this._activated = !1),
(this._protocolsBySwarmId = new Map()),
(this._presenceSubscriptions = []))
for (const e of this.swarm.connections) this.addConnection(e)
this.server.shouldActivate(this.callId) && this.activate()
}
close() {
for (const { connection: e, channel: s } of this._protocolsBySwarmId.values()) {
const t = l(e)
;(t && t.unpair(this._description), s && s.close())
}
;(this._protocolsBySwarmId.clear(), (this.closed = !0))
}
_bumpSubscriptions(e) {
for (const s of this._presenceSubscriptions) s.push(e)
}
addConnection(e) {
const s = l(e),
o = e.remotePublicKey,
n = t.encode(o),
i = { connection: e, remotePublicKey: o, remoteId: n, muxer: s, channel: null, senders: null }
;(this._protocolsBySwarmId.set(n, i),
s.pair(this._description, () => this._createChannel(e)),
this._activated && this._createChannel(e).catch(r))
}
removeConnection(e) {
const s = t.encode(e.remotePublicKey)
this._protocolsBySwarmId.delete(s)
}
activate() {
if (!this._activated) {
this._activated = !0
for (const { connection: e } of this._protocolsBySwarmId.values())
this._createChannel(e).catch(r)
}
}
deactivate() {
if (this._activated) {
;((this._activated = !1), this.setNotPresent())
for (const e of this._protocolsBySwarmId.values())
(e.channel && e.channel.close(), (e.channel = null), (e.senders = null))
}
}
async _createChannel(e) {
const s = t.encode(e.remotePublicKey)
if (
(!this.activeFirewall && this.firewall && (this.activeFirewall = this.firewall(this.callId)),
this.activeFirewall && !(await this.activeFirewall.allow(e.remotePublicKey)))
)
return
const r = this._protocolsBySwarmId.get(s)
if (!r) return
const o = l(e).createChannel({
id: this.topic,
protocol: c,
onopen: () => this._onChannelOpen(r),
onclose: () => this._onChannelClose(r),
messages: [
{ encoding: i, onmessage: (e) => this._onPresenceUpdate(r, !1, e) },
{ encoding: n, onmessage: (e) => this._onStreamsUpdate(r, e) },
],
})
null !== o &&
((r.channel = o),
(r.senders = { updatePresence: r.channel.messages[0], updateStreams: r.channel.messages[1] }),
r.channel.open())
}
_onChannelOpen(e) {
for (const s of this._directPresenceIds)
e.senders.updatePresence.send({ key: t.decode(s), present: !0 })
this._localStreams &&
(e.senders.updatePresence.send({ key: t.decode(this.swarm.keyPair.publicKey), present: !0 }),
e.senders.updateStreams.send(this._localStreams))
}
_onChannelClose(e) {
;(this._onStreamsUpdate(e, []),
this._onPresenceUpdate(e, !0, { key: e.remotePublicKey, present: !1 }),
(e.channel = null),
(e.senders = null))
}
_onPresenceUpdate(e, s, { key: r, present: n }) {
if (o.equals(r, this.swarm.keyPair.publicKey)) return
const i = t.encode(r),
a = e.remoteId,
c = i === a
n ? this._onPeerPresent(a, i, c) : this._onPeerNotPresent(a, i, c, s)
}
_onPeerPresent(e, s, r) {
const o = this._directPresenceIds.has(s),
n = this._relayedPresenceBySwarmId.has(s)
if (r) (this._directPresenceIds.add(e), this._relayedPresenceBySwarmId.delete(e))
else {
let t = this._relayedPresenceBySwarmId.get(s)
;(t || ((t = new Set()), this._relayedPresenceBySwarmId.set(s, t)), t.add(e))
}
if (r && !o)
for (const s of this._protocolsBySwarmId.values())
s.senders &&
s.remoteId !== e &&
s.senders.updatePresence.send({ key: t.decode(e), present: !0 })
o || n || this._bumpSubscriptions({ type: a.PEER_PRESENT, swarmId: s })
}
_removeRelayedPeer(e, s) {
const t = this._relayedPresenceBySwarmId.get(s)
return (
!!t &&
(t.delete(e),
!t.size && (this._relayedPresenceBySwarmId.delete(s), !this._directPresenceIds.has(s)))
)
}
_onPeerNotPresent(e, s, r, o) {
const n = new Set()
if (r) {
if (
(n.add(e), this._directPresenceIds.delete(e), this._relayedPresenceBySwarmId.delete(e), o)
)
for (const s of this._relayedPresenceBySwarmId.keys())
this._removeRelayedPeer(e, s) && n.add(s)
for (const s of this._protocolsBySwarmId.values())
s.senders &&
s.remoteId !== e &&
s.senders.updatePresence.send({ key: t.decode(e), present: !1 })
} else this._removeRelayedPeer(e, s) && n.add(s)
const i = 0 === this._directPresenceIds.size && 0 === this._relayedPresenceBySwarmId.size
!this._activated &&
this.firewall &&
this._firewallSubscription &&
i &&
(this._firewallSubscription.destroy(), (this._firewallSubscription = null))
for (const e of n) this._bumpSubscriptions({ type: a.PEER_NOT_PRESENT, swarmId: e })
}
_onStreamsUpdate(e, s) {
const t = this._streamsBySwarmId.get(e.remoteId),
r = new Map()
for (const e of s) r.set(e.id, e)
if (t)
for (const [s, o] of t)
r.has(s) ||
this._bumpSubscriptions({
type: a.STREAM_UNAVAILABLE,
swarmId: e.remoteId,
streamId: o.id,
streamName: o.name,
})
for (const [s, o] of r)
(t && t.has(s)) ||
this._bumpSubscriptions({
type: a.STREAM_AVAILABLE,
swarmId: e.remoteId,
streamId: o.id,
streamName: o.name,
})
for (const [s, o] of r) {
const r = t && t.get(s)
;(r && d(r, o)) ||
this._bumpSubscriptions({
type: a.STREAM_STATUS_CHANGED,
swarmId: e.remoteId,
streamId: o.id,
audioMuted: o.audioMuted,
videoMuted: o.videoMuted,
})
}
r.size ? this._streamsBySwarmId.set(e.remoteId, r) : this._streamsBySwarmId.delete(e.remoteId)
}
setPresence(e) {
if (!this._localStreams) {
this._localStreams = []
for (const e of this._protocolsBySwarmId.values())
e.senders &&
e.senders.updatePresence.send({ key: this.swarm.keyPair.publicKey, present: !0 })
}
for (const s of this._protocolsBySwarmId.values())
s.senders && s.senders.updateStreams.send(e.streams)
this._localStreams = e.streams
}
setNotPresent() {
if (this._localStreams) {
this._localStreams = null
for (const e of this._protocolsBySwarmId.values())
e.senders &&
e.senders.updatePresence.send({ key: this.swarm.keyPair.publicKey, present: !1 })
}
for (const e of this._protocolsBySwarmId.values()) e.senders && e.senders.updateStreams.send([])
}
subscribePresence() {
const s = new e({
destroy: (e) => {
const t = this._presenceSubscriptions.indexOf(s)
;(-1 !== t && this._presenceSubscriptions.splice(t, 1), e(null))
},
})
this._presenceSubscriptions.push(s)
const t = this.getPresentPeerEvents()
for (const e of t) s.push(e)
return s
}
getPresentPeerEvents() {
const e = []
for (const s of this._directPresenceIds) e.push({ type: a.PEER_PRESENT, swarmId: s })
for (const s of this._relayedPresenceBySwarmId.keys())
this._directPresenceIds.has(s) || e.push({ type: a.PEER_PRESENT, swarmId: s })
for (const [s, t] of this._streamsBySwarmId)
for (const [r, { name: o, audioMuted: n, videoMuted: i }] of t)
(e.push({ type: a.STREAM_AVAILABLE, streamName: o, swarmId: s, streamId: r }),
e.push({
type: a.STREAM_STATUS_CHANGED,
swarmId: s,
streamId: r,
audioMuted: n,
videoMuted: i,
}))
return e
}
}