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 | (ns ird.p2p.db.impl.orbitdb
"OrbitDB behind ird.p2p.db.core: the keystore seeded with the derived
identity keys, the Identities system, one OrbitDB instance over the Helia
node, and one `events` (append-only log) database opened by name — or by
address, to join a database created elsewhere. Exposed as a plain handle
whose surface is pure data plus core.async channels:
{:address the OrbitDB address (/orbitdb/<manifest hash>)
:name the name the manifest carries
:identity-id the identity's id (hex of the identity key)
:identity-hash the identity's hash (what entries carry as :identity)
:write the database's write list (identity ids)
:add (fn [value] -> channel of the appended entry map / error)
:all (fn [] -> channel of [entry maps], oldest first as OrbitDB traverses)
:get (fn [hash] -> channel of the entry map or nil)
:entry-bytes (fn [hash] -> channel of the stored entry bytes, base64, or nil:
exactly the bytes OrbitDB hashes, signs, stores and sends)
:peers (fn [] -> [peer ids currently synced with])
:events channel of {:type :update :entry <map>} / {:type :join :peer :heads}
/ {:type :leave :peer} / {:type :error :message}
:close (fn [] -> channel)}
An entry map is {:hash :payload <the stored value> :next [hashes]
:clock {:id :time} :identity :key} — payloads are whatever was added
(this library adds sealed records as base64 strings).
Internal interop namespace: js/, dynamic import and property access live
only here."
(: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.db.identity :as identity]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.impl.node :as node]))
(defn- imp [name] (a/promise->chan (js/ird_dyn_import name)))
(defn- named [m k] (unchecked-get m k))
;; ---- the ^js interop seam --------------------------------------------------------------------
(defn- entry->clj [^js e]
(when e
(let [clock (.-clock e)
payload (.-payload e)]
{:hash (.-hash e)
:payload (when payload (.-value payload))
:next (vec (array-seq (or (.-next e) #js [])))
:clock {:id (when clock (.-id clock)) :time (when clock (.-time clock))}
:identity (.-identity e)
:key (.-key e)})))
(defn- ks-add-key [^js ks id private-key-b64]
(.addKey ks id #js {:privateKey (bytes/b64->u8 private-key-b64)}))
(defn- ids-create [^js ids id] (.createIdentity ids #js {:id id}))
(defn- identity-id [^js i] (.-id i))
(defn- identity-hash [^js i] (.-hash i))
(defn- odb-open [^js o address opts] (.open o address opts))
(defn- odb-stop [^js o] (.stop o))
(defn- db-address [^js db] (.-address db))
(defn- db-name [^js db] (.-name db))
(defn- db-add [^js db v] (.add db v))
(defn- db-close [^js db] (.close db))
(defn- db-log [^js db] (.-log db))
(defn- log-values [^js log] (.values log))
(defn- log-get [^js log hash] (.get log hash))
(defn- log-storage-get [^js log hash] (.get (.-storage log) hash))
(defn- db-on [^js db event handler] (.on (.-events db) event handler))
(defn- db-peers [^js db] (mapv str (js/Array.from (.-peers db))))
(defn- ac-write [^js db] (vec (array-seq (or (.-write (.-access db)) #js []))))
(defn- append-error [e]
(let [msg (node/error-message e)]
(if (str/includes? msg "not allowed to write")
(err/append-rejected "OrbitDB refused the entry: this identity is not in the database's write list" {:message msg})
(err/node-error (str "append failed: " msg) {}))))
(defn- <stop [p]
(go (let [v (<! (a/promise->chan p))] (when (node/error? v) (node/log-error "ird.p2p.db: close failed:" (node/error-message v))) nil)))
;; ---- the database ------------------------------------------------------------------------------
(defn create-db
"Seed a keystore with the identity's keys, create the OrbitDB instance
over IPFS (a started Helia node) and open the database. OPTS:
:ipfs the Helia instance (from ird.p2p.db.impl.helia)
:identity the derived identity (ird.p2p.db.identity/from-secret)
:name the name the OrbitDB manifest carries (ird.p2p.db.core passes
the opaque derived manifest name; creates or opens the manifest
for this name and this identity — the same on every peer)
:address open an existing database by its OrbitDB address instead
:path a directory: keystore, heads and index become level-backed
under <path>/orbitdb; without it everything is in memory
Channel of the db handle or a node error."
[{:keys [ipfs identity name address path]}]
(go
(try
(let [m (<? (imp "@orbitdb/core"))
createOrbitDB (named m "createOrbitDB")
Identities (named m "Identities")
KeyStore (named m "KeyStore")
MemoryStorage (named m "MemoryStorage")
LevelStorage (named m "LevelStorage")
storage-for (fn [sub] (if path (LevelStorage #js {:path (str path "/orbitdb/" sub)}) (MemoryStorage)))
keystore (<? (a/promise->chan (KeyStore #js {:storage (<? (a/promise->chan (storage-for "keystore")))})))]
(doseq [{:keys [id private-key]} (identity/keystore-records identity)]
(<? (a/promise->chan (ks-add-key keystore id private-key))))
(let [identities (<? (a/promise->chan (Identities #js {:keystore keystore :ipfs ipfs})))
ident (<? (a/promise->chan (ids-create identities identity/keystore-label)))
_ (when (not= (identity-id ident) (:orbit-id identity))
(throw (err/node-error "OrbitDB derived a different identity than expected" {})))
orbitdb (<? (a/promise->chan (createOrbitDB #js {:ipfs ipfs :identities identities :identity ident
:directory (str (or path ".") "/orbitdb")})))
open-opts (if path
#js {:type "events"}
#js {:type "events"
:headsStorage (<? (a/promise->chan (MemoryStorage)))
:indexStorage (<? (a/promise->chan (MemoryStorage)))})
db (<? (a/promise->chan (odb-open orbitdb (or address name) open-opts)))
log (db-log db)
{:keys [out put! close!]} (a/unbounded-chan)]
(db-on db "update" (fn [e] (put! {:type :update :entry (entry->clj e)})))
(db-on db "join" (fn [peer heads] (put! {:type :join :peer (str peer) :heads (count (array-seq (or heads #js [])))})))
(db-on db "leave" (fn [peer] (put! {:type :leave :peer (str peer)})))
(db-on db "error" (fn [e] (put! {:type :error :message (node/error-message e)})))
{:address (db-address db)
:name (db-name db)
:identity-id (identity-id ident)
:identity-hash (identity-hash ident)
:write (ac-write db)
:add (fn [v]
(go (try (let [h (<? (a/promise->chan (db-add db v)))]
(entry->clj (<? (a/promise->chan (log-get log h)))))
(catch :default e (append-error e)))))
:all (fn []
(go (try (mapv entry->clj (array-seq (<? (a/promise->chan (log-values log)))))
(catch :default e (err/node-error (str "reading the log failed: " (node/error-message e)) {})))))
:get (fn [h]
(go (try (let [e (<! (a/promise->chan (log-get log h)))] (when-not (node/error? e) (entry->clj e)))
(catch :default _ nil))))
:entry-bytes (fn [h]
(go (try (let [u8 (<! (a/promise->chan (log-storage-get log h)))]
(when (and u8 (not (node/error? u8))) (bytes/buffer->b64 u8)))
(catch :default _ nil))))
:peers (fn [] (db-peers db))
:events out
:close (fn []
(go (close!)
(<! (<stop (db-close db)))
(<! (<stop (odb-stop orbitdb)))
nil))}))
(catch :default e
(if (err/db-error? e)
e
(err/node-error (str "OrbitDB failed to open: " (node/error-message e)) {}))))))
|