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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285 | (ns ird.p2p.db.impl.helia
"The Helia node behind ird.p2p.db.core: a Helia (IPFS) node over a
PRIVATE libp2p swarm — loopback/LAN tcp and websockets, noise, yamux,
identify and gossipsub, and nothing that reaches the public network: no
bootstrap list, no DHT, no delegated HTTP routing, no public gateways, no
relay discovery, no mDNS unless asked for. Peers meet by explicit dial (or
mDNS on a LAN when :mdns? is set), or through a circuit relay v2 hop the
user runs and names (:relay; a Node peer runs the hop with :relay-hop?).
Created via dynamic import — helia and the libp2p packages are ESM-only,
so they cannot be required — and exposed as a plain node handle whose
surface is pure data plus core.async channels:
{:peer-id <libp2p peer id string>
:ipfs the Helia instance (opaque; handed to ird.p2p.db.impl.orbitdb)
:platform :node or :browser
:multiaddrs (fn [] -> channel of [multiaddr strings])
:peers (fn [] -> [connected peer id strings])
:connections (fn [] -> [{:peer <id> :addr <multiaddr string> :limited? <bool>}]:
the open connections and the address each was made on
(a relayed one carries /p2p-circuit/ and, when the hop
applies limits, :limited? true)
:dial (fn [multiaddrs] -> channel of true / dial error)
:sniff! (fn [] -> channel of {:topic <s> :data <b64>}: every raw
pubsub message this node receives, for tests that assert
what crosses the wire; registers on first call)
:routers :block-brokers :services the names of what the node runs
:stop (fn [] -> channel)}
In a browser (no Node `process`) the same namespace builds the browser
variant: websockets and the circuit relay transport only (a browser dials
out, it cannot listen, so it is reachable only through a relay it
reserves on), no tcp, no mDNS, and the loopback/LAN dial restriction of
libp2p's browser connection gater lifted, since every address dialed here
is the user's own configuration. `:path` then names IndexedDB databases
(Level's browser backend) instead of directories. The packages arrive
through the same `ird_dyn_import` helper, which a browser page points at
a pre-bundled chunk (README, \"Browser peers\").
Internal interop namespace: js/, dynamic import and property access live
only here (through the ^js-hinted helpers below)."
(:require [clojure.string :as str]
[cljs.core.async :as async :refer [go <!]]
[ird.p2p.async :as a :refer [<?]]
[ird.p2p.db.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.node :as node]))
;; helia and libp2p are ESM-only, so they 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 library
;; in an import-callback-less context.
(defn- imp [name] (a/promise->chan (js/ird_dyn_import name)))
;; ESM namespace objects are plain JS objects, so pull named exports with
;; unchecked-get (keyword destructuring would yield nil)
(defn- named [m k] (unchecked-get m k))
(def default-listen
"Loopback tcp and websockets on ephemeral ports: private by default; pass
:listen to bind a LAN interface or fixed ports. A browser listens on
nothing but the relay it reserves on."
["/ip4/127.0.0.1/tcp/0" "/ip4/127.0.0.1/tcp/0/ws"])
(def platform
"`:node` when a Node `process` is present, else `:browser`."
(if (and (exists? js/process) (some? (.-versions js/process)) (some? (.-node (.-versions js/process))))
:node
:browser))
(defn- browser? [] (= :browser platform))
(defn relay-listen-addrs
"The /p2p-circuit listen addresses for RELAYS (multiaddr strings of hops
this node reserves a slot on): libp2p dials each relay at start and the
node becomes reachable at <relay>/p2p-circuit/p2p/<peer id>."
[relays]
(mapv #(str % "/p2p-circuit") relays))
;; mDNS announces and queries under this service tag instead of libp2p's
;; default _p2p._udp.local, so :mdns? finds peers of this library only: the
;; node neither dials nor answers the other libp2p nodes (an IPFS daemon,
;; say) a LAN may hold
(def mdns-service-tag "_ird-p2p-db._udp.local")
(defn- relays-of [relay] (cond (nil? relay) [] (string? relay) [relay] :else (vec relay)))
;; ---- the ^js interop seam --------------------------------------------------------------------
(defn- mdns-discovery
"The mDNS peer discovery factory under `mdns-service-tag`, calling
(ON-PEER id multiaddrs) — a peer id object and the Multiaddr objects it
announced — each time a peer answers a query: libp2p only records a
discovered peer, it never dials one."
[mdns on-peer]
(let [factory (mdns #js {:serviceTag mdns-service-tag})]
(fn [components]
(let [^js discovery (factory components)]
(.addEventListener discovery "peer"
(fn [^js ev] (let [^js d (.-detail ev)] (on-peer (.-id d) (array-seq (.-multiaddrs d))))))
discovery))))
(defn- inbound-gater
"libp2p's denyInboundEncryptedConnection over GATER, a function of the
remote multiaddr string returning true to allow: the hook runs once noise
has established who is calling and before anything else is spoken, so
the string always ends in /p2p/<the remote peer id>. A gater that throws
denies."
[gater]
(fn [^js peer-id ^js ma-conn]
(let [addr (.toString (.-remoteAddr ma-conn))
suffix (str "/p2p/" (.toString peer-id))
addr (if (str/ends-with? addr suffix) addr (str addr suffix))]
(not (try (gater addr) (catch :default _ false))))))
(defn- libp2p-config
"The libp2p init: transports tcp + websockets (Node) or websockets alone
(browser), plus the circuit relay transport when this node uses a relay
(always in the browser); the hop service when :relay-hop?. Without
relay options and without a :gater on Node this is exactly the
private-swarm config it always was."
[{:keys [tcp webSockets circuitRelayTransport circuitRelayServer noise yamux identify gossipsub mdns on-mdns-peer listen gater]}]
(let [transports (cond-> []
tcp (conj (tcp))
true (conj (webSockets))
circuitRelayTransport (conj (circuitRelayTransport)))
services #js {:identify (identify)
:pubsub (gossipsub #js {:allowPublishToZeroTopicPeers true :emitSelf false})}
cfg #js {:addresses #js {:listen (clj->js listen)}
:transports (to-array transports)
:connectionEncrypters #js [(noise)]
:streamMuxers #js [(yamux)]
:peerDiscovery (if mdns #js [(mdns-discovery mdns on-mdns-peer)] #js [])
:services services}]
(when circuitRelayServer
;; a user-run hop: no per-connection duration/data limits, the relay is theirs
(unchecked-set services "relay" (circuitRelayServer #js {:reservations #js {:applyDefaultLimit false}})))
(when (or (browser?) gater)
(let [gate #js {}]
;; libp2p's browser gater refuses loopback/LAN dials by default (a web page
;; must not probe its host's network); here every dialed address is the
;; user's own relay or peer, so dial exactly what is configured
(when (browser?) (unchecked-set gate "denyDialMultiaddr" (fn [_] false)))
;; :gater gates inbound connections only, on both platforms
(when gater (unchecked-set gate "denyInboundEncryptedConnection" (inbound-gater gater)))
(unchecked-set cfg "connectionGater" gate)))
cfg))
(defn- helia-init [libp2p blockstore datastore]
;; :http with no delegated routers and no recursive gateways leaves the
;; libp2p router as the only router, so the trustless-gateway block broker
;; (which discovers gateways through the routers) has nowhere to go
(let [init #js {:libp2p libp2p
:http #js {:delegatedRouters #js [] :recursiveGateways #js []}}]
(when blockstore (unchecked-set init "blockstore" blockstore))
(when datastore (unchecked-set init "datastore" datastore))
init))
(defn- thenable? [^js v] (and (some? v) (fn? (unchecked-get v "then"))))
(defn- h-status [^js h] (.-status h))
(defn- h-start [^js h] (.start h))
(defn- h-stop [^js h] (.stop h))
(defn- h-libp2p [^js h] (.-libp2p h))
(defn- h-router-names [^js h] (mapv (fn [^js r] (.-name r)) (array-seq (.-routers (.-routing h)))))
(defn- h-broker-names [^js h] (mapv (fn [^js b] (.-name b)) (array-seq (.-blockBrokers h))))
(defn- lp-peer-id [^js lp] (.toString (.-peerId lp)))
(defn- lp-multiaddrs [^js lp] (mapv #(.toString %) (array-seq (.getMultiaddrs lp))))
(defn- lp-peers [^js lp] (mapv #(.toString %) (array-seq (.getPeers lp))))
(defn- lp-connections [^js lp]
(mapv (fn [^js c] {:peer (.toString (.-remotePeer c))
:addr (.toString (.-remoteAddr c))
:limited? (some? (.-limits c))})
(array-seq (.getConnections lp))))
(defn- lp-services [^js lp] (vec (js-keys (.-services lp))))
(defn- lp-pubsub [^js lp] (.-pubsub (.-services lp)))
(defn- lp-dial [^js lp multiaddr-fn addrs]
;; libp2p dial takes Multiaddr objects, not strings
(.dial lp (clj->js (mapv multiaddr-fn addrs))))
(defn- lp-dial-discovered
;; the announced addresses carry no peer id: put it back, so that noise
;; verifies who answers. One that cannot be reached is simply not a peer
[^js lp ^js id addrs]
(when (zero? (alength (.getConnections lp id)))
(let [suffix (str "/p2p/" (.toString id))]
(.catch (.dial lp (to-array (map (fn [^js ma] (.encapsulate ma suffix)) addrs))) (fn [_] nil)))))
(defn- ps-on-message [^js ps handler] (.addEventListener ps "message" handler))
(defn- ev->message [^js ev]
(let [d (.-detail ev)] {:topic (.-topic d) :data (bytes/buffer->b64 (.-data d))}))
(defn- store-close [^js s] (.close s))
(defn- level-store [Ctor path] (new Ctor path))
(defn- <stop [p]
;; a stop that fails must not keep a close from finishing
(go (let [v (<! (a/promise->chan p))] (when (node/error? v) (node/log-error "ird.p2p.db: stop failed:" (node/error-message v))) nil)))
;; ---- the node --------------------------------------------------------------------------------
(defn create-node
"Create and start a Helia node on a private swarm. OPTS:
:listen multiaddr strings to listen on (default `default-listen` on
Node; nothing in a browser, which cannot listen)
:mdns? discover LAN peers via mDNS and dial each one found, again
whenever it answers while not connected (default false:
explicit dial only; Node only). Announced and queried under
`mdns-service-tag`, so only peers of this library meet
:path a directory: the blockstore and datastore become level-backed
under <path>/helia (same peer id across restarts); without it
everything is in memory. In a browser Level is IndexedDB, so
:path names the databases instead of a directory
:relay multiaddr string (or strings) of a circuit relay v2 hop the
user runs: adds the circuit relay transport, dials the relay
at start and reserves a slot, so the node listens at
<relay>/p2p-circuit/p2p/<peer id> as well and can dial other
peers' /p2p-circuit addresses. Unreachable relay = node error
:relay-hop? run the circuit relay v2 hop service on this node so that
other peers can reserve on it and reach each other through it
(Node only; no limits applied to relayed connections)
:gater (fn [remote-multiaddr-string]) -> true to allow: gates INBOUND
connections, asked once noise has identified the caller (the
string ends in /p2p/<its peer id>) and before any protocol
runs; false, nil or a throw refuses the connection. Outbound
dials are never gated. Default nil: everyone may connect
Channel of the node handle or a node error."
[{:keys [listen mdns? path relay relay-hop? gater]}]
(go
(try
(let [relays (relays-of relay)
_ (when (and (browser?) mdns?) (throw (err/node-error "mDNS is not available in a browser" {})))
_ (when (and (browser?) relay-hop?) (throw (err/node-error "a browser cannot run a circuit relay hop" {})))
createHelia (named (<? (imp "helia")) "createHelia")
tcp (when-not (browser?) (named (<? (imp "@libp2p/tcp")) "tcp"))
webSockets (named (<? (imp "@libp2p/websockets")) "webSockets")
circuit (when (or (browser?) (seq relays) relay-hop?) (<? (imp "@libp2p/circuit-relay-v2")))
circuitRelayTransport (when (or (browser?) (seq relays)) (named circuit "circuitRelayTransport"))
circuitRelayServer (when relay-hop? (named circuit "circuitRelayServer"))
noise (named (<? (imp "@chainsafe/libp2p-noise")) "noise")
yamux (named (<? (imp "@chainsafe/libp2p-yamux")) "yamux")
identify (named (<? (imp "@libp2p/identify")) "identify")
gossipsub (named (<? (imp "@libp2p/gossipsub")) "gossipsub")
multiaddr (named (<? (imp "@multiformats/multiaddr")) "multiaddr")
mdns (when mdns? (named (<? (imp "@libp2p/mdns")) "mdns"))
;; answers can arrive while the node is still starting: hold them until it is up
found (atom {:lp nil :early []})
on-mdns-peer (fn [id addrs]
(if-let [lp (:lp @found)]
(lp-dial-discovered lp id addrs)
(swap! found update :early conj [id addrs])))
blockstore (when path (level-store (named (<? (imp "blockstore-level")) "LevelBlockstore") (str path "/helia/blocks")))
datastore (when path (level-store (named (<? (imp "datastore-level")) "LevelDatastore") (str path "/helia/data")))
listen (into (vec (or (seq listen) (if (browser?) [] default-listen))) (relay-listen-addrs relays))
libp2p (libp2p-config {:tcp tcp :webSockets webSockets :circuitRelayTransport circuitRelayTransport
:circuitRelayServer circuitRelayServer :noise noise :yamux yamux :identify identify
:gossipsub gossipsub :mdns mdns :on-mdns-peer on-mdns-peer :listen listen :gater gater})
created (createHelia (helia-init libp2p blockstore datastore))
helia (if (thenable? created) (<? (a/promise->chan created)) created)
_ (when (not= "started" (h-status helia)) (<? (a/promise->chan (h-start helia))))
lp (h-libp2p helia)
_ (doseq [[id addrs] (:early (swap! found assoc :lp lp))] (lp-dial-discovered lp id addrs))
sniff (atom nil)]
{:peer-id (lp-peer-id lp)
:ipfs helia
:platform platform
:multiaddrs (fn [] (go (lp-multiaddrs lp)))
:peers (fn [] (lp-peers lp))
:connections (fn [] (lp-connections lp))
:dial (fn [addrs]
(go (try (<? (a/promise->chan (lp-dial lp multiaddr (vec addrs)))) true
(catch :default e (err/dial-error (str "dial failed: " (node/error-message e)) {:multiaddrs (vec addrs)})))))
:sniff! (fn []
(or @sniff
(let [{:keys [out put!]} (a/unbounded-chan)]
(ps-on-message (lp-pubsub lp) (fn [ev] (put! (ev->message ev))))
(reset! sniff out))))
:routers (h-router-names helia)
:block-brokers (h-broker-names helia)
:services (lp-services lp)
:stop (fn []
(go
(<! (<stop (h-stop helia)))
;; Helia does not close level stores it was handed: release their locks
(when blockstore (<! (<stop (store-close blockstore))))
(when datastore (<! (<stop (store-close datastore))))
nil))})
(catch :default e
(err/node-error (str "Helia node failed to start: " (node/error-message e)) {})))))
|