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 | (ns ird.mailbox
"The p2p offline mailbox: deposit an encrypted payload for a room and let
a peer that shows up later replay it. This is the catch-up path for peers
that are never online at the same time — for js-libp2p/Helia/OrbitDB apps,
where encrypted log entries and head CIDs go, keyed by the database
address.
Unlike the rest of this client these calls do not go to api.ird.ro: they
go to the app's own node at `https://<hostname>/mailbox/v1`, authenticated
with a short-lived bearer token. Mint one with `credentials` (which wraps
ird.p2p/mint-mailbox-credentials) and pass the returned map as the first
argument to `deposit` and `replay`; it carries both the :endpoint and the
:token. The app needs :mailbox_enabled — see ird.p2p/update-app.
The node stores exactly the bytes you send and holds no keys: encrypt
before depositing. Payloads are capped by the app's :mailbox_max_message_kb
(default 64), retained for :mailbox_ttl_hours (default 24) and bounded by
a per-app :mailbox_max_mb ring buffer — so replayed history can be
truncated under you; see `history-gap?`."
(:require
[clojure.string :as str]
[ird.errors :as err]
[ird.p2p :as p2p]
[ird.transport :as transport])
(:import
[java.io File InputStream]
[java.net URLEncoder]
[java.util Base64]))
(defn credentials
"Mint a mailbox token for APP-ID and return the credentials map directly —
{:token :ttl :expires_at :endpoint :room} — ready to pass to `deposit` and
`replay`. CLIENT is the ird API client; OPTS may carry :room (scope the
token to one room) and :ttl_sec (default 3600, capped at 86400).
Tokens verify offline and cannot be revoked short of rotating the node
secret, so mint short and reuse one across calls: every mint is a charged
request."
([client app-id] (credentials client app-id nil))
([client app-id opts]
(:credentials (p2p/mint-mailbox-credentials client app-id opts))))
(defn- encode-room
"Percent-encode ROOM as a single path segment. Room ids are opaque
customer strings and routinely contain slashes (an OrbitDB address is
\"/orbitdb/zdpu…\"), so the whole id is encoded — the node proxies the raw
URI and unescapes it back into one room."
[room]
(when (str/blank? (str room))
(throw (err/config-error "Mailbox room must be a non-empty string"
{:room room})))
(-> (URLEncoder/encode (str room) "UTF-8")
(str/replace "+" "%20")))
(defn- messages-url
[{:keys [endpoint]} room]
(when (str/blank? (str endpoint))
(throw (err/config-error
"Mailbox credentials must carry an :endpoint (mint them with `credentials`)"
{:endpoint endpoint})))
(str (str/replace (str endpoint) #"/*$" "")
"/rooms/" (encode-room room) "/messages"))
(defn- coerce-payload
"Coerce PAYLOAD into request-body bytes. A string is its own UTF-8 bytes —
payloads are data here, never file paths (contrast
ird.transport/upload-bytes)."
[payload]
(cond
(bytes? payload) payload
(string? payload) (.getBytes ^String payload "UTF-8")
(instance? InputStream payload) payload
(instance? File payload) payload
:else (throw (err/transport-error
"Unsupported mailbox payload; expected a byte array, string, InputStream, or File"
{:payload-class (some-> payload class str)}))))
(defn deposit
"POST /mailbox/v1/rooms/{room}/messages — Deposit PAYLOAD into ROOM for
peers that arrive later. CREDS is a credentials map from `credentials`.
PAYLOAD is a byte array, a string (sent as its UTF-8 bytes — not a file
path), an InputStream, or a File. Encrypt it yourself: the node cannot.
Identical payloads deduplicate per room within the retention window, so
retries and the same block from several peers are safe and cost nothing
extra. Returns {:seq :id :expires_at :deduped} — :deduped true means the
payload was already stored. Throws an :ird/api-error whose :code is the
node's, notably \"message_too_large\" (over the app's
:mailbox_max_message_kb) and \"node_storage_full\"."
[creds room payload]
(transport/request-url :post (messages-url creds room)
{:token (:token creds)
:headers {"Content-Type" "application/octet-stream"}
:body (coerce-payload payload)}))
(defn replay
"GET /mailbox/v1/rooms/{room}/messages — Replay ROOM, oldest first. CREDS
is a credentials map from `credentials`.
OPTS:
:after replay forward from this :seq, exclusive (the usual cursor)
:before scroll back from this :seq, exclusive
:limit page size, default 100, capped at 500
Returns {:messages [{:seq :id :ts :payload}] :has_more :oldest_seq
:latest_seq}. Each :payload is standard base64 — `payload-bytes` decodes
one. Every :seq is a string and must stay one: it packs a millisecond
clock into its high bits and is far past what a JSON number survives.
Compare seqs as opaque ordered tokens.
There is no server-side ack: the cursor is yours to keep. Check it against
:oldest_seq with `history-gap?` on every replay."
([creds room] (replay creds room nil))
([creds room opts]
(transport/request-url :get (messages-url creds room)
{:token (:token creds)
:query (select-keys opts [:after :before :limit])})))
(defn payload-bytes
"Decode one replayed MESSAGE's base64 :payload into a byte array."
^bytes [message]
(.decode (Base64/getDecoder) ^String (:payload message)))
(defn payload-string
"Decode one replayed MESSAGE's :payload and read it as UTF-8 text. Only
meaningful for payloads that were deposited as text — mailbox contents are
usually ciphertext."
[message]
(String. (payload-bytes message) "UTF-8"))
(defn- seq-compare
"Compare two :seq strings as the unbounded integers they encode. Never
exposed: callers treat seqs as opaque."
[a b]
(.compareTo (BigInteger. (str a)) (BigInteger. (str b))))
(defn history-gap?
"True when CURSOR — the :seq you last saw — is older than PAGE's
:oldest_seq, meaning messages were evicted (TTL or the app's ring buffer)
between your visits. Report that as truncated history, not as a quiet
room. A nil/blank cursor (a first visit) is never a gap."
[cursor page]
(boolean
(when-not (str/blank? (str cursor))
(when-let [oldest (:oldest_seq page)]
(neg? (seq-compare cursor oldest))))))
(defn replay-all
"Replay ROOM forward from :after until :has_more is exhausted, collecting
every message. Returns {:messages [...] :cursor :gap?} — :cursor is the
:seq to resume from next time (the last one seen, or the :after you passed
when the room had nothing new), and :gap? reports `history-gap?` on the
first page.
OPTS: :after (the cursor to resume from) and :limit (page size).
Suits a reconnecting peer catching up on a bounded cache. Every page is a
charged request, so pass the largest :limit you can use."
([creds room] (replay-all creds room nil))
([creds room {:keys [after limit]}]
(loop [cursor after
acc []
gap? nil]
(let [page (replay creds room (cond-> {:after cursor} limit (assoc :limit limit)))
msgs (:messages page)
gap? (if (nil? gap?) (history-gap? after page) gap?)
acc' (into acc msgs)
cursor' (or (:seq (last msgs)) cursor)]
(if (and (:has_more page) (seq msgs))
(recur cursor' acc' gap?)
{:messages acc'
:cursor (or (:seq (last acc')) after)
:gap? (boolean gap?)})))))
|