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 | ;; ported-from: src/lbgossip.ts @ v1.2.0
;;
;; All-time leaderboard convergence: hubs (and anyone else) gossip verified
;; receipts on one public topic per game. A peer asks with `lb-want`; peers
;; holding receipts answer with a small batch of their newest; every receiver
;; re-verifies from scratch before folding. Receipts are public, co-signed
;; data — plaintext on purpose. Eventually-consistent and partial by design:
;; there is no server that knows the whole board.
(ns ardegazu.social.lbgossip
(:require [ardegazu.id.crypto :as id-crypto]
[ardegazu.social.receipts :as receipts]
[shadow.cljs.modern :refer (js-await)]))
(def ^:private WANT-INTERVAL-MS 300000)
(def ^:private ANSWER-COOLDOWN-MS 120000)
(def ^:private BATCH 20)
(def ^:private td (js/TextDecoder.))
(defn- fold-recs [recs i opts]
(if (>= i (.-length ^js recs))
(js/Promise.resolve nil)
(js-await [v (receipts/verify-receipt (aget recs i) (unchecked-get opts "verify"))]
(if (nil? v)
(fold-recs recs (inc i) opts)
(js-await [added (.add ^js (unchecked-get opts "receipts") (unchecked-get v "rec"))]
(when added
(when-let [cb (unchecked-get opts "onNew")]
(cb (unchecked-get v "rec"))))
(fold-recs recs (inc i) opts))))))
(defn- subscribe-topics [games i topics]
(if (>= i (.-length ^js games))
(js/Promise.resolve topics)
(let [g (aget games i)]
(js-await [t (receipts/lb-topic g)]
(.set ^js topics t g)
(subscribe-topics games (inc i) topics)))))
(defn- on-topic-message [opts topics last-answer evt]
(let [detail (unchecked-get evt "detail")
topic (unchecked-get detail "topic")
g (.get ^js topics topic)]
(when-not (identical? g js/undefined)
(-> (js/Promise.resolve nil)
(.then
(fn [_]
(let [msg (try (js/JSON.parse (.decode td (unchecked-get detail "data")))
(catch :default _ nil))]
(when (some? msg)
(when (identical? (unchecked-get msg "g") g)
(cond
(identical? (unchecked-get msg "k") "lb-want")
(let [now (js/Date.now)
last (let [x (.get ^js last-answer g)]
(if (identical? x js/undefined) 0 x))]
(when-not (< (- now last) ANSWER-COOLDOWN-MS)
(let [mine (-> (.all ^js (unchecked-get opts "receipts"))
(.filter (fn [r] (identical? (unchecked-get r "g") g)))
(.sort (fn [a b] (- (unchecked-get b "ts") (unchecked-get a "ts"))))
(.slice 0 BATCH))]
(when (pos? (.-length mine))
(.set ^js last-answer g now)
(let [^js pubsub (unchecked-get opts "pubsub")]
(-> (.publish pubsub topic
(id-crypto/utf8 (js/JSON.stringify
(js-obj "k" "lb-recs" "g" g "recs" mine))))
(.catch (fn [_] nil)))))))) ; nobody listening
(and (identical? (unchecked-get msg "k") "lb-recs")
(js/Array.isArray (unchecked-get msg "recs")))
(fold-recs (.slice ^js (unchecked-get msg "recs") 0 BATCH) 0 opts)
:else nil))))))))))
(defn- want-loop [opts entries i]
(if (>= i (.-length ^js entries))
(js/Promise.resolve nil)
(let [e (aget entries i)
t (aget e 0)
g (aget e 1)]
(-> (js/Promise.resolve nil)
(.then (fn [_]
(let [^js pubsub (unchecked-get opts "pubsub")]
(.publish pubsub t
(id-crypto/utf8 (js/JSON.stringify (js-obj "k" "lb-want" "g" g)))))))
(.catch (fn [_] nil)) ; no peers on this topic yet
(.then (fn [_] (want-loop opts entries (inc i))))))))
(defn- want-all [opts topics]
(when (identical? (.-visibilityState js/document) "visible")
(want-loop opts (js/Array.from (.entries ^js topics)) 0)))
(defn start-lb-gossip [opts]
(js-await [topics (subscribe-topics (unchecked-get opts "games") 0 (js/Map.))]
(let [pubsub (unchecked-get opts "pubsub")
last-answer (js/Map.)
topic-keys (js/Array.from (.keys topics))]
(dotimes [i (.-length topic-keys)]
(.subscribe ^js pubsub (aget topic-keys i)))
(.addEventListener ^js pubsub "message"
(fn [evt] (on-topic-message opts topics last-answer evt)))
;; give the mesh a moment to form before the first ask
(let [first-t (js/window.setTimeout (fn [] (want-all opts topics)) 8000)
timer (js/window.setInterval (fn [] (want-all opts topics)) WANT-INTERVAL-MS)]
(js-obj "stop"
(fn []
(js/clearTimeout first-t)
(js/clearInterval timer)
(let [ks (js/Array.from (.keys topics))]
(dotimes [i (.-length ks)]
(.unsubscribe ^js pubsub (aget ks i))))
js/undefined))))))
|