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 | (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?`.
`deposit`, `replay` and `replay-all` have no client, so they take an
optional trailing OPTS map with :http-fn and :timeout-ms (see
ird.transport/request-url). Every network function returns a channel."
(:require [clojure.string :as str]
[ird.async :refer [go-try <?]]
[ird.errors :as err]
[ird.node :as node]
[ird.p2p :as p2p]
[ird.transport :as transport]))
(defn credentials
"Mint a mailbox token for APP-ID and deliver 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]
(go-try (: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})))
(transport/encode (str room)))
(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
(string? payload) (node/utf8-bytes payload)
(node/bytes? payload) payload
:else (throw (err/transport-error
"Unsupported mailbox payload; expected a string, Uint8Array, ArrayBuffer, Blob or ReadableStream"
{:payload-type (goog/typeOf payload)}))))
(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 string (sent as its UTF-8 bytes — not a file path), a
Uint8Array/Buffer, ArrayBuffer, Blob or ReadableStream. 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. Delivers {:seq :id :expires_at :deduped} — :deduped true means the
payload was already stored. An :ird/api-error's :code is the node's,
notably \"message_too_large\" (over the app's :mailbox_max_message_kb) and
\"node_storage_full\"."
([creds room payload] (deposit creds room payload nil))
([creds room payload {:keys [http-fn timeout-ms]}]
(go-try
(<? (transport/request-url :post (messages-url creds room)
{:token (:token creds)
:headers {"Content-Type" "application/octet-stream"}
:body (coerce-payload payload)
:http-fn http-fn
:timeout-ms timeout-ms})))))
(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
:http-fn / :timeout-ms transport options
Delivers {: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 {:keys [http-fn timeout-ms] :as opts}]
(go-try
(<? (transport/request-url :get (messages-url creds room)
{:token (:token creds)
:query (select-keys opts [:after :before :limit])
:http-fn http-fn
:timeout-ms timeout-ms})))))
(defn payload-bytes
"Decode one replayed MESSAGE's base64 :payload into a Uint8Array."
[message]
(node/base64-decode (: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]
(node/utf8-string (payload-bytes message)))
(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. Seqs are
compared as the arbitrary-precision integers they encode (BigInt), never
as strings or doubles."
[cursor page]
(boolean
(when-not (str/blank? (str cursor))
(when-let [oldest (:oldest_seq page)]
(< (js/BigInt (str cursor)) (js/BigInt (str oldest)))))))
(defn replay-all
"Replay ROOM forward from :after until :has_more is exhausted, collecting
every message. Delivers {: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), :limit (page size), :http-fn,
:timeout-ms.
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 http-fn timeout-ms]}]
(go-try
(loop [cursor after
acc []
gap? nil]
(let [page (<? (replay creds room (cond-> {:after cursor :http-fn http-fn :timeout-ms timeout-ms}
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?)}))))))
|