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 | (ns ird.p2p.db.core
"An encrypted, append-only, peer-to-peer database for the holders of a
secret — the companion of ird-p2p-cljs rooms: rooms are ephemeral
end-to-end messaging, this is shared history that survives churn. Every
participant who holds the database secret can append records and
replicate them peer to peer, through OrbitDB over Helia, on a private
libp2p swarm: no server, no account, no key ever leaves the peers.
Records are sealed before OrbitDB sees them (ird.p2p.db.crypto): OrbitDB
hashes, signs, stores and syncs ciphertext, giving tamper-evident
ordering on top of confidentiality. Write access is possession of the
secret: the database's single writer identity is derived from it
(ird.p2p.db.identity).
Everything is plain ClojureScript data in and out, and every asynchronous
call returns a core.async channel that delivers exactly one value — the
result or an ex-info error (ird.p2p.db.errors). Long-lived streams
(records, replication events) are channels to drain.
(go (let [db (<? (db/open {:secret secret :app-salt \"myapp/v1\" :name \"chat\"}))]
(<? (db/append db {:text \"salut\"}))
(loop [] (when-let [e (<! (db/entries-stream db))] (println (:payload e)) (recur)))))"
(:require [cljs.core.async :as async :refer [go go-loop <!]]
[ird.p2p.async :as a :refer [go-try <?]]
[ird.p2p.db.crypto :as crypto]
[ird.p2p.db.errors :as err]
[ird.p2p.db.identity :as identity]
[ird.p2p.db.impl.helia :as helia]
[ird.p2p.db.impl.orbitdb :as orbitdb]))
;; ---- secrets -----------------------------------------------------------------------------------
(def random-secret
"`(random-secret)`: a fresh database secret, 32 random bytes as URL-safe
base64 — share it out of band, e.g. in the fragment of an invite link
(ird.p2p/invite-url works for these too)."
crypto/random-secret)
;; ---- records -------------------------------------------------------------------------------------
(defn canonical-order
"ENTRIES sorted by [clock time, hash]: the same total order on every
peer, however each peer's log happened to be traversed. Concurrent
records (same clock time — possible when peers append without having
synced in between) fall back to their hash."
[entries]
(vec (sort-by (juxt #(get-in % [:clock :time]) :hash) entries)))
(defn- ->record
"The public record of a decrypted entry: {:payload :clock :by :hash}."
[entry payload]
{:payload payload
:clock (:clock entry)
:by (:identity entry)
:hash (:hash entry)})
(defn- decrypt-all
"Channel of the records of ENTRIES that decrypt (the rest are dropped
silently), in canonical order."
[dc entries]
(go-loop [es (canonical-order entries) acc []]
(if-let [e (first es)]
(let [payload (<! (crypto/open dc (:payload e)))]
(recur (rest es) (if payload (conj acc (->record e payload)) acc)))
acc)))
(defn- unseen-from
"Walk from ENTRY down its :next links, stopping at hashes already in SEEN
(marking every entry visited): OrbitDB fires one update per received
head, and a head may bring a whole chain of entries fetched from peers
along with it. Channel of the newly seen entries."
[db seen entry]
(go-loop [stack [entry] acc []]
(if (empty? stack)
acc
(let [e (peek stack)
stack (pop stack)]
(if (or (nil? e) (contains? @seen (:hash e)))
(recur stack acc)
(do (swap! seen conj (:hash e))
(let [nexts (loop [hs (:next e) es []]
(if-let [h (first hs)]
(recur (rest hs) (conj es (<! ((:get db) h))))
es))]
(recur (into stack (remove nil? nexts)) (conj acc e)))))))))
;; ---- opening ------------------------------------------------------------------------------------------
(defn open
"Open (creating it if needed) the database NAME of the holders of SECRET.
OPTS:
:secret the database secret, URL-safe base64 of at least 16 bytes — required
:app-salt namespaces every derivation; never reuse another app's salt — required
:name the database name (no `|`, at most 128 characters) — required
:path a directory: records, heads, identity keys and the peer id persist
under it across restarts; without it everything is in memory
:listen multiaddrs to listen on (default loopback tcp + ws, ephemeral ports;
a browser cannot listen and defaults to none)
:dial multiaddrs of peers to dial as soon as the node is up — a peer's
<relay>/p2p-circuit/p2p/<id> address reaches it through its relay
:mdns? discover LAN peers via mDNS and dial each one found (default
false: explicit dial only). Only peers of this library answer;
to meet across machines, :listen on a LAN address
:relay multiaddr string(s) of a circuit relay v2 hop you run: the node
dials it at open, reserves a slot and becomes reachable at
<relay>/p2p-circuit/p2p/<its id> (see `multiaddrs`), and can dial
other peers' /p2p-circuit addresses. A browser peer's only way in
:relay-hop? run the circuit relay hop on this (Node) peer, so peers that
name it in :relay reach each other through it — your own relay,
never a built-in one
:gater a function of one argument, the remote multiaddr string of a peer
connecting TO this node — always ending in /p2p/<its peer id>, a
relayed one carrying /p2p-circuit — returning true to let it in.
Refused peers never get a connection, so nothing replicates
with them unless this node dials them itself: outbound dials
are not gated. Default nil: every inbound connection is allowed
:address join an existing database by its OrbitDB address instead of by
name (advanced; the write list still requires the secret's
identity — with another secret every append is rejected).
:name is still required: it selects the key and the AAD
:node-fn :db-fn injectable node and database constructors (tests)
Channel of the handle or an error (config error, node error, dial error).
Handle keys are public data: :name :app-salt :address :peer-id :identity
(the derived identity's :orbit-id and :signing-key) :manifest-name (the
opaque name the OrbitDB manifest carries instead of :name).
The same namespace runs in a browser (README, \"Browser peers\"): there
the node has no tcp and cannot listen, so :relay is how it is reached
and :dial (a relay's /p2p-circuit address or a websockets one) how it
reaches others; :path names IndexedDB databases."
[{:keys [secret app-salt name path listen dial mdns? relay relay-hop? gater address node-fn db-fn] :as opts}]
(go-try
(let [dc (<? (crypto/db-crypto {:secret secret :app-salt app-salt :name name}))
ident (<? (identity/from-secret {:secret secret :app-salt app-salt :name name}))
node (<? ((or node-fn helia/create-node) {:listen listen :mdns? mdns? :path path :relay relay :relay-hop? relay-hop? :gater gater}))
db (try
(when (seq dial) (<? ((:dial node) dial)))
(<? ((or db-fn orbitdb/create-db) {:ipfs (:ipfs node) :identity ident :name (:manifest-name dc) :address address :path path}))
(catch :default e
(<! ((:stop node)))
(throw e)))
entries (a/unbounded-chan)
peers (a/unbounded-chan)
seen (atom #{})
state (atom :open)
handle {:name name
:app-salt app-salt
:address (:address db)
:manifest-name (:manifest-name dc)
:peer-id (:peer-id node)
:identity (select-keys ident [:orbit-id :signing-key])
:crypto dc
:node node
:db db
:entries (:out entries)
:peers (:out peers)
:seen seen
:state state}
emit! (fn [es]
(go (doseq [r (<! (decrypt-all dc es))]
((:put! entries) r))))]
;; what the log already holds, then whatever arrives
(let [existing (<? ((:all db)))]
(swap! seen into (map :hash existing))
(<! (emit! existing)))
(go-loop []
(if-let [ev (<! (:events db))]
(do (case (:type ev)
:update (<! (emit! (<! (unseen-from db seen (:entry ev)))))
:join ((:put! peers) {:type :peer-joined :peer (:peer ev) :heads (:heads ev)})
:leave ((:put! peers) {:type :peer-left :peer (:peer ev)})
:error ((:put! peers) {:type :sync-error :message (:message ev)})
nil)
(recur))
(do ((:close! entries)) ((:close! peers)))))
handle)))
(defn- check-open [handle]
(when (not= :open @(:state handle)) (throw (err/closed))))
;; ---- using it -------------------------------------------------------------------------------------------
(defn append
"Seal PAYLOAD (a map; ordered JSON, at most 256 KiB serialized) and
append it. Channel of the record {:payload :clock :by :hash}, or an
error: closed, config error (not a map), payload-too-large,
append-rejected, node error. The record also arrives on the entries
stream, like everyone else's."
[handle payload]
(go-try
(check-open handle)
(let [env (<? (crypto/seal (:crypto handle) payload))
entry (<? ((:add (:db handle)) env))]
(->record entry payload))))
(defn entries
"Every record of the database that decrypts, in canonical order (the
same on every peer). Channel of a vector."
[handle]
(go-try
(check-open handle)
(<! (decrypt-all (:crypto handle) (<? ((:all (:db handle))))))))
(defn entries-stream
"The records channel: everything the log held when the database was
opened (in canonical order), then each record as it becomes known —
local appends and replicated ones alike — once; drain it. Records that
do not decrypt never appear. Closes when the handle is closed."
[handle]
(:entries handle))
(defn peers-stream
"Replication events: {:type :peer-joined :peer <id> :heads <n>} once a
peer has exchanged heads with us, {:type :peer-left :peer <id>},
{:type :sync-error :message <s>}. Closes when the handle is closed."
[handle]
(:peers handle))
(defn dial
"Connect to a peer by its multiaddrs (strings). Channel of true or a dial error."
[handle multiaddrs]
(go-try
(check-open handle)
(<? ((:dial (:node handle)) multiaddrs))))
(defn multiaddrs
"This node's listen multiaddrs (strings, with its peer id), to hand to a
peer's `dial`. Channel of a vector."
[handle]
(go-try
(check-open handle)
(<! ((:multiaddrs (:node handle))))))
(defn peers
"The peer ids currently synced with this database."
[handle]
(if (= :open @(:state handle)) ((:peers (:db handle))) []))
(defn address
"The OrbitDB address of the database: derived from the name and the
identity, so every holder of the secret computes the same one."
[handle]
(:address handle))
(defn peer-id
"This node's libp2p peer id."
[handle]
(:peer-id handle))
(defn close
"Close the database, stop the node and close the streams. Channel that
closes when done; a second close is a no-op."
[handle]
(go
(when (compare-and-set! (:state handle) :open :closing)
(<! ((:close (:db handle))))
(<! ((:stop (:node handle))))
(reset! (:state handle) :closed))
nil))
|