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 | (ns ird.p2p.mailbox-test
"The mailbox against scripted :http-fn transports: URL shape, bearer
tokens, the open mint's caching and single re-mint, replay paging and
cursors, sealed round trips, the outbox's retry/backoff/drop rules."
(:require [clojure.string :as str]
[cljs.test :refer [deftest is testing]]
[cljs.core.async :refer [<!]]
[ird.p2p.crypto :as crypto]
[ird.p2p.errors :as err]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.mailbox :as mbx]
[ird.p2p.test-util :as tu :refer [deftest-async]]))
(def endpoint "https://app.example.test/mailbox/v1/")
(defn token-fn [& [t]] (mbx/static-token-fn {:token (or t "tok") :expires_at 9999999999}))
(defn client [http & [opts]] (mbx/make-client (merge {:endpoint endpoint :token-fn (token-fn) :http-fn http} opts)))
(deftest room-is-one-percent-encoded-segment
(is (= "https://app.example.test/mailbox/v1/rooms/lobby/messages" (mbx/messages-url endpoint "lobby")))
(is (= "https://app.example.test/mailbox/v1/rooms/%2Forbitdb%2Fzdpu123%2Fchat/messages" (mbx/messages-url endpoint "/orbitdb/zdpu123/chat")))
(is (= "https://app.example.test/mailbox/v1/rooms/a%20b%3Ac%3Fd%23e%25f%2Bg/messages" (mbx/messages-url endpoint "a b:c?d#e%f+g")))
(is (= "https://app.example.test/mailbox/v1/rooms/%C4%83%C8%99/messages" (mbx/messages-url endpoint "ăș")))
(is (thrown? :default (mbx/validate-room " ")))
(is (thrown? :default (mbx/validate-room (apply str (repeat 4097 "x"))))))
(deftest seq-orders-beyond-double-precision
(is (mbx/seq< "1873421000000000000001" "1873421000000000000002"))
(is (mbx/seq< "9" "10"))
(is (not (mbx/seq< "99999999999999999999999" "100000000000000000000")))
(is (true? (mbx/history-gap? "400" {:oldest_seq "500"})))
(is (false? (mbx/history-gap? "500" {:oldest_seq "500"})))
(is (false? (mbx/history-gap? nil {:oldest_seq "500"})))
(is (false? (mbx/history-gap? "" {:oldest_seq "500"})))
(is (false? (mbx/history-gap? "5" {})))
(is (true? (mbx/history-gap? "18446744073709551615" {:oldest_seq "18446744073709551616"}))))
(deftest-async deposit-sends-bearer-and-octet-stream
(let [captured (atom [])
c (client (tu/scripted captured [[201 {:seq "1873421" :id "00ff" :expires_at 1787003600 :deduped false}]]) {:origin "https://o.example"})
r (<! (mbx/deposit-raw c "lobby" (bytes/bytes->b64 [1 2 3])))
req (first @captured)]
(is (= {:seq "1873421" :id "00ff" :expires_at 1787003600 :deduped false} r))
(is (= :post (:method req)))
(is (= "https://app.example.test/mailbox/v1/rooms/lobby/messages" (:url req)))
(is (= "Bearer tok" (get-in req [:headers "Authorization"])))
(is (= "application/octet-stream" (get-in req [:headers "Content-Type"])))
(is (= "https://o.example" (get-in req [:headers "Origin"])))
(is (= {:b64 "AQID"} (:body req)))
(let [e (<! (mbx/deposit-raw c "lobby" ""))]
(is (err/config-error? e))
(is (= "mailbox payload is empty" (ex-message e))))
(let [e (<! (mbx/deposit-raw c "" "AQID"))]
(is (err/config-error? e)))))
(deftest-async deduped-deposit-is-200
(let [c (client (tu/responding 200 {:seq 5 :id "aa" :expires_at 1 :deduped true}))
r (<! (mbx/deposit-raw c "r" "AQ=="))]
(is (true? (:deduped r)))
(is (= "5" (:seq r)) "seq is always a string")))
(deftest-async auth-failure-remints-once-then-errors
(let [http (atom []) mint (atom [])
now (atom 1000000000)
tokens (mbx/open-mint-token-fn {:mint-url "https://app.example.test/mailbox-credentials" :origin "https://o.example"
:http-fn (tu/responding mint 200 {:token "t2" :ttl 60 :expires_at 2000000000 :endpoint endpoint})
:now-fn (fn [] (* 1000 @now))})
c (client (tu/responding http 401 {:error "token_expired"}) {:token-fn tokens})
e (<! (mbx/deposit-raw c "r" "AQ=="))]
(is (err/http-error? e))
(is (= 401 (:status (ex-data e))))
(is (= "token_expired" (:code (ex-data e))))
(is (= {:error "token_expired"} (:response (ex-data e))))
(is (= 2 (count @http)) "original + one retry")
(is (= 2 (count @mint)) "initial mint + re-mint")
(is (= "https://app.example.test/mailbox-credentials?ttl=86400" (:url (first @mint))) "app-scoped: no room parameter")
(is (= "https://o.example" (get-in (first @mint) [:headers "Origin"])))))
(deftest-async open-mint-caches-and-renews-five-minutes-early
(let [now (atom 1000000)
mint (atom [])
tokens (mbx/open-mint-token-fn {:mint-url "https://app.example.test/mailbox-credentials"
:http-fn (tu/responding mint 200 {:token "t" :expires_at 1000600})
:now-fn (fn [] (* 1000 @now)) :ttl-sec 600 :scope-to-room? true})]
(<! (tokens {:room "r/1"}))
(<! (tokens {:room "r/1"}))
(is (= 1 (count @mint)))
(is (= "https://app.example.test/mailbox-credentials?room=r%2F1&ttl=600" (:url (last @mint))))
(<! (tokens {:room "other"}))
(is (= 2 (count @mint)) "room-scoped: one per room")
(swap! now + 301)
(<! (tokens {:room "r/1"}))
(is (= 3 (count @mint)) "299 s left < 300 s renew window")))
(deftest-async mint-errors-surface-as-http
(let [tokens (mbx/open-mint-token-fn {:mint-url "https://app.example.test/mailbox-credentials"
:http-fn (tu/responding 404 {:error "not_found"})})
e (<! (tokens {:room "r"}))]
(is (err/http-error? e))
(is (= "not_found" (:code (ex-data e))))))
(deftest-async replay-decodes-and-pages
(let [pages [{:messages [{:seq "10" :id "a" :ts 1712345678901 :payload "AQID"} {:seq "11" :id "b" :ts 1712345678902 :payload "BA=="}]
:has_more true :oldest_seq "10" :latest_seq "12"}
{:messages [{:seq 12 :id "c" :ts 1712345678903 :payload "BQ=="}] :has_more false :oldest_seq "10" :latest_seq "12"}]
captured (atom [])
http (tu/routing captured [["after=11&limit=100" (fn [_] [200 (second pages)])]
["before=12&limit=500" (fn [_] [200 (second pages)])]
["after=5&limit=100" (fn [_] [200 (first pages)])]])
c (client http)
r (<! (mbx/replay-all-raw c "lobby" {:after "5" :limit 100}))]
(is (= ["AQID" "BA==" "BQ=="] (mapv :payload (:messages r))))
(is (= "12" (:cursor r)))
(is (true? (:gap? r)) "cursor 5 < oldest 10")
(is (= 2 (count @captured)))
(let [page (<! (mbx/replay-raw c "lobby" {:before "12" :limit 9999}))]
(is (str/ends-with? (:url (last @captured)) "?before=12&limit=500"))
(is (false? (:has_more page)))
(is (= "12" (:seq (first (:messages page))))))
(is (= :get (:method (first @captured))))
(is (= "Bearer tok" (get-in (first @captured) [:headers "Authorization"])))))
(deftest-async non-success-maps-to-http-error-with-code
(let [c (client (tu/responding 413 {:error "message_too_large"}))
e (<! (mbx/deposit-raw c "r" "AQ=="))]
(is (err/http-error? e))
(is (= 413 (:status (ex-data e))))
(is (= "message_too_large" (:code (ex-data e))))
(is (= "message_too_large" (ex-message e)))))
(deftest-async outbox-retries-backs-off-and-drops
(let [statuses (atom [429 507 201 413 201])
http (fn [_] (let [s (first @statuses)]
(swap! statuses subvec 1)
(cljs.core.async/go {:status s :headers {} :body (tu/->json (if (= 201 s) {:seq "1" :id "a" :expires_at 1} {:error "x"}))})))
slept (atom [])
outbox (mbx/make-outbox {:client (client http) :sleep-fn (fn [ms] (swap! slept conj ms) (tu/no-sleep ms))})]
(mbx/outbox-enqueue outbox "r" "AQ==")
(mbx/outbox-enqueue outbox "r" "Ag==")
(mbx/outbox-enqueue outbox "r" "Aw==")
(is (= 3 (count (mbx/outbox-pending outbox))))
(let [outcomes (<! (mbx/outbox-flush outbox))]
(is (= [:deferred :deferred :deposited :dropped :deposited] (mapv :outcome outcomes)))
(is (= [1000 2000] @slept))
(is (= {:room "r" :payload "AQ==" :attempts 1} (:item (first outcomes))))
(is (= 1000 (:retry-after-ms (first outcomes))))
(is (= 2000 (:retry-after-ms (second outcomes))))
(is (= "message_too_large" (:reason (nth outcomes 3))))
(is (= "Ag==" (get-in (nth outcomes 3) [:item :payload])))
(is (empty? (mbx/outbox-pending outbox)))))
(is (= [1000 2000 4000 300000] (map mbx/retry-delay [0 1 2 20]))))
(deftest-async sealed-mailbox-round-trip
(let [secret (crypto/random-secret)
rc (<! (crypto/room-crypto {:secret secret :app-salt "t" :peer-id (crypto/random-peer-id)}))
mailbox-room (<! (crypto/mailbox-room-id secret "t"))
store (atom [])
http (tu/routing [["/messages" (fn [req]
(if (= :post (:method req))
(do (swap! store conj {:seq (str (inc (count @store))) :id (str "m" (count @store)) :ts 1 :payload (get-in req [:body :b64])})
[201 {:seq (str (count @store)) :id "x" :expires_at 1 :deduped false}])
[200 {:messages @store :has_more false :oldest_seq "1" :latest_seq (str (count @store))}]))]])
m (mbx/make-mailbox {:rc rc :client (client http) :mailbox-room-id mailbox-room :sleep-fn tu/no-sleep})]
(is (not= (:room-id m) (:mailbox-room-id m)))
(is (= "1" (:seq (<! (mbx/deposit m {:kind :json :value {:text "offline"}})))))
(is (= "2" (:seq (<! (mbx/deposit m {:kind :data :payload "AQID"})))))
;; a foreign blob and garbage in the room are dropped silently
(swap! store conj {:seq "3" :id "junk" :ts 1 :payload "AgECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJA=="})
(let [{:keys [items cursor gap?]} (<! (mbx/replay-all m nil))]
(is (= [{:kind :json :value {:text "offline"}} {:kind :data :payload "AQID"}] (mapv :message items)))
(is (= ["1" "2"] (mapv :seq items)))
(is (= "3" cursor))
(is (false? gap?)))
(let [{:keys [items page]} (<! (mbx/replay m {:limit 10}))]
(is (= 2 (count items)))
(is (= 3 (count (:messages page)))))
;; the at-rest payload is the 0x02 envelope
(is (= 2 (bytes/byte-at (:payload (first @store)) 0)))
;; enqueue + flush
(is (true? (<! (mbx/enqueue m {:kind :json :value "queued"}))))
(is (= 1 (count (mbx/pending m))))
(is (= [:deposited] (mapv :outcome (<! (mbx/flush-outbox m)))))
(is (empty? (mbx/pending m)))
(is (err/config-error? (<! (mbx/deposit m {:kind :nope}))))))
|