1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127 | (ns ird.p2p.impl.libp2p
"The real libp2p node behind ird.p2p.transport.libp2p. Creates a Node
libp2p node (tcp + websockets, noise, yamux, identify, gossipsub) via
dynamic import — the libp2p stack is ESM-only, so it cannot be required —
and exposes a plain node handle whose surface is pure data plus a
core.async channel:
{:peer-id <libp2p peer id string>
:hex <32 lowercase hex derived from the peer id>
:multiaddrs (fn [] -> channel of [multiaddr strings])
:dial (fn [multiaddrs] -> channel of true/error)
:subscribe (fn [topic] -> nil)
:unsubscribe (fn [topic] -> nil)
:publish (fn [topic b64] -> channel of true/error)
:messages <channel of {:topic <s> :data <b64>}>
:stop (fn [] -> channel)}
Internal interop namespace: js/, dynamic import and property access live
only here (through the ^js-hinted helpers below). ird.p2p.transport.libp2p
speaks only this handle's data.
Stack: the interface-v3 line — libp2p 3.x with @libp2p/gossipsub 17.x
(the renamed successor of @chainsafe/libp2p-gossipsub), @libp2p/noise,
@libp2p/yamux, tcp, websockets, identify, circuit-relay-v2 — every
package resolving to one @libp2p/interface 3.3.x. The always-on two-node
loopback test (ird.p2p.libp2p-node-test) is the proof that this set
actually delivers frames, and its relay-hop sibling proves the same
exchange through a circuit-relay hop; the adapter is also covered offline
over an injected bus."
(:require [cljs.core.async :as async :refer [go <!]]
[ird.p2p.async :as a]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.node :as node]
[ird.p2p.impl.webcrypto :as webcrypto]))
;; the libp2p stack is ESM-only, so it cannot be required. ird_dyn_import is the
; dynamic-import helper defined in the entry file (see shadow-cljs.edn :prepend);
; a consumer's own :node-script/:node-library/:esm build gets import() natively,
; but may define the same global if it loads this SDK in an import-callback-less
; context.
(defn- imp [name] (a/promise->chan (js/ird_dyn_import name)))
;; ---- the ^js interop seam: every property/method access is a hinted helper,
;; so nothing leaks an :infer-warning and the go block below stays clean ------
(defn- build-node [createLibp2p tcp webSockets circuitRelayTransport circuitRelayServer noise yamux identify gossipsub listen]
(let [services #js {:identify (identify)
:pubsub (gossipsub #js {:allowPublishToZeroTopicPeers true :emitSelf false})}]
;; a relay hop: no default data/duration limit on relayed connections —
;; gossipsub does not run over a limited connection
(when circuitRelayServer
(unchecked-set services "relay" (circuitRelayServer #js {:reservations #js {:applyDefaultLimit false}})))
(createLibp2p
#js {:addresses #js {:listen listen}
:transports #js [(tcp) (webSockets) (circuitRelayTransport)]
:connectionEncrypters #js [(noise)]
:streamMuxers #js [(yamux)]
:services services})))
(defn- node-pubsub [^js n] (.-pubsub (.-services n)))
(defn- node-peer-id [^js n] (.toString (.-peerId n)))
(defn- node-multiaddrs [^js n] (mapv #(.toString %) (array-seq (.getMultiaddrs n))))
(defn- node-dial [^js n multiaddr-fn multiaddrs]
;; libp2p dial needs Multiaddr objects, not strings
(.dial n (clj->js (mapv multiaddr-fn multiaddrs))))
(defn- node-stop [^js n] (.stop n))
(defn- ps-subscribe [^js ps topic] (.subscribe ps topic))
(defn- ps-unsubscribe [^js ps topic] (.unsubscribe ps topic))
(defn- ps-publish [^js ps topic u8] (.publish ps topic u8))
(defn- ev->message [^js ev]
(let [d (.-detail ev)] {:topic (.-topic d) :data (bytes/buffer->b64 (.-data d))}))
(defn- ps-on-message [^js ps handler] (.addEventListener ps "message" handler))
(defn peer-id->hex
"A stable 32-lowercase-hex relay peer id derived from a libp2p peer id
string (SHA-256, first 16 bytes), so K_sig(from) works identically."
[peer-id-string]
(subs (webcrypto/sha256-hex (bytes/utf8->b64 peer-id-string)) 0 32))
(defn create-node
"Create and start a libp2p node. OPTS: :listen (multiaddrs, default an
ephemeral tcp+ws on loopback), :relay-multiaddrs (dialed on start when
present, for circuit-relay reachability), :relay-server? (also serve as a
circuit-relay hop with unlimited relayed connections — the relay-hop
loopback test's third node). A node reserves a slot on the relays it is
connected to when :listen includes \"/p2p-circuit\". Channel of the node
handle or an error value."
[{:keys [listen relay-multiaddrs relay-server?]}]
(go
(try
;; ESM namespace objects are plain JS objects, so pull named exports
;; with unchecked-get (keyword destructuring would yield nil)
(let [createLibp2p (unchecked-get (<! (imp "libp2p")) "createLibp2p")
gossipsub (unchecked-get (<! (imp "@libp2p/gossipsub")) "gossipsub")
noise (unchecked-get (<! (imp "@libp2p/noise")) "noise")
yamux (unchecked-get (<! (imp "@libp2p/yamux")) "yamux")
tcp (unchecked-get (<! (imp "@libp2p/tcp")) "tcp")
webSockets (unchecked-get (<! (imp "@libp2p/websockets")) "webSockets")
identify (unchecked-get (<! (imp "@libp2p/identify")) "identify")
relay-v2 (<! (imp "@libp2p/circuit-relay-v2"))
circuitRelayTransport (unchecked-get relay-v2 "circuitRelayTransport")
circuitRelayServer (when relay-server? (unchecked-get relay-v2 "circuitRelayServer"))
multiaddr (unchecked-get (<! (imp "@multiformats/multiaddr")) "multiaddr")
listen (clj->js (or listen ["/ip4/127.0.0.1/tcp/0" "/ip4/127.0.0.1/tcp/0/ws"]))
n (<! (a/promise->chan (build-node createLibp2p tcp webSockets circuitRelayTransport circuitRelayServer noise yamux identify gossipsub listen)))
pubsub (node-pubsub n)
{:keys [out put! close!]} (a/unbounded-chan)
peer-id (node-peer-id n)]
(ps-on-message pubsub (fn [ev] (put! (ev->message ev))))
(when (seq relay-multiaddrs)
(doseq [ma relay-multiaddrs] (<! (a/promise->chan (node-dial n multiaddr [ma])))))
{:peer-id peer-id
:hex (peer-id->hex peer-id)
:multiaddrs (fn [] (go (node-multiaddrs n)))
:dial (fn [multiaddrs]
(go (try (<! (a/promise->chan (node-dial n multiaddr (vec multiaddrs)))) true
(catch :default e (err/transport-error (str "dial failed: " (node/error-message e)) {})))))
:subscribe (fn [topic] (ps-subscribe pubsub topic) nil)
:unsubscribe (fn [topic] (try (ps-unsubscribe pubsub topic) (catch :default _ nil)) nil)
:publish (fn [topic b64]
(go (try (<! (a/promise->chan (ps-publish pubsub topic (bytes/b64->u8 b64)))) true
(catch :default e (err/transport-error (str "publish failed: " (node/error-message e)) {})))))
:messages out
:stop (fn [] (go (close!) (<! (a/promise->chan (node-stop n))) nil))})
(catch :default e
(err/transport-error (str "libp2p node failed to start: " (node/error-message e)) {})))))
|